如何在Python中建立一个安全的WebSocket服务器

通常情况下，当一个Web应用需要从外部服务器获得一些东西时，客户端会向该服务器发送一个请求，服务器会做出响应，随后连接会被关闭。

考虑一个显示股票价格的网络应用。客户端必须反复向服务器请求更新价格以提供最新的价格。这种方法效率很低，因为它不断地破坏HTTP连接，而且也不太理想，因为存储在数据库中的股票的实时价格可能会改变，而应用程序仍然显示旧的价格。

然而，通过双向通信，服务器可以在每次信息更新时独立向客户端推送新的股票价格。工程师们设计了 WebSocket协议，以实现这种类型的持久性连接，同时避免了与老式的长线投票方法相关的问题。

实现WebSocket的一个流行选择是 Socket.IO 库。它的主要优势之一是，这个库可以在各种编程语言环境中使用，包括Python。它还为我们在两种环境中提供了更直接和一致的API，而不需要直接使用原始的WebSocket API。此外，该库使我们能够轻松实现安全措施，如跨源资源共享（CORS）、用户认证和有效载荷大小限制。

本教程将探讨在 Python 中建立一个安全的WebSocket服务器，使用 python-socket.io 和JavaScript Socket.IO客户端来连接我们的客户端和服务器。

用Python创建一个WebSocket服务器

安装和设置

要开始工作，你必须在你的机器上安装Python 3.x和它的软件包管理器pip。幸运的是，所有最近的Python版本都包括pip。如果你没有Python，请为你的系统下载最新的版本。

现在，让我们来创建一些源文件。首先，创建一个文件夹来存放示例代码。之后，用pip安装 python-socketio 包。

