當(dāng)前位置：首頁(yè) > 编程资源 > 综合教程 >内容正文

综合教程

Python实现WebSocket

發(fā)布時(shí)間：2023/12/19 综合教程 25 生活家

生活随笔收集整理的這篇文章主要介紹了 Python实现WebSocket 小編覺(jué)得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.

Python實(shí)現(xiàn)WebSocket

一、WebSocket建立連接

1. 握手環(huán)節(jié)

目的:驗(yàn)證服務(wù)端是否支持Websocket協(xié)議

流程：

客戶端瀏覽器第一次訪問(wèn)服務(wù)器的時(shí)候，瀏覽器內(nèi)部會(huì)自動(dòng)生成一個(gè)隨機(jī)字符串，將該隨機(jī)字符串發(fā)送給服務(wù)端（基于http）協(xié)議）瀏覽器也保留隨機(jī)生成的字符串（在請(qǐng)求頭里面）
服務(wù)端接收隨機(jī)字符串之后，會(huì)將字符串與magic string(全球統(tǒng)一)做字符串拼接，然后利用加密算法對(duì)拼接好的字符串進(jìn)行加密處理（sha1/base64）,此時(shí)客戶端也對(duì)產(chǎn)生的隨機(jī)字符串做上述的拼接和加密操作
接著服務(wù)器將產(chǎn)生好的隨機(jī)字符串發(fā)送給客戶端的瀏覽器（響應(yīng)頭里面），客戶端瀏覽器會(huì)對(duì)比服務(wù)器發(fā)送的隨機(jī)字符串與瀏覽器本地操作的隨機(jī)字符串進(jìn)行對(duì)比，如果一致說(shuō)明該服務(wù)端支持websocket，如果不一致服務(wù)端則不支持。

請(qǐng)求協(xié)議

GET / HTTP/1.1\r\n  # 請(qǐng)求首行，握手階段還是使用http協(xié)議
Host: 127.0.0.1:8080\r\n # 請(qǐng)求頭
Connection: Upgrade\r\n  # 表示要升級(jí)協(xié)議
Pragma: no-cache\r\n
Cache-Control: no-cache\r\n
User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.72 Safari/537.36\r\n
Upgrade: websocket\r\n  # 要升級(jí)協(xié)議到websocket協(xié)議
Origin: http://localhost:63342\r\n
Sec-WebSocket-Version: 13\r\n  # 表示websocket的版本。如果服務(wù)端不支持該版本，需要返回一個(gè)Sec-WebSocket-Versionheader，里面包含服務(wù)端支持的版本號(hào)
Accept-Encoding: gzip, deflate, br\r\n
Accept-Language: zh-CN,zh;q=0.9,en;q=0.8\r\n
Sec-WebSocket-Key: 07EWNDBSpegw1vfsIBJtkg==\r\n # 對(duì)應(yīng)服務(wù)端響應(yīng)頭的Sec-WebSocket-Accept，由于沒(méi)有同源限制，websocket客戶端可任意連接支持websocket的服務(wù)。這個(gè)就相當(dāng)于一個(gè)鑰匙一把鎖，避免多余的，無(wú)意義的連接
Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bits\r\n\r\n

Sec-WebSocket-Key 是一個(gè) Base64 encode 的值，這個(gè)是瀏覽器隨機(jī)生成的，發(fā)送給服務(wù)器
服務(wù)端從請(qǐng)求(HTTP的請(qǐng)求頭)信息中提取 Sec-WebSocket-Key，利用magic_string 和 Sec-WebSocket-Key 先進(jìn)行拼接，然后采用hmac1加密，再進(jìn)行base64加密
將加密結(jié)果響應(yīng)給客戶端，服務(wù)器會(huì)返回下列東西，表示已經(jīng)接受到請(qǐng)求，成功建立 WebSocket

響應(yīng)協(xié)議

HTTP/1.1 101 Switching Protocols\r\n  # 響應(yīng)首行，還是使用http協(xié)議
Upgrade:websocket\r\n                 # 表示要升級(jí)到websocket協(xié)議
Connection: Upgrade\r\n               # 表示要升級(jí)協(xié)議
Sec-WebSocket-Accept: 07EWNDBSpegw1vfsIBJtkg==\r\n # 根據(jù)客戶端請(qǐng)求首部的Sec-WebSocket-Key計(jì)算出來(lái)。
WebSocket-Location: ws://127.0.0.1:8000\r\n\r\n

2. 收發(fā)數(shù)據(jù)（send/onmessage）

驗(yàn)證成功之后就可以數(shù)據(jù)交互了但是交互的數(shù)據(jù)是加密的需要解密處理

