Python Websockets 创建 pem 文件

Question

您好，我正在使用 python 库 Websockets .在开发中一切正常，但在服务器上它崩溃了，因为它需要使用 WSS。上面的链接给出了如何执行此操作的示例:

#!/usr/bin/env python

# WSS (WS over TLS) server example, with a self-signed certificate

import asyncio
import pathlib
import ssl
import websockets

async def hello(websocket, path):
    name = await websocket.recv()
    print(f"< {name}")

    greeting = f"Hello {name}!"

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

ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
localhost_pem = pathlib.Path(__file__).with_name("localhost.pem")
ssl_context.load_cert_chain(localhost_pem)

start_server = websockets.serve(
    hello, "localhost", 8765, ssl=ssl_context
)

asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

这段代码非常简单，但我完全不知道如何生成它想要的文件(服务器和客户端)。我研究过“创建 pem 文件”但无济于事，并且收到了各种 ssl 错误。有人可以解释一下如何为这个应用程序创建 pem 文件吗？谢谢

编辑:根据我使用的答案

sudo openssl req -newkey rsa:2048 -new -nodes -x509 -days 3650 -keyout key.pem -out cert.pem

这创建了两个文件。

我的服务器现在通过执行以下操作成功监听:

ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
path_cert = pathlib.Path(__file__).with_name("cert.pem")
path_key = pathlib.Path(__file__).with_name("key.pem")
ssl_context.load_cert_chain(path_cert, keyfile=path_key)

print("Listening for connection...")
start_server = websockets.serve(handler, HOSTNAME, PORT, ssl=ssl_context)

唯一剩下的问题是让客户端连接，我尝试:

ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
path_cert = pathlib.Path(__file__).with_name("cert.pem")
ssl_context.load_cert_chain(path_cert)

async with websockets.connect(uri, ssl=ssl_context) as websocket:

但我收到错误:ssl.SSLError: [SSL] PEM lib (_ssl.c:3854)

我也试过:

ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
path_cert = pathlib.Path(__file__).with_name("cert.pem")
path_key = pathlib.Path(__file__).with_name("key.pem")
ssl_context.load_cert_chain(path_cert, keyfile=path_key)

async with websockets.connect(uri, ssl=ssl_context) as websocket:

并得到错误 ssl.SSLCertVerificationError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: self signed certificate (_ssl.c:1076)

编辑2: 根据我为客户试过的答案:

ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ssl_context.load_verify_locations()

async with websockets.connect(uri, ssl=ssl_context) as websocket:

这会产生一个新错误:TypeError: cafile, capath and cadata cannot be all omitted

尝试第二个建议:

ssl_context = ssl.create_default_context()
ssl_context.load_verify_locations(certifi.where())

async with websockets.connect(uri, ssl=ssl_context) as websocket:

生成错误:ssl.SSLCertVerificationError: [SSL: CERTIFICATE_VERIFY_FAILED] 证书验证失败:自签名证书 (_ssl.c:1076)

编辑3:最终工作客户端:

ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
path_cert = pathlib.Path(__file__).with_name("cert.pem")
ssl_context.load_verify_locations(path_cert)

async with websockets.connect(uri, ssl=ssl_context) as websocket:

Answer 1

PEM 文件包含有关公钥或|和私钥或证书的内容，并且它具有 base64 编码的数据位。 PEM 表示用于邮件安全标准的 Privacy-Enhanced Mail。它包括表格的页眉和页脚行

Create PEM

如何创建自签名 PEM 文件

openssl req -newkey rsa:2048 -new -nodes -x509 -days 3650 -keyout key.pem -out cert.pem

如何从形成链的现有证书文件创建 PEM 文件(可选)按照下面列出的步骤从私钥中删除密码:

openssl rsa -in server.key -out nopassword.key

How to create a PEM file with the help of an automated script:

1. Download NetIQ Cool Tool OpenSSL-Toolkit.

2. Select Create Certificates | PEM with key and entire trust chain

3. Provide the full path to the directory containing the certificate files.

4. Provide the filenames of the following: private key public key (server crt) (conditional) password for private key (conditional) any intermediate certificate chain file(s)

你会得到这样的东西:

-----BEGIN RSA PRIVATE KEY----- 
(Private Key: domain_name.key contents) 
-----END RSA PRIVATE KEY-----
-----BEGIN CERTIFICATE----- 
(Primary SSL certificate: domain_name.crt contents) 
-----END CERTIFICATE----- 
-----BEGIN CERTIFICATE----- 
(Intermediate certificate: certChainCA.crt contents) 
-----END CERTIFICATE----

你可以解码它:

openssl x509 -in cert.pem -text -noout

你会得到这样的东西:

Certificate:
    Data:
        Version: 3 (0x2)
        Serial Number: 0 (0x0)
    Signature Algorithm: ecdsa-with-SHA256
        Issuer: C = BE, O = GnuTLS, OU = GnuTLS certificate authority, ST = Leuven, CN = GnuTLS certificate authority
        Validity
            Not Before: May 23 20:38:21 2011 GMT
            Not After : Dec 22 07:41:51 2012 GMT
        Subject: C = BE, O = GnuTLS, OU = GnuTLS certificate authority, ST = Leuven, CN = GnuTLS certificate authority
        Subject Public Key Info:
            Public Key Algorithm: id-ecPublicKey
                Public-Key: (256 bit)
                pub:
                    04:52:d8:8d:23:8a:e3:67:d7:86:36:b1:20:0b:09:
                    7d:c8:c9:ba:a2:20:95:2f:c5:4a:63:fa:83:5f:ce:
                    78:2f:8f:f3:62:ca:fd:b7:f7:80:56:9d:6e:17:b9:
                    0e:11:4c:48:b2:c0:af:3b:59:17:16:30:68:09:07:
                    99:17:fe:dd:a7
                ASN1 OID: prime256v1
                NIST CURVE: P-256
        X509v3 extensions:
            X509v3 Basic Constraints: critical
                CA:TRUE
            X509v3 Key Usage: critical
                Certificate Sign, CRL Sign
            X509v3 Subject Key Identifier: 
                F0:B4:81:FE:98:12:BF:B5:28:B9:64:40:03:CB:CC:1F:66:4E:28:03
    Signature Algorithm: ecdsa-with-SHA256
         30:45:02:20:31:ae:c0:3d:4a:3f:21:be:85:17:fc:f0:c7:b2:
         31:07:2a:38:56:43:d1:36:d5:95:e1:7e:52:c0:06:43:87:a7:
         02:21:00:97:8c:0e:b8:3c:0a:41:af:ae:a5:cf:06:7e:d5:c4:
         d8:2f:ff:e2:62:80:34:10:ba:22:dd:35:81:46:93:22:9a

Create PEM File

对于您写的客户部分:

ssl_context.load_cert_chain(path_cert, keyfile=path_key)

替换它:

import json
import asyncio
import websockets
import ssl
import certifi


ssl_context = ssl.create_default_context()
ssl_context.load_verify_locations(certifi.where())


query =  {
    "jsonrpc": "2.0",
    "method": "queryHeadsets",
    "params": {},
    "id": 1
    }
json = json.dumps(query)

async def query(json):

    async with websockets.connect("wss://yourserver.com:54321", ssl=ssl_context) as ws:
        await ws.send(json)
        response = await ws.recv()
        print(response)

asyncio.get_event_loop().run_until_complete(query(json))