WebSocket协议分析

0 实时跟服务端通信方案

# 1 轮询：
客户端向服务端无限循环发送http请求，一旦服务端有最新消息，从当次http响应中带回，客户端就能收到变化
# 2 长轮回（web版微信采用此方式）
客户端和服务端保持一个长连接（http），等服务端有消息返回就断开，
如果没有消息，就会hold住，等待一定时间，然后再重新连接，也是个循环的过程
# 3 websocket协议
客户端发起HTTP握手，告诉服务端进行WebSocket协议通讯，并告知WebSocket协议版本。
服务端确认协议版本，升级为WebSocket协议。之后如果有数据需要推送，会主动推送给客户端

1 WebSocket介绍

WebSocket是一种通信协议，可在单个TCP连接上进行全双工通信。WebSocket使得客户端和服务器
之间的数据交换变得更加简单，允许服务端主动向客户端推送数据。在WebSocket API中，
浏览器和服务器只需要完成一次握手，两者之间就可以建立持久性的连接，并进行双向数据传输。

"""说白了就是WebSocket可以主动向客户端推送消息实现双工通信"""

握手阶段采用 HTTP 协议。
数据格式轻量，性能开销小。客户端与服务端进行数据交换时，服务端到客户端的数据包头
只有2到10字节，客户端到服务端需要加上另外4字节的掩码。HTTP每次都需要携带完整头部。
更好的二进制支持，可以发送文本，和二进制数据
没有同源限制，客户端可以与任意服务器通信

1.1 请求协议

GET / HTTP/1.1rn  # 请求首行，握手阶段还是使用http协议
Host: 127.0.0.1:8080rn # 请求头
Connection: Upgradern  # 表示要升级协议
Pragma: no-cachern
Cache-Control: no-cachern
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.36rn
Upgrade: websocketrn  # 要升级协议到websocket协议
Origin: http://localhost:63342rn
Sec-WebSocket-Version: 13rn  # 表示websocket的版本。如果服务端不支持该版本，需要返回一个Sec-WebSocket-Versionheader，里面包含服务端支持的版本号
Accept-Encoding: gzip, deflate, brrn
Accept-Language: zh-CN,zh;q=0.9,en;q=0.8rn
Sec-WebSocket-Key: 07EWNDBSpegw1vfsIBJtkg==rn # 对应服务端响应头的Sec-WebSocket-Accept，由于没有同源限制，websocket客户端可任意连接支持websocket的服务。这个就相当于一个钥匙一把锁，避免多余的，无意义的连接
Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bitsrnrn

1.2 响应协议（响应成功，便握手成功，就可以双工通信了）

HTTP/1.1 101 Switching Protocolsrn  # 响应首行，还是使用http协议
Upgrade:websocketrn                 # 表示要升级到websocket协议
Connection: Upgradern               # 表示要升级协议
Sec-WebSocket-Accept: 07EWNDBSpegw1vfsIBJtkg==rn # 根据客户端请求首部的Sec-WebSocket-Key计算出来。
WebSocket-Location: ws://127.0.0.1:8000rnrn

1.3 通信数据帧格式

详见WebSocket协议

2 使用python的socket模块写一个websocket服务端

2.1 服务端

import base64
import hashlib
import socket

sock = socket.socket()
# 绑定host，默认端口8000
sock.bind(("127.0.0.1", 8000))
# 半连接池大小
sock.listen(5)


def get_headers(data):
    """
    将请求头转换为字典
    """
    header_dict = {}
    data = str(data, encoding="utf-8")
    """
    请求头格式：
    GET / HTTP/1.1rn  # 请求首行，握手阶段还是使用http协议
    Host: 127.0.0.1:8080rn # 请求头
    Connection: Upgradern  # 表示要升级协议
    Pragma: no-cachern
    Cache-Control: no-cachern
    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.36rn
    Upgrade: websocketrn  # 要升级协议到websocket协议
    Origin: http://localhost:63342rn
    Sec-WebSocket-Version: 13rn  # 表示websocket的版本。如果服务端不支持该版本，需要返回一个Sec-WebSocket-Versionheader，里面包含服务端支持的版本号
    Accept-Encoding: gzip, deflate, brrn
    Accept-Language: zh-CN,zh;q=0.9,en;q=0.8rn
    Sec-WebSocket-Key: 07EWNDBSpegw1vfsIBJtkg==rn # 对应服务端响应头的Sec-WebSocket-Accept，由于没有同源限制，websocket客户端可任意连接支持websocket的服务。这个就相当于一个钥匙一把锁，避免多余的，无意义的连接
    Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bitsrnrn
    """
    header, body = data.split("rnrn", 1)  # 因为请求头信息结尾都是rn,并且最后末尾部分是rnrn；
    # 所以以此分割
    header_list = header.split("rn")
    for i in range(0, len(header_list)):
        if i == 0:
            if len(header_list[0].split(" ")) == 3:
                header_dict['method'], header_dict['url'], header_dict['protocol'] = header_list[0].split(" ")
        else:
            k, v = header_list[i].split(":", 1)
            header_dict[k] = v.strip()
    return header_dict


