python3發送需要雙向認證的wss請求

python3發送需要雙向認證的wss請求

websocket鏈接python有很多封裝好的庫：websocket-client、websockets、aiowebsocket 這里用的websokets

此次接口要求：1、需要雙向認證 2、wss協議 3、發送數據和接受數據都需要序列化和反序列化（probuff）

# encoding = utf-8
import asyncio
import pathlib
import ssl
import websockets
import base64
from pb import conn_ready_pb2
import uuid
import datetime,time
import json
from google.protobuf import json_format
# 服務端聲明CLIENT_AUTH
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
#Purpose.CLIENT_AUTH該值表示該上下文可用於對Web客戶端進行身份驗證（因此，它將用於創建服務器端套接字），carfile服務端證書
ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH,cafile="D://my_project//Ajcloud//ca.crt")
# certfile和keyfile參數為指定包含證書被用來識別連接的本地端可選的文件，客戶端證書
ssl_context.load_cert_chain(certfile="D://my_project//Ajcloud//mgr.crt", keyfile="D://my_project//Ajcloud//mgr.key")
ssl_context.verify_mode = ssl.CERT_REQUIRED

async def hello():
    uri = "wss://xxxxxxxxxx"
    async with websockets.connect(
            uri, ssl=ssl_context
    ) as websocket:
        message = "xxxxx"
        reg = conn_ready_pb2.ConnReadyRequest()
        res = conn_ready_pb2.ConnReadyResponse()
        reg.header.message_id = str(uuid.uuid1())
        datetime_now = datetime.datetime.now()
        date_stamp = str(int(time.mktime(datetime_now.timetuple())))
        data_microsecond = str("%06d" % datetime_now.microsecond)[0:3]
        date_stamp = date_stamp + data_microsecond
        reg.header.message_ts = date_stamp
        r = reg.SerializeToString()#序列化
        b = base64.b64encode(r)  # 將二進制轉為base64編碼
        b2 = str(b, 'utf-8')
        message = {"xxx": "xxx", "yyyy": b2}
        await websocket.send(json.dumps(message))
        print(f"> {message}")
        greeting = await websocket.recv()
        payload = json.loads(greeting)["payload"]
        # base64轉碼成二進制
        payload =base64.b64decode(payload)
        res.ParseFromString(payload)#反序列化
        json_result = json_format.MessageToJson(res)  # pb轉成json格式
        print(json_result)
        print(f"< {greeting}")


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