自己写一个websocket

import socket, hashlib, base64sock = socket.socket()sock.bind((‘127.0.0.1‘, 9000))sock.listen(5)‘‘‘b‘GET / HTTP/1.1\r\nHost: 127.0.0.1:9000\r\nConnection: Upgrade\r\nPragma: no-cache\r\nCache-Control: no-cache\r\nUser-Agent: Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/69.0.3497.81 Safari/537.36\r\nUpgrade: websocket\r\n --建立--Origin: http://localhost:63342\r\nSec-WebSocket-Version: 13\r\nAccept-Encoding: gzip, deflate, br\r\nAccept-Language: zh-CN,zh;q=0.9\r\nSec-WebSocket-Key: vNtQ8bVSNRPL5p+xo1X5CA==\r\n 验证 一按钥匙Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bits\r\n\r\n‘‘‘‘‘‘b‘GET / HTTP/1.1\r\n -- 单纯http地址栏访问Host: 127.0.0.1:9000\r\nConnection: keep-alive\r\nPragma: no-cache\r\nCache-Control: no-cache\r\nUpgrade-Insecure-Requests: 1\r\nUser-Agent: Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/69.0.3497.81 Safari/537.36\r\nAccept: text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8\r\nAccept-Encoding: gzip, deflate, br\r\nAccept-Language: zh-CN,zh;q=0.9\r\n\r\n‘‘‘‘conn, addr = sock.accept()data = conn.recv(1024)# print(data)def get_header(data): # 获取头部信息 header_dict = {} header_str = data.decode(‘utf8‘) for i in header_str.split(‘\r\n‘): if str(i).startswith(‘Sec-WebSocket-Key‘): header_dict[‘Sec-WebSocket-Key‘] = i.split(‘:‘)[1].strip() return header_dictheaders = get_header(data)magic_string = ‘258EAFA5-E914-47DA-95CA-C5AB0DC85B11‘ # magic_string = 魔法字符串value = headers[‘Sec-WebSocket-Key‘] + magic_stringprint(value) #mzrsVOiYogmzB8fYcyMBpw==258EAFA5-E914-47DA-95CA-C5AB0DC85B11ac = base64.b64encode(hashlib.sha1(value.encode(‘utf-8‘)).digest()) # 固定加密算法print(ac) #b‘0Cu1dAA7U6tFsJS4EVkIDLcvKDU=‘response_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:9527\r\n\r\n"response_str = response_tpl % (ac.decode(‘utf-8‘))conn.send(response_str.encode(‘utf8‘)) # 响应握手信息while True: msg = conn.recv(8096) print(msg)

解密过程  分析

hashstr = b‘\x81\x89\xbc\xb8,\xb5Y6\xa5P\x12\x0b\xc8\x0f:‘# b‘\x81 \x83 \xceH\xb6\x85\xffz\x85‘# 将第二个字节也就是 \x83 第9-16位 进行与127进行 与位 运算payload = hashstr[1] & 127print(payload)if payload == 127: extend_payload_len = hashstr[2:10] # 顾头不顾尾 mask = hashstr[10:14] decoded = hashstr[14:]# 当位运算结果等于127时,则第3-10个字节为数据长度# 第11-14字节为mask 解密所需字符串# 则数据为第15字节至结尾if payload == 126: extend_payload_len = hashstr[2:4] mask = hashstr[4:8] decoded = hashstr[8:]# 当位运算结果等于126时,则第3-4个字节为数据长度 255*255+510=65535 21845汉字# 第5-8字节为mask 解密所需字符串# 则数据为第9字节至结尾if payload <= 125: extend_payload_len = None mask = hashstr[2:6] decoded = hashstr[6:]# 当位运算结果小于等于125时,则这个数字就是数据的长度# 第3-6字节为mask 解密所需字符串# 则数据为第7字节至结尾str_byte = bytearray() # 字节流for i in range(len(decoded)): byte = decoded[i] ^ mask[i % 4] # 异或运算 上下为0就是0 上下有1就是1 str_byte.append(byte)print(str_byte.decode("utf8")) # 主要解决中文 厉害了# # 1字节 = 8 bit 与位运算# 01111111# 11111111 最大255# 2/4/8/16/32/64/138/256

加密过程 -- 分析

import structmsg_bytes = "hello".encode("utf8") #转bytestoken = b"\x81" # 数据标准格式length = len(msg_bytes) #第二位就是数据长度if length < 126: token += struct.pack("B", length)elif length == 126: token += struct.pack("!BH", 126, length)else: token += struct.pack("!BQ", 127, length)msg = token + msg_bytesprint(msg)

总结

WebSocket的工作原理1.握手 客户端 Sec-WebSocket-Key 响应 base64(sha1(Sec-WebSocket-Key + magic_string)) 2.解密： 与127 进行"与"位运算 结果是两个数的最小值 1. == 127 第3-10个字节表示该数据的长度 2. == 126 第3-4个字节表示该数据的长度 00000010 00000000 3. <= 125 当前这个数字就是Websocket发送的数据长度