def get_data(info):
    """
    对返回消息进行解码
    比较复杂，详见数据帧格式解析：
    https://www.cnblogs.com/chyingp/p/websocket-deep-in.html
    """
    payload_len = info[1] & 127
    if payload_len == 126:
        mask = info[4:8]
        decoded = info[8:]
    elif payload_len == 127:
        mask = info[10:14]
        decoded = info[14:]
    else:
        mask = info[2:6]
        decoded = info[6:]
    bytes_list = bytearray()  # 使用字节将数据全部收集，再去字符串编码，这样不会导致中文乱码
    for i in range(len(decoded)):
        chunk = decoded[i] ^ mask[i % 4]  # 解码方式
        bytes_list.append(chunk)
    body = str(bytes_list, encoding='utf-8')
    return body


# 等待用户连接
conn, addr = sock.accept()
print("客户端连接了：", conn, addr)
# 获取握手消息，magic string ,sha1加密
data = conn.recv(1024)
headers = get_headers(data)

# 对请求头中的sec-websocket-key进行加密
res_tpl = "HTTP/1.1 101 Switching Protocolsrn" 
          "Upgrade:websocketrn" 
          "Connection: Upgradern" 
          "Sec-WebSocket-Accept: %srn" 
          "WebSocket-Location: ws://%s%srnrn"
# 固定的RFC规定的
magic_string = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'
# 确认握手Sec-WebSocket-Key固定格式：headers头部的Sec-WebSocket-Key+'258EAFA5-E914-47DA-95CA-C5AB0DC85B11'
# 确认握手的秘钥值为 传入的秘钥+magic_string，使用sha1算法加密，然后base64转码
value = headers['Sec-WebSocket-Key'] + magic_string
ac = base64.b64encode(hashlib.sha1(value.encode('utf-8')).digest())
response_str = res_tpl % (ac.decode('utf-8'), headers['Host'], headers['url'])

# 响应握手信息，这样就建立连接成功，双方就可以互通信息了
conn.send(bytes(response_str, encoding='utf-8'))
# 可以进行通信--接收客户端发送的消息
while True:
    data = conn.recv(1024)
    data = get_data(data)
    print(data)

2.2 客户端

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>websocket通信</title>
    <script src="https://cdn.bootcdn.net/ajax/libs/jquery/3.6.0/jquery.js"></script>
</head>
<body>
请输入：<input type="text" id="chat-message-input">
<button id="chat-message-submit">发送</button>
<hr>
通信内容：
<br>
<textarea id="chat-log" cols="30" rows="10"></textarea>
</body>
<script>
    var chatSocket = new WebSocket('ws://127.0.0.1:8000');

    chatSocket.onmessage = function (e) {
        var data = JSON.parse(e.data);
        var message = data['message'];
        console.log(message)
        var datamsg = $('#chat-log').val() + message + 'n'
        $('#chat-log').val(datamsg)
    };

    chatSocket.onclose = function (e) {
        console.error(e);
    };
    $('#chat-message-submit').click(function () {
        console.log(chatSocket.readyState)
        if (chatSocket.readyState === 1) {
            var message = $('#chat-message-input').val()
            chatSocket.send(JSON.stringify({
                'message': message
            }));
            $('#chat-message-input').val("")
        } else {
            console.log("还没有连接")
        }


    })
</script>
</html>

3 django实现websocket集成

3.1 使用dwebsocket集成

该模块在django 3.0以后就不支持了，推荐使用channels
详见Django实现websocket完成实时通讯，聊天室，在线客服等

3.2 使用django+channels集成

本案例使用的环境是(channels2.0最低django版本要求是1.11+，python3.5+)
python 3.6
Django 2.2.2
channels 2.1.7
channels-redis 2.3.3

第一步：安装模块

Django==2.1.4 #channels2.0最低django版本要求是1.11+，python3.5+
channels==2.1.7
channels-redis==2.3.3

第二步：配置settings.py