數(shù)據(jù)基于網(wǎng)絡(luò)傳輸都是二進(jìn)制格式，單位換算 8bit = 1bytes
步驟一：讀取第二個(gè)字節(jié)的后七位稱(chēng)之為payload，根據(jù)payload大小決定不同的處理方式：

=127 再讀取8個(gè)字節(jié) 作為數(shù)據(jù)報(bào)
=126 再讀取2個(gè)字節(jié) 作為數(shù)據(jù)報(bào)
<=125 不再往后讀了

步驟二：

# 步驟1之后 會(huì)對(duì)剩下的數(shù)據(jù)再讀取4個(gè)字節(jié)(masking-key)		
# 之后依據(jù)masking-key算出真實(shí)數(shù)據(jù)
var DECODED = "";
for(var i = 0; i < ENCODED.length; i++) {
       DECODED[i] = ENCODED[i] ^ MASK[i % 4];
  }

二、Python實(shí)現(xiàn)

Python后端：

from django.test import TestCase

# Create your tests here.
import socket
import base64
import hashlib

# 正常的socket代碼
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 防止linux/mac報(bào)錯(cuò)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind(('127.0.0.1', 8080))
sock.listen(5)

conn, address = sock.accept()
data = conn.recv(1024)  # 獲取客戶端發(fā)送的消息


def get_headers(data):
    """
    將請(qǐng)求頭格式化成字典
    :param data:
    :return:
    """
    
    """
    請(qǐng)求頭格式：
    GET / HTTP/1.1\r\n  # 請(qǐng)求首行，握手階段還是使用http協(xié)議
    Host: 127.0.0.1:8080\r\n # 請(qǐng)求頭
    Connection: Upgrade\r\n  # 表示要升級(jí)協(xié)議
    Pragma: no-cache\r\n
    Cache-Control: no-cache\r\n
    User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.72 Safari/537.36\r\n
    Upgrade: websocket\r\n  # 要升級(jí)協(xié)議到websocket協(xié)議
    Origin: http://localhost:63342\r\n
    Sec-WebSocket-Version: 13\r\n  # 表示websocket的版本。如果服務(wù)端不支持該版本，需要返回一個(gè)Sec-WebSocket-Versionheader，里面包含服務(wù)端支持的版本號(hào)
    Accept-Encoding: gzip, deflate, br\r\n
    Accept-Language: zh-CN,zh;q=0.9,en;q=0.8\r\n
    Sec-WebSocket-Key: 07EWNDBSpegw1vfsIBJtkg==\r\n # 對(duì)應(yīng)服務(wù)端響應(yīng)頭的Sec-WebSocket-Accept，由于沒(méi)有同源限制，websocket客戶端可任意連接支持websocket的服務(wù)。這個(gè)就相當(dāng)于一個(gè)鑰匙一把鎖，避免多余的，無(wú)意義的連接
    Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bits\r\n\r\n
    """
    header_dict = {}
    data = str(data, encoding='utf-8')

    header, body = data.split('\r\n\r\n', 1) # 因?yàn)檎?qǐng)求頭信息結(jié)尾都是\r\n,并且最后末尾部分是\r\n\r\n；
    header_list = header.split('\r\n')
    for i in range(0, len(header_list)):
        if i == 0:
            if len(header_list[i].split(' ')) == 3:
                header_dict['method'], header_dict['url'], header_dict['protocol'] = header_list[i].split(' ')
        else:
            k, v = header_list[i].split(':', 1)
            header_dict[k] = v.strip()
    return header_dict


# 想將http協(xié)議的數(shù)據(jù)處理成字典的形式方便后續(xù)取值
header_dict = get_headers(data)  # 將一大堆請(qǐng)求頭轉(zhuǎn)換成字典數(shù)據(jù)  類(lèi)似于wsgiref模塊
client_random_string = header_dict['Sec-WebSocket-Key']  # 獲取瀏覽器發(fā)送過(guò)來(lái)的隨機(jī)字符串

# magic string拼接
magic_string = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'  # 全球共用的隨機(jī)字符串 一個(gè)都不能寫(xiě)錯(cuò)
# 確認(rèn)握手Sec-WebSocket-Key固定格式：headers頭部的Sec-WebSocket-Key+'258EAFA5-E914-47DA-95CA-C5AB0DC85B11'
# 確認(rèn)握手的秘鑰值為 傳入的秘鑰+magic_string，使用sha1算法加密，然后base64轉(zhuǎn)碼
value = client_random_string + magic_string  # 拼接

# 算法加密 對(duì)請(qǐng)求頭中的sec-websocket-key進(jìn)行加密
ac = base64.b64encode(hashlib.sha1(value.encode('utf-8')).digest())  # 加密處理

