如何创建安全的Python WebSocket客户端请求？

Question

我的Python安全WebSocket客户端代码会给我抛出以下异常：

[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed (_ssl.c:748)

我已经创建了自己的私有证书和签名证书，但是使用以下Python脚本无法连接：

import json
from websocket import create_connection


class subscriber:
   def listenForever(self):
   try:
      # ws = create_connection("wss://localhost:9080/websocket")
      ws = create_connection("wss://nbtstaging.westeurope.cloudapp.azure.com:9090/websocket")
      ws.send("test message")
      while True:
          result = ws.recv()
          result = json.loads(result)
          print("Received '%s'" % result)

      ws.close()
  except Exception as ex:
      print("exception: ", format(ex))


try:
    subscriber().listenForever()
except:
    print("Exception occured: ")

我的Python https/wss服务器脚本如下，使用了Tornado框架:

import tornado.web
import tornado.websocket
import tornado.httpserver
import tornado.ioloop
import os
import ssl

ssl_root = os.path.join(os.path.dirname(__file__), 'ssl1_1020')


class WebSocketHandler(tornado.websocket.WebSocketHandler):
    def check_origin(self, origin):
        return True

    def open(self):
        pass

    def on_message(self, message):
        self.write_message("Your message was: " + message)
        print("message received: ", format(message))

    def on_close(self):
        pass


class IndexPageHandler(tornado.web.RequestHandler):
    def get(self):
        self.render("index.html")


class Application(tornado.web.Application):
    def __init__(self):
        handlers = [
           (r'/', IndexPageHandler),
           (r'/websocket', WebSocketHandler),
        ]

        settings = {
            'template_path': 'templates'
        }
        tornado.web.Application.__init__(self, handlers, **settings)


ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
ssl_ctx.load_cert_chain(ssl_root+"/server.crt",
                    ssl_root + "/server.pem")


if __name__ == '__main__':
    ws_app = Application()
    server = tornado.httpserver.HTTPServer(ws_app, ssl_options=ssl_ctx,)
    server.listen(9081, "0.0.0.0")
    print("server started...")
    tornado.ioloop.IOLoop.instance().start()

创建 SSL 签名证书的步骤：

openssl genrsa -des3 -out server.key 1024
openssl rsa -in server.key -out server.pem
openssl req -new -nodes -key server.pem -out server.csr
openssl x509 -req -days 365 -in server.csr -signkey server.pem -out server.crt

- Ravi Anand

---

你使用MacOS吗？这很可能是系统权限的问题，你可能需要将证书添加到钥匙串中...你可以查看这篇文章https://dev59.com/4VgR5IYBdhLWcg3wH6Zs。 - securecurve

不，全部都是Linux Ubuntu 16.04环境操作系统。 - Ravi Anand

{btsdaf} - securecurve

Answer 1

最终我找到了解决方案，我在连接到安全的WebSocket URL时，更新了Python客户端脚本以忽略证书请求，如下所示：

最终我找到了解决方案，我在连接到安全的WebSocket URL时，更新了Python客户端脚本以忽略证书请求，如下所示：

 import ssl
 import websocket

 ws = websocket.WebSocket(sslopt={"cert_reqs": ssl.CERT_NONE})
 ws.connect("wss://xxx.com:9090/websocket")

- Ravi Anand

Answer 2

对我来说，忽略错误不是一个选择，由于复杂的物联网环境中存在SSL pinning，我必须使用自签名证书：

For me, ignoring the errors is not an options, I had to use my self signed certificate because of SSL pinning in a complex IoT environment:

import asyncio
import pathlib
import ssl
import websockets

ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
localhost_pem = pathlib.Path(__file__).with_name("localhost.pem")
ssl_context.load_verify_locations(localhost_pem)

async def hello():
    uri = "wss://localhost:8765"
    async with websockets.connect(
        uri, ssl=ssl_context
    ) as websocket:
        name = input("What's your name? ")

        await websocket.send(name)
        print(f"> {name}")

        greeting = await websocket.recv()
        print(f"< {greeting}")

asyncio.get_event_loop().run_until_complete(hello())

在websocket存储库的示例文件夹中找到了它。

P.S

我将其从SSLContext(ssl.PROTOCOL_TLS_CLIENT)更改为SSLContext(ssl.PROTOCOL_TLSv1_2)使其正常工作。

- Or Duan

---

终于，在尝试修复错误三天后，这对我起作用了。非常感谢！ - Nagarjun

Answer 3

如果将来有人对wss Python服务器为何失败感到好奇，原因在于Tornado文档中的以下内容：

当使用安全websocket连接（wss://）和自签名证书时，浏览器的连接可能会失败，因为它想显示“接受此证书”对话框，但没有任何地方可以显示。您必须先访问一个使用相同证书的常规HTML页面以接受它，然后websocket连接才能成功。

- Master Splinter

---

例如，如果WebSocket在localhost:4000上，您需要访问https://localhost:4000并接受风险。 - Rojo

Answer 4

仅限于测试目的，请勿在生产环境中使用以下内容。以下是一种极不安全的应急措施：

Original Answer翻译成"最初的回答"

import asyncio, ssl, websockets

#todo kluge
#HIGHLY INSECURE
ssl_context = ssl.SSLContext()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE
#HIGHLY INSECURE
#todo kluge

uri = "wss://myAwesomeSSL.wss.kluge"

async with websockets.connect(uri, ssl=ssl_context) as websocket:
        greeting = await websocket.recv()
        print(f"< {greeting}")

- JDOaktown

Answer 5

似乎大多数答案都建议禁用TLS验证，这并不理想。

以下是一个工作示例，启用了TLS验证，我认为这应该是默认情况下的做法，除非你在本地主机或类似环境中工作。

import asyncio
import certifi
import ssl
import websockets


async def wss_connect(url):

    # Define an SSL context that will connect to TLS 1.2+ servers
    ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2
    ssl_context.maximum_version = ssl.TLSVersion.TLSv1_3

    # Load the certificates authorities via the certifi library 
    ssl_context.load_verify_locations(certifi.where())

    async with websockets.connect(url, ssl=ssl_context) as websocket:
        await websocket.send("hello!")
        response = await websocket.recv()
        print(f"< {response}")
        return response


uri = "wss://yourhost.wss.example/api/v1/test"
asyncio.get_event_loop().run_until_complete(wss_connect(uri))

- Cosimo