Copy#1 加入app
INSTALLED_APPS = [
    ...
    'app01.apps.App01Config',
    'channels',
]
#2 配置channels layer
ASGI_APPLICATION = 'django_websocket.routing.application' #自己routing的路径
CHANNEL_LAYERS = {
    # 真实上线使用redis
    # 'default': {
    #     'BACKEND': 'channels_redis.core.RedisChannelLayer',
    #     'CONFIG': {
    #         "hosts": [('127.0.0.1', 6379)],  # 需修改redis的地址
    #     },
    # },
  # 测试阶段，放到内存中即可
    'default': {
        'BACKEND': 'channels.layers.InMemoryChannelLayer',
    },
}

第三步：配置路由
在项目settings文件同级目录中新增routing.py

from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from app01 import routing

application = ProtocolTypeRouter({
    # 'websocket': AuthMiddlewareStack( # 使用AuthMiddlewareStack，后续在视图类中可以取出self.scope，相当于request对象
    #     URLRouter(
    #         routing.websocket_urlpatterns# 指明路由文件是django_websocket/routing.py,类似于路由分发
    #     )
    # ),
    'websocket': URLRouter(
            routing.websocket_urlpatterns  # 指明路由文件是django_websocket/routing.py,类似于路由分发
),
})

最后在app里配置路由和对应的消费者，(我是app01下的routing.py)：

from django.urls import path
from . import consumers

websocket_urlpatterns = [
    path('ws/chat', consumers.Chatting), #consumers.Chatting 是该路由的消费者
]

第四步:在app目录下编写consumers.py

from channels.generic.websocket import AsyncWebsocketConsumer
import json
class Chatting(AsyncWebsocketConsumer):
    async def connect(self):
        print('connect')
        # AuthMiddlewareStack：封装了django的auth模块，使用这个配置我们就可以在consumer中通过下边的代码获取到用户的信息
        print(self.scope)
        self.room_group_name = 'xiaoyuanqujing'
        # 加入聊天室
        await self.channel_layer.group_add(
            self.room_group_name,
            self.channel_name
        )

        await self.accept()

    async def disconnect(self, close_code):
        # 离开聊天室
        print('disconnect')
        print(self.scope)
        await self.channel_layer.group_discard(
            self.room_group_name,
            self.channel_name
        )

    # 通过WebSocket，接收数据
    async def receive(self, text_data):
        print('receive')
        text_data_json = json.loads(text_data)
        message = text_data_json['message']
        print(message)
        # Send message to room group
        await self.channel_layer.group_send(
            self.room_group_name,
            {
                'type': 'chat_message', # 写一个当前类中的方法名，会触发该方法执行,组里有多少人会触发多少次
                'message': message,
                'username':str(self.scope['query_string']) # 把当前客户端的名字传入（也可以去ip地址）
            }
        )

    # Receive message from room group
    async def chat_message(self, event):
        print('chat_message')
        # 取出发言人的名字和说的话
        message = event['username'] + event['message']
        # 通过WebSocket发送
        await self.send(text_data=json.dumps({
            'message': message
        }))

第五步：前端发起websocket请求

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>websocket通信</title>
    <script src="https://cdn.bootcdn.net/ajax/libs/jquery/3.6.0/jquery.js"></script>
</head>
<body>
<p>姓名：<input type="text" id="username">
    <button id="btn">加入聊天室</button>
</p>
请输入：<input type="text" id="chat-message-input">
<button id="chat-message-submit">发送</button>
<hr>
通信内容：
<br>
<textarea id="chat-log" cols="30" rows="10"></textarea>
</body>
<script>
    var chatSocket
    $('#btn').click(function () {
        var username = $('#username').val()
        chatSocket = new WebSocket('ws://' + window.location.host + '/ws/chat?name=' + username);
        chatSocket.onopen = function (e) {
            console.log(e)

        }
        chatSocket.onmessage = function (e) {
            var data = JSON.parse(e.data);
            var message = data['message'];
            console.log(message)
            var datamsg = $('#chat-log').val() + message + 'n'
            $('#chat-log').val(datamsg)
        };

        chatSocket.onclose = function (e) {
            console.error(e);
        };
    })


    $('#chat-message-submit').click(function () {
        console.log(chatSocket.readyState)
        if (chatSocket.readyState === 1) {
            var message = $('#chat-message-input').val()
            chatSocket.send(JSON.stringify({
                'message': message
            }));
            $('#chat-message-input').val("")
        } else {
            console.log("还没有连接")
        }


    })
</script>
</html>

第六步：启动redis，启动项目

注意（layer介绍）

信道层是一种通信系统。它允许多个消费者实例彼此交谈，以及与Django的其他部分交谈。