# 將處理好的結(jié)果再發(fā)送給客戶端校驗(yàn)
tpl = "HTTP/1.1 101 Switching Protocols\r\n" \
      "Upgrade:websocket\r\n" \
      "Connection: Upgrade\r\n" \
      "Sec-WebSocket-Accept: %s\r\n" \
      "WebSocket-Location: ws://127.0.0.1:8080\r\n\r\n"
response_str = tpl % ac.decode('utf-8')  # 處理到響應(yīng)頭中

# 將隨機(jī)字符串給瀏覽器返回回去
print(f"建立連接,加密驗(yàn)證key{ac}")
conn.send(bytes(response_str, encoding='utf-8'))


def get_data(info):
    """
    前后端進(jìn)行通信,對(duì)前端發(fā)生消息進(jìn)行解密
    對(duì)返回消息進(jìn)行解碼比較復(fù)雜，詳見(jiàn)數(shù)據(jù)幀格式解析
    """
    payload_len = info[1] & 127
    if payload_len == 126:
        extend_payload_len = info[2:4]
        mask = info[4:8]
        decoded = info[8:]
    elif payload_len == 127:
        extend_payload_len = info[2:10]
        mask = info[10:14]
        decoded = info[14:]
    else:
        extend_payload_len = None
        mask = info[2:6]
        decoded = info[6:]
    bytes_list = bytearray() # 使用字節(jié)將數(shù)據(jù)全部收集，再去字符串編碼，這樣不會(huì)導(dǎo)致中文亂碼
    for i in range(len(decoded)):
        chunk = decoded[i] ^ mask[i % 4]  # 異或運(yùn)算
        bytes_list.append(chunk)

    body = str(bytes_list, encoding='utf-8')
    return body


# 基于websocket通信
while True:
    # ws.send("info")
    data = conn.recv(1024)  # 數(shù)據(jù)是加密處理的
    # print(data)
    # 對(duì)data進(jìn)行解密操作
    value = get_data(data)
    print(value)

前端:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
    <meta name="viewport" content="width=device-width, initial-scale=1">
    <link  rel="stylesheet">
    <script src="https://cdn.bootcss.com/jquery/3.3.1/jquery.min.js"></script>
    <script src="https://cdn.bootcss.com/twitter-bootstrap/3.4.1/js/bootstrap.min.js"></script>
</head>
<body>

<script>
    var ws = new WebSocket('ws://127.0.0.1:8080/')
    // 上面這句代碼干了很多事
    // 1 產(chǎn)生隨機(jī)字符串并發(fā)送給服務(wù)端  Sec-WebSocket-Key: EMy5N4dwjl/jHoU0eYDDGQ==
    // 2 服務(wù)端發(fā)送處理好的內(nèi)容過(guò)來(lái)之后 自動(dòng)校驗(yàn)
</script>
</body>
</html>

前端后端發(fā)送消息

三、總結(jié)

客戶端第一次訪問(wèn)服務(wù)端，客戶端會(huì)隨機(jī)生成一個(gè)字符串發(fā)送給服務(wù)端Sec-WebSocket-Key，瀏覽器也會(huì)保存一份，然后服務(wù)端解析獲取隨機(jī)字符串與magic string進(jìn)行拼接，然后通過(guò)base64和sha1進(jìn)行加密，返回給瀏覽器，瀏覽器保留的字符串也會(huì)經(jīng)過(guò)上面操作之后與后端生成的加密字符串進(jìn)行比較，如果相同則說(shuō)明服務(wù)端支持websocket，如果不一致則不支持（握手環(huán)節(jié)）。驗(yàn)證成功之后，就可以數(shù)據(jù)交互，交互的數(shù)據(jù)是加密的，所以需要解密，解密：首先讀取第二個(gè)字節(jié)后七位，后七位也稱(chēng)之為payload,根據(jù)payload的大小進(jìn)行不同的操作，如果 =127 再讀取后面8個(gè)字節(jié)作為數(shù)據(jù)報(bào)，如果=126在讀取后面2個(gè)字節(jié)作為數(shù)據(jù)報(bào)，如果<125 則不再往后讀取數(shù)據(jù)，最后會(huì)對(duì)剩下的數(shù)據(jù)再次讀取4個(gè)字節(jié)之后，依據(jù)masking-key 算出(異或)真實(shí)的數(shù)據(jù)結(jié)果。

在當(dāng)下的階段，必將由程序員來(lái)主導(dǎo)，甚至比以往更甚。

總結(jié)

以上是生活随笔為你收集整理的Python实现WebSocket的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。

上一篇： IxChariot：网络吞吐量及延迟)测
下一篇： 14年来最差季度业绩！三星2023年第一