mkdir quick-socket
cd quick-socket
pip install python-socketio
然后，在quick-socket 中创建一个新的app.py 文件。这就是你要放套接字服务器代码的地方。
在Python中创建一个WebSocket服务器
让我们首先导入Python的Socket.IO库，并在app.py 中创建一个异步套接字服务器实例。我们将通过把async_mode 设置为asgi 来使它成为一个 ASGI 服务器。之后，在实例上设置事件监听器。
import socketio
server_io = socketio.AsyncServer(async_mode='asgi')
# a Python dictionary comprised of some heroes and their names
hero_names = {
  "ironMan": "Tony Stark",
  "hulk": "Bruce Banner",
  "wonderWoman": "Diana",
  "batMan": "Bruce Wayne",
  "blackPanther": "T'Challa"
# Triggered when a client connects to our socket. 
@server_io.event
def connect(sid, socket):    
    print(sid, 'connected')
# Triggered when a client disconnects from our socket
@server_io.event
def disconnect(sid):
    print(sid, 'disconnected')
@server_io.event
def get_name(sid, data):
    """Takes a hero, grabs corresponding “real” name, and sends it back to the client
    Key arguments:
    sid - the session_id, which is unique to each client
    data - payload sent from the client
    print(data["hero"])
    server_io.emit("name", {'hero_name': hero_names[data["hero"]]}, to=sid)
当一个套接字连接到我们的服务器时，第一个事件将发生。相应的connect 函数需要两个参数。
sid 或 是一个唯一的ID，代表一个连接的客户端。session id
socket 是一个字典，包含所有与客户端有关的信息。传递给 ，以便在验证客户端时检查识别信息（例如，一个用户名）。connect()
最后一个事件是在客户端应用程序上的一个按钮被点击时触发的。有效载荷信息（英雄）被用来从文件顶部定义的字典中访问相应的名字。
你需要运行一个网络服务器，将基于服务器的套接字应用程序暴露给客户端。为此，我们将使用ASGI。幸运的是，python-socket.io库附带了ASGIApp ，以帮助我们将app.py 转变为一个ASGI应用程序，你可以连接到WSGI服务器（如Gunicorn）。
你要在server_io = socketio.AsyncServer() 下面的行中实例化这个类的实例，传入 socket 服务器实例和你先前创建的静态文件的路径。
app = socketio.ASGIApp(server_io, static_files={
    '/': '/client/index.html'
    '/index.js': '/client/index.js'
最后，你需要部署该应用程序。Socket.IO服务器有各种部署策略，如Gunicorn和Eventlet。
使用Socket.IO创建一个WebSocket客户端
现在你已经创建了一个WebSocket服务器，现在是时候制作一个WebSocket客户端来与之通信。
首先，在你的根文件夹内创建一个client 文件夹。然后，创建两个文件：index.html 和index.js 。
在index.html ，定义一些简单的HTML标记，并嵌入到Socket.IO客户端、Bootstrap CSS和你的本地脚本文件index.js 的链接。
<!DOCTYPE html>
<html lang="en">
    <title>SocketIO Demo</title>
    <!-- Scripts -->
    <script src="https://cdn.socket.io/4.5.0/socket.io.min.js" integrity="sha384-7EyYLQZgWBi67fBtVxw60/OWl1kjsfrPFcaU0pp0nAh+i8FD068QogUvg85Ewy1k" crossorigin="anonymous"></script>
    <script src="index.js"></script>
    <!-- Bootstrap CDN -->
    <link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/bootstrap@3.3.7/dist/css/bootstrap.min.css" integrity="sha384-BVYiiSIFeK1dGmJRAkycuHAHRg32OmUcww7on3RYdg4Va+PmSTsz/K68vbdEjh4u" crossorigin="anonymous">
    <!-- Custom styles -->
    <style>
            text-align: center;
            margin-top: 10px;
            margin-bottom: 40px;
        div {
            width: 80%;
            text-align: center;
            margin: 0 auto;
    </style>
</head>
    <h1>Get Hero Name</h1>
        <button id="ironMan" type="button" class="btn btn-primary btn-lg btn-block">Iron Man</button>
        <button id="hulk" type="button" class="btn btn-primary btn-lg btn-block">Hulk</button>
        <button id="wonderWoman" type="button" class="btn btn-primary btn-lg btn-block">Wonder Woman</button>
        <button id="blackPanther" type="button" class="btn btn-primary btn-lg btn-block">Black Panther</button>
        <button id="batMan" type="button" class="btn btn-primary btn-lg btn-block">Batman</button>
    </div>
</body>
</html>
你将在index.js 内处理服务器发出的所有事件，当页面上的任何按钮被点击时，将发出一个事件。
  const client_io = io();
  const buttons = document.getElementsByTagName("button");
  client_io.on('connect', () => {
    console.log("connected")
    for (i=0; i<buttons.length; i++) {
      // add click event listener to all buttons
      buttons[i].addEventListener('click', function () {
      // emit hero to server
        client_io.emit("get_name", { hero: this.id})            
  client_io.on('disconnect', () => {
    console.log("disconnected")
  client_io.on("name", (data) => {
    alert(data.hero_name)
上面的代码抓取被点击的按钮的id 属性，并将该值发射给服务器。作为回应，服务器将触发name 事件。在执行name 事件时运行的最后一个回调中，我们使用Javascriptalert 函数显示名称。
在你的浏览器中，导航到127.0.0.1:8000/index.html。该页面应该显示一个按钮的列表，每个英雄都有一个。
点击任何一个按钮，发出get_name 事件。作为回应，你会得到一个显示英雄名字的警报。
让我们继续讨论WebSocket的安全性。
保护你的WebSockets
你需要在服务器上实现某些功能来保护你的WebSockets。
其中一些功能在python-socketio中是默认启用的，与其他许多WebSocket服务器库一样。例如，该库在WebSocket连接中自动执行HTTP压缩，这有助于提高传输速度和带宽使用。
你必须在初始化WebSocket服务器时实现或启用其他功能。
使用 CORS 来允许或阻止域名
跨源资源共享（CORS）是一种机制，使网络客户能够进行跨源请求。CORS限制有助于保护服务器免受跨站请求伪造（CSRF）攻击，在这种攻击中，攻击者通常会使受害者进行非预期的、通常是有害的行动，如转移个人资金。
然而，CORS所提供的保护仅限于HTTP连接。CORS策略不适用于WebSockets，因为WebSocket连接使用WebSocket（WS）或WebSocketSecure（WSS）协议。在这些协议中，初始握手通过HTTP升级请求发生，响应体被忽略，HTTP/HTTPS协议升级为WS/WSS协议。
由于CORS不限制对WebSocket协议的访问，恶意用户可以轻易地进行跨源WebSocket连接，以发送和接收恶意数据。解决方案是通过在客户端请求中添加upgrade 头域，将协议升级为WS。
Connection: Upgrade
Upgrade: websocket
作为回应，服务器将发送一个101 switching protocols 消息，确认后续通信可以通过WebSocket进行。幸运的是，Socket.IO客户端库会自动执行这一过程。
服务器验证Upgrade 请求上的Origin 头，以防止不需要的跨源 WS 连接。python-socket.io库提供了一种通过cors_allowed_origins 来实现的方法。在服务器端初始化WebSocket时，这个参数可以接受一个单一的起源或一个起源列表（一个URL数组）。
你也可以将此参数设置为'*' ，以允许所有来源，或设置为[] ，以阻止所有来源。
server_io = socketio.Server(cors_allowed_origins = '*')
你也可能想阻止来自你以外的来源的WebSocket连接，以防止跨站WebSocket劫持（CSWSH）攻击。这种类型的攻击是CSRF攻击的一个变种，通过WebSocket实现读/写通信。
在这里，攻击者可以在他们的域名上做一个恶意的网页，并伪装成用户建立与服务器套接字的连接。然后，恶意的应用程序可以读取服务器发送的信息，并直接写入服务器。
请注意，一些非浏览器客户端可以轻易地设置Origin头。因此，确保你用其他形式的客户端认证来补充这种方法。
在连接前验证WebSocket客户端
随着网络的规模和复杂性的增加，网络攻击的性质也在增加。认证系统是防止数据被盗和确保客户端-服务器通信隐私的最有效方法之一。
用户认证系统就像客户端和服务器之间的壁垒。任何想要访问服务器上资源的用户必须首先在登录请求中提供独特的识别信息--通常是一个用户名和密码。
服务器将验证用户，并根据这些信息批准或拒绝登录请求。在服务器计算机拒绝请求的情况下，它会向用户显示出错的原因，例如，"您的登录信息不正确"。这有助于确保资源不会落入坏人手中。
用户可以创建一个强大的密码，并使用一个密码管理器工具来提高认证的安全性。然而，这些步骤都取决于用户。这是在应用程序的服务器端实施强大的认证系统的另一个关键原因。
对于服务器端，有许多可用的认证库。你的选择取决于你使用哪个服务器端框架来构建应用程序。
对于我们的WebSocket例子，你可以在public 文件夹中创建一个login.html 文件，以渲染一个浏览器内的HTML表单。然后，用户可以提供他们的用户名和密码。
然后，你将在客户端脚本public/index.js ，等待一个submit 事件。当用户提交登录表格时，你将初始化客户端和服务器之间的WebSocket连接，将提供的用户名和密码作为extraHeaders 。
const client_io = null
document.getElementById('submit-btn').addEventListener('submit', function(ev){
  ev.preventDefault();
  client_io = io({
    transport_options: {
      polling: {
        extraHeaders: {
          'USERNAME': document.getElementById('username')
          'PASSWORD': document.getElementById('password')
上面的配置根据输入的id 属性检索用户名和密码值，将其编码为HTTP头，并在客户端连接后将其传输到WebSocket服务器。
为了在服务器代码中验证用户，app.py ，我们将从request 检索用户名和密码。
@server_io.event
def connect(sid, request):    
    username = request.get('USERNAME')
    password = request.get('PASSWORD')
    if not username && password:
        return False
    // Authenticate user here. Check if user exists in database, get related info.        
    with server_io.session(sid) as session:
        session['username'] = username
    server_io.emit('user_authorized', userdata, to=sid)
在这里，我们采取用户名和密码，并进行测试检查，看是否有认证数据。如果认证数据不存在，说明用户没有发送，我们就返回False 。在这种情况下，我们不能认证该用户。
否则，我们就继续对用户进行认证。通常，你会检查用户名是否存在于数据库中，然后用它来检索用户的特定内容。然后，你会创建一个用户会话，并向客户端发送user_authorized ，在客户端代码中使用JavaScript来向用户显示这些内容。
如果不对WebSockets进行认证，任何人都有可能连接到服务器并窃取敏感数据。注意，一旦用户被认证并登录，后续的认证应该使用令牌，而不是要求用户重新输入用户名和密码。
另外，在处理从客户端发送的数据时，你应该小心。确保在处理任何客户端的输入之前对其进行验证。像SQL注入这样的攻击可以通过WebSockets进行，就像在传统的HTTP连接中一样。
使用速率限制来保护你的WebSocket服务器免受攻击
速率限制是一种防止拒绝服务（DoS）和分布式拒绝服务（DDoS）攻击的技术。在这些攻击中，攻击者会试图通过多次重复调用，使服务对合法用户不可用，从而使服务不堪重负，甚至崩溃。
速率限制通过限制每个用户可以发出的API请求的频率来防止这些攻击。我们可以根据IP地址、API密钥或其他独特的标识符（如UniqueId ）等限制键来执行这一限制。
速率和分配限制，或者说配额，规定了一个客户在特定的时间范围内对API的请求数量。服务提供商通常采用这种方法来确保公平消费基于API的服务和资源。
在你的服务器上执行速率限制要求你首先知道为什么需要它--保护一项服务，为不同的计划设置配额，等等。接下来，你需要确定并选择最适合你的情况的限制性密钥。然后，你采用一个限制性的实现，根据你选择的密钥来跟踪API的使用。
虽然目前还没有专门针对python-socketio的速率限制包，但只要稍加努力，就可以实现一个解决方案。
限制有效载荷大小以保护你的WebSocket服务器
在WebSockets中发送巨大的有效载荷很可能会影响性能，并最终导致socket服务器崩溃。
为避免WebSocket变慢或崩溃，你可能希望限制通过WebSocket连接发送的消息的最大有效载荷大小。这将有助于你避免因发送过大的消息（如zip轰炸）而使服务器崩溃。
此外，调节消息大小可极大地减少WebSocket连接的延迟，并可显著提高传输速度。WebSocket协议限制了帧的大小，并帮助确定消息需要多少压缩，消息被压缩。压缩消息流会消耗内存和CPU资源，但通常是值得做的，因为它可以极大地减少网络流量。
使用TLS来创建安全套接字通信
使用TLS/SSL是在互联网上部署网站的必要条件。没有它，客户端和服务器之间传输的敏感信息很容易通过中间人攻击被盗。为了保证传输的安全，你应该使用一个安全的协议，如https://，而不是不安全的https://协议。
对于WebSockets，你应该使用wss://（TLS加密的WebSockets）协议而不是不安全的ws://协议。使用后一种方法使连接容易受到第三方的干扰。相反，wss://协议对WebSocket内发送的所有数据进行加密。
有了这种加密，任何第三方都无法读取或修改通过WebSocket发送的信息，从而确保敏感信息的安全。如果连接是安全的，其他类型的攻击也变得不可能。
此外，在与服务器建立WebSocket连接之前，确保请求网站也使用https:// - 没有它，恶意行为者可以轻易篡改请求。
TLS对于任何有或没有WebSockets的网站都是必要的。幸运的是，许多托管服务在部署应用程序时提供免费的TLS。一些最受欢迎的Python应用程序的托管平台是谷歌云、AWS、Azure、Heroku和Fly.io。
WebSocket 总结