通道层提供以下抽象：
通道是一个可以将邮件发送到的邮箱。每个频道都有一个名称。任何拥有频道名称的人都可以向频道发送消息。
一组是一组相关的通道。一个组有一个名称。任何具有组名称的人都可以按名称向组添加/删除频道，并向组中的所有频道发送消息。无法枚举特定组中的通道。
每个使用者实例都有一个自动生成的唯一通道名，因此可以通过通道层进行通信。
在我们的聊天应用程序中，我们希望同一个房间中的多个聊天消费者实例相互通信。为此，我们将让每个聊天消费者将其频道添加到一个组，该组的名称基于房间名称。这将允许聊天用户向同一房间内的所有其他聊天用户发送消息。
我们将使用一个使用redis作为后备存储的通道层。要在端口6379上启动Redis服务器，首先系统上安装redis，并启动

3.3 django3.0默认支持websocket

新建一个Django3.0的项目（我是用django3.2.2），会发现项目路径下有一个asgi.py

import os
from django.core.asgi import get_asgi_application
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'django_channels_3.settings')
application = get_asgi_application() ```
该文件提供了默认的Django ASGI设置，并公开了一个名为application的ASGI应用程序
可以使用uvicorn或daphne等ASGI服务器运行该应用程序

ASGI介绍

ASGI:“异步服务器网关接口”是用于使用Python构建异步Web服务的规范。
它是WSGI的精神继承者，WSGI已被Django和Flask等框架使用了很长时间。
ASGI使您可以使用Python的本机异步/等待功能来构建支持长期连接的Web服务，例如Websockets和Server Sent Events。
ASGI应用程序是一个异步函数，它带有3个参数：作用域（当前请求的上下文），接收（一个异步函数，可让您侦听传入的事件）和发送（一个异步函数，可将事件发送至客户端）。
在ASGI应用程序内部，您可以根据范围字典中的值路由请求。
例如，您可以通过检查scope [‘type’]的值来检查该请求是HTTP请求还是Websocket请求。要侦听来自客户端的数据，您可以等待接收功能。准备好将数据发送到客户端时，可以等待发送功能，然后将要发送给客户端的任何数据传递给客户端

第一步：修改asgi.py

import os

from django.core.asgi import get_asgi_application
from app01.websocket import websocket_application

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'django_channels_3.settings')
application_asgi = get_asgi_application()
async def application(scope, receive, send):
    print(scope)
    if scope['type'] == 'http':
        print('http')
        await application_asgi(scope, receive, send)
    elif scope['type'] == 'websocket':
        print('websocket')
        await websocket_application(scope, receive, send)
    else:
        raise NotImplementedError("Unknown scope type %s"%scope['type'])

第二步：在app01下新建一个websocket.py

async def websocket_application(scope, receive, send):

    while True:
        event = await receive()
        if event['type'] == 'websocket.connect':
            await send({
                'type': 'websocket.accept'
            })

        if event['type'] == 'websocket.disconnect':
            break

        if event['type'] == 'websocket.receive':
            print(event['text'])
            import json
            # 收到的内容
            rec=json.loads(event['text'])['message']
            await send({
                'type': 'websocket.send',
                'text': json.dumps({'message':'收到了你的：%s'%rec})
            })

第三步：在app01的view.py中写入

def index(request):
    return render(request,'index.html')

第四步：配置路由urls.py

from app01 import views
urlpatterns = [
    path('admin/', admin.site.urls),
    path('', views.index),
]
第五步：编写测试html页面（index.html）
<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>websocket通信</title>
    <script src="https://cdn.bootcdn.net/ajax/libs/jquery/3.6.0/jquery.js"></script>
</head>
<body>
<button id="btn">点我连接</button>
<hr>
请输入：<input type="text" id="chat-message-input">
<button id="chat-message-submit">发送</button>
<hr>
通信内容：
<br>
<textarea id="chat-log" cols="30" rows="10"></textarea>
</body>
<script>
    var chatSocket
    $('#btn').click(function () {
        chatSocket = new WebSocket('ws://127.0.0.1:8000');

        chatSocket.onmessage = function (e) {
            console.log(e)
            var data = JSON.parse(e.data);

            var message = data['message'];
            console.log(message)
            var datamsg = $('#chat-log').val() + message + 'n'
            $('#chat-log').val(datamsg)
        };

        chatSocket.onclose = function (e) {
            console.error(e);
        };

    })

    $('#chat-message-submit').click(function () {
        console.log(chatSocket.readyState)
        if (chatSocket.readyState === 1) {
            var message = $('#chat-message-input').val()
            chatSocket.send(JSON.stringify({
                'message': message
            }));
            $('#chat-message-input').val("")
        } else {
            console.log("还没有连接")
        }


    })
</script>
</html>

第六步：使用uvicorn 启动项目

# pip3 install uvicorn
uvicorn django_channels_3.asgi:application

4 基于django+websocket+paramiko实现跳板机(堡垒机)

4.1 app01下的routing.py

websocket_urlpatterns = [
    path('ws/chat', consumers.Chatting), #consumers.Chatting 是该路由的消费者
    re_path('ws/webssh/(?P<host>.*)/(?P<username>w+)/(?P<password>w+)', consumers.SSHConsumer),
]

4.2 app01下的views.py

def webterminal(request):
    if request.method=='GET':
        return render(request,'webterminal_input.html')
    else:
        host=request.POST.get('host')
        username=request.POST.get('username')
        password=request.POST.get('password')
        print(host)
        print(username)
        return render(request,'webterminal.html',locals())

4.3 webterminal_input.html和webterminal.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>请输入远端</title>
    <script src="https://cdn.bootcdn.net/ajax/libs/jquery/3.6.0/jquery.js"></script>
</head>
<body>
<form action="" method="post">
    <p>地址：<input type="text" name="host"></p>
    <p>用户名：<input type="text" name="username"></p>
    <p>密码：<input type="text" name="password"></p>
    <input type="submit" value="连接">
</form>

</body>

<script>
</script>
</html>
<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>终端</title>
    <script src="https://cdn.bootcdn.net/ajax/libs/jquery/3.6.0/jquery.js"></script>
    <link rel="stylesheet" href="/static/css/xterm.css"/>
    <script src="/static/js/xterm.js"></script>



</head>
<body style="margin: 0;padding: 0;">
<div id="terminal"></div>
</body>
<script>


    function terminal_size() {
        return {
            cols: Math.floor($('#terminal').width() / 9),
            rows: Math.floor($(window).height() / 17),
        }
    }

    let cols = terminal_size().cols;
    let rows = terminal_size().rows;

    var term = new Terminal({cursorBlink: true, useStyle: true, cols: cols, rows: rows});
    term.open(document.getElementById('terminal'));
    term.focus();
    var socket = new WebSocket('ws://' + window.location.host + '/ws/webssh/{{ host }}/{{ username }}/{{ password }}');
    socket.onopen = function () {

        term.onData(data => {
            let send_data = JSON.stringify({
                'flag': 'input',
                'data': data,
            });

            console.log(send_data)
            socket.send(send_data)
        })
    }


    socket.onerror = function (event) {
        console.log('error:' + event);
    };

    socket.onmessage = function (event) {
        data = JSON.parse(event.data);
        term.write(data.message);

        if (data.flag === 'failed') {
            socket.close();
        }
    };

    socket.onclose = function (event) {
        term.write('nrx1B[1;3;31m连接关闭.x1B[0m');
    };


</script>
</html>

4.4 app01下的consumers.py

import paramiko
from threading import Thread


class SSHConsumer(WebsocketConsumer):
    def connect(self):
        print(self.scope)
        query_params = self.scope['url_route']['kwargs']
        host = query_params['host']
        username = query_params['username']
        password = query_params['password']

        self.accept()
        # paramiko 建立连接
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        try:
            ssh.connect(hostname=host, port=22, username=username, password=password, timeout=8)
        except Exception as e:
            message = json.dumps({'flag': 'error', 'message': str(e)})
            self.send(message)
            return False

        # 打开一个ssh通道并建立连接
        transport = ssh.get_transport()
        self.ssh_channel = transport.open_session()
        self.ssh_channel.get_pty(term='xterm')
        self.ssh_channel.invoke_shell()
        recv = self.ssh_channel.recv(1024).decode('utf-8')
        message = json.dumps({'flag': 'success', 'message': recv})
        self.send(message)

    def disconnect(self, close_code):
        self.ssh.close()

    def receive(self, text_data=None):
        text_data = json.loads(text_data)
        # run_shell(data=text_data.get('data', ''))
        # 向远程服务器发送命令
        print(text_data.get('data', ''))
        Thread(target=self.ssh_channel.send,args=[text_data.get('data', '')]).start()
        # self.ssh_channel.send(text_data.get('data', ''))
        def recv_from_host():
            # 从远程服务器接收命令，返回给前端
            while not self.ssh_channel.exit_status_ready():
                data = self.ssh_channel.recv(1024).decode('utf-8', 'ignore')
                if len(data) != 0:
                    message = {'flag': 'success', 'message': data}
                    self.send(json.dumps(message))
                else:
                    break

        Thread(target=recv_from_host).start()

4.5 最终效果

5 flask实现websocket集成

待续……

6 FastApi实现websocket集成

待续……

代码007普通

打赏 收藏 海报 链接