Skip to main content
Snyk ブログ

In this article

Pythonで安全なWebSocketサーバーを構築する方法

著者

Kingsley Ubah

feature websocket server

2022年7月5日

0 分で読めます

一般に、Webアプリが外部サーバーから情報を取得する場合、クライアントがサーバーにリクエストを送信し、サーバーが応答した後、接続は終了します。

株価を表示するWebアプリを考えてみましょう。最新の株価を表示するには、クライアントがサーバーに繰り返しリクエストを送信する必要があります。この方法ではHTTP接続を何度も切断・再確立するため効率が悪く、アプリに古い価格が表示されている間に、データベースに保存されている実際の株価が変わる可能性もあるため、理想的とは言えません。

一方、双方向通信を使えば、情報が更新されるたびにサーバーからクライアントへ新しい株価を独立してプッシュできます。このような永続的な接続を実現し、従来のロングポーリングに伴う問題を避けるために、エンジニアはWebSocketプロトコルを開発しました。

WebSocketを実装する方法として広く使われているのが、Socket.IOライブラリです。主な利点の一つは、Pythonを含むさまざまなプログラミング言語の環境で利用できることです。また、生のWebSocket APIを直接使わなくても、どちらの環境でも一貫性のある、よりシンプルなAPIを利用できます。さらに、クロスオリジンリソース共有（CORS）、ユーザー認証、ペイロードサイズの制限といったセキュリティ対策も簡単に実装できます。

このチュートリアルでは、python-socket.ioとJavaScriptのSocket.IO clientを使ってクライアントをサーバーに接続し、安全なWebSocketサーバーをPythonで構築する方法を紹介します。

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関数は、次の2つの引数を受け取ります。

  • sid（session id）は、接続中のクライアントを識別する一意のIDです。

  • socketは、クライアントに関するすべての情報を含む辞書です。これをconnect()に渡すと、クライアントの認証時にユーザー名などの識別情報を確認できます。

最後のイベントは、クライアントアプリでボタンがクリックされたときに発生します。ペイロードメッセージ（hero）を使って、ファイルの先頭で定義した辞書から対応する名前を取得します。

サーバーベースのソケットアプリケーションをクライアントに公開するには、Webサーバーを実行する必要があります。ここではASGIを使用します。幸い、python-socket.ioライブラリにはASGIAppが用意されており、app.pyをASGIアプリケーションに変換して、WSGIサーバー（Gunicornなど）に接続できます。

このクラスのインスタンスを、server_io = socketio.AsyncServer()の次の行で生成します。その際、ソケットサーバーのインスタンスと、先ほど作成した静的ファイルへのパスを渡します。

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の2つのファイルを作成します。

index.html内にシンプルなHTMLマークアップを定義し、Socket.IO client、Bootstrap CSS、ローカルのスクリプトファイルindex.jsへのリンクを埋め込みます。

<!DOCTYPE html>
<html lang="en">
<head>    
    <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>
        h1 {
            text-align: center;
            margin-top: 10px;
            margin-bottom: 40px;
        }

        div {
            width: 80%;
            text-align: center;
            margin: 0 auto;
        }
    </style>
</head>
<body>
    <h1>Get Hero Name</h1>

    <div>
        <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>
「ヒーロー名を取得」というタイトルのウェブページのモックアップ。Iron Man、Hulk、Wonder Woman、Black Panther、Batmanと表示されたボタンが並んでいる。

サーバーが送信するすべてのイベントは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イベントの実行時に呼び出される最後のコールバックでは、JavaScriptのalert関数を使って名前を表示します。

ブラウザーで127.0.0.1:8000/index.htmlにアクセスします。ヒーローごとに1つずつ、ボタンの一覧が表示されます。

任意のボタンをクリックするとget_nameイベントが送信されます。これに応じて、ヒーローの名前を示すアラートが表示されます。

ブラウザーのデモ画面。Hulk、Wonder Woman、Black Panther、Batmanと書かれたボタンの上に、「Bruce Wayne」というポップアップメッセージが表示されています。

次に、WebSocketのセキュリティについて見ていきましょう。

WebSocketを安全に保護する

WebSocketを安全に保つには、サーバーにいくつかの機能を実装する必要があります。

他の多くのWebSocketサーバーライブラリと同様に、python-socketioでは一部の機能がデフォルトで有効になっています。たとえば、このライブラリはWebSocket接続でHTTP圧縮を自動的に実行し、転送速度と帯域幅の使用効率を向上させます。

その他の機能は、WebSocketサーバーの初期化時に実装または有効化する必要があります。

CORSで許可するドメインとブロックするドメインを設定する

クロスオリジンリソース共有（CORS）は、Webクライアントが異なるオリジンにリクエストを送信できるようにする仕組みです。CORSの制限は、攻撃者が被害者に個人資金の送金など、意図しない、しばしば有害な操作を実行させるクロスサイトリクエストフォージェリ（CSRF）攻撃からサーバーを保護するのに役立ちます。

ただし、CORSによる保護はHTTP接続に限られます。WebSocket接続ではWebSocket（WS）またはWebSocketSecure（WSS）プロトコルを使用するため、CORSポリシーは適用されません。これらのプロトコルでは、最初のハンドシェイクはHTTP Upgradeリクエストを介して行われ、レスポンスボディは無視されます。その後、HTTP/HTTPSプロトコルがWS/WSSプロトコルにアップグレードされます。

CORSではWebSocketプロトコルへのアクセスを制限できないため、悪意のあるユーザーが簡単にクロスオリジンWebSocket接続を確立し、悪意のあるデータを送受信できてしまいます。この問題を解決するには、クライアントのリクエストにupgradeヘッダーフィールドを追加して、プロトコルをWSにアップグレードします。

Connection: Upgrade
Upgrade: websocket

これに応じて、サーバーは101 switching protocolsメッセージを送信し、以降の通信をWebSocketで行えることを示します。幸い、Socket.IO clientライブラリがこの処理を自動で行います。その他の設定オプションについてはこちらをご覧ください。

サーバーはOriginヘッダーをUpgradeリクエストで検証し、望ましくないクロスオリジンのWS接続を防ぎます。python-socket.ioライブラリでは、cors_allowed_originsを使ってこの検証を行えます。このパラメーターには、サーバー側でWebSocketを初期化する際に、単一のオリジンまたはオリジンのリスト（URLの配列）を指定できます。

この引数に'*'を指定してすべてのオリジンを許可することも、[]を指定してすべてのオリジンをブロックすることもできます。

server_io = socketio.Server(cors_allowed_origins = '*')

また、クロスサイトWebSocketハイジャック（CSWSH）攻撃を防ぐため、自分のオリジン以外からのWebSocket接続を拒否することも検討してください。この攻撃はCSRF攻撃の一種で、WebSocketを介した読み書きの通信を可能にします。

攻撃者は、自分のドメインに悪意のあるWebページを作成し、ユーザーを装ってサーバーソケットに接続できます。すると、悪意のあるアプリケーションからサーバーが送信するメッセージを読み取ったり、サーバーに直接データを書き込んだりできてしまいます。

ブラウザー以外の一部のクライアントでは、Originヘッダーを簡単に設定できる点に注意してください。そのため、この方法に加えて、他の形式のクライアント認証も必ず導入してください。

接続前にWebSocketクライアントを認証する

Webが規模と複雑さを増すにつれて、サイバー攻撃も変化しています。認証システムは、データの窃取を防ぎ、クライアントとサーバー間の通信のプライバシーを守るための最も効果的な方法の一つです。

ユーザー認証システムは、クライアントとサーバーの間の防壁として機能します。サーバー上のリソースにアクセスするユーザーは、まずログインリクエストで一意の識別情報（通常はユーザー名とパスワード）を提示する必要があります。

サーバーは提示された情報でユーザーを検証し、ログインリクエストを承認または拒否します。サーバーがリクエストを拒否した場合は、「ログイン情報が正しくありません」など、何が問題だったのかをユーザーに表示します。これにより、リソースが悪意のある者の手に渡るのを防ぎます。

ユーザーは強力なパスワードを作成し、パスワードマネージャーを使うことで認証の安全性を高められます。しかし、こうした対策はすべてユーザーの行動に依存します。これも、アプリケーションのサーバー側に堅牢な認証システムを実装すべき重要な理由です。

サーバー側の認証には、さまざまなライブラリを利用できます。どれを選ぶかは、アプリケーションの構築に使用するサーバーサイドフレームワークによって異なります。

このWebSocketの例では、publicフォルダーにlogin.htmlファイルを作成し、ブラウザー上にHTMLフォームを表示できます。ユーザーはここでユーザー名とパスワードを入力します。

次に、クライアントスクリプトpublic/index.js内でsubmitイベントを待ち受けます。ユーザーがログインフォームを送信したら、入力されたユーザー名とパスワードをextraHeadersとして渡し、クライアントとサーバー間のWebSocket接続を初期化します。

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を使って、このコンテンツをユーザーに表示します。

WebSocketを認証しなければ、誰でもサーバーに接続し、機密データを盗み出せる可能性があります。ユーザーの認証とログインが完了した後は、ユーザーにユーザー名とパスワードを再入力させるのではなく、トークンを使って以降の認証を行ってください。

また、クライアントから送信されたデータの取り扱いにも注意が必要です。クライアントからの入力は処理する前に必ず検証してください。従来のHTTP接続と同様に、WebSocket経由でもSQLインジェクション攻撃が行われる可能性があります。

レート制限でWebSocketサーバーを攻撃から保護する

レート制限は、サービス拒否（DoS）攻撃や分散型サービス拒否（DDoS）攻撃からサービスを保護する手法です。こうした攻撃では、サービスが正当なユーザーに利用できなくなるまで、攻撃者が大量のリクエストを繰り返し送信し、サービスを圧倒したりクラッシュさせたりしようとします。

レート制限は、ユーザーごとに実行できるAPIリクエストの頻度を制限することで、こうした攻撃を防ぎます。IPアドレス、APIキー、UniqueIdのような一意の識別子など、制限キーに基づいて制限を適用できます。

レート制限や割り当て上限（クォータ）では、クライアントが一定期間内にAPIへ送信できるリクエスト数を定めます。サービスプロバイダーは通常、APIベースのサービスやリソースを公平に利用できるよう、この方法を適用します。

サーバーでレート制限を適用するには、まずサービスの保護やプランごとのクォータ設定など、必要な理由を把握する必要があります。次に、状況に最適な制限キーを特定して選びます。そして、選択したキーに基づいてAPIの使用状況を追跡する仕組みを導入します。

現在、python-socketio専用のレート制限パッケージはありませんが、少し手を加えれば実装できる方法があります。

WebSocketサーバーを保護するため、ペイロードサイズを制限する

WebSocketで巨大なペイロードを送信すると、パフォーマンスが損なわれ、最終的にはソケットサーバーがクラッシュする可能性が高くなります。

WebSocketの速度低下やクラッシュを防ぐには、WebSocket接続で送信するメッセージの最大ペイロードサイズを制限するとよいでしょう。これにより、過大なメッセージ（Zip爆弾など）の送信によるサーバーのクラッシュを防げます。

また、メッセージサイズを調整すると、WebSocket接続のレイテンシーが大幅に低下し、転送速度も大きく向上します。WebSocketプロトコルではフレームサイズに上限があり、メッセージに必要な圧縮の程度を判断するのに役立ちます。メッセージは圧縮されます。メッセージストリームの圧縮にはメモリやCPUリソースを消費しますが、ネットワークトラフィックを大幅に減らせるため、多くの場合その価値があります。

安全なソケット通信にはTLSを使用する

Webサイトをインターネットに公開するには、TLS/SSLの使用が不可欠です。使用しない場合、クライアントとサーバー間で送受信される機密情報が、中間者攻撃によって簡単に盗まれる可能性があります。通信を保護するには、安全でないhttp://ではなく、https://のような安全なプロトコルを使用してください。

WebSocketでは、安全でないws://ではなく、wss://（TLSで暗号化されたWebSocket）プロトコルを使用してください。ws://を使用すると、接続が第三者による干渉を受けやすくなります。一方、wss://プロトコルでは、WebSocketで送信されるすべてのデータが暗号化されます。

この暗号化により、第三者がWebSocket経由で送信されるメッセージを読み取ったり改ざんしたりできなくなり、機密情報が保護されます。接続が安全であれば、ほかの種類の攻撃も防げます。

さらに、サーバーとのWebSocket接続を確立する前に、リクエスト元のWebサイトもhttps://を使用していることを確認してください。使用していない場合、悪意のある第三者がリクエストを簡単に改ざんできてしまいます。

TLSは、WebSocketの有無にかかわらず、あらゆるWebサイトに必要です。幸い、多くのホスティングサービスでは、アプリケーションのデプロイ時に無料でTLSを提供しています。Pythonアプリ向けの人気の高いホスティングプラットフォームには、Google Cloud、AWS、Azure、Heroku、Fly.ioなどがあります。

WebSocketのまとめ

このチュートリアルでは、python-socketioを使って基本的なWebSocketサーバーを作成し、JavaScriptクライアントに接続しました。また、WebSocketの仕組みを学び、安全に使用するためのセキュリティ上の注意点についても解説しました。

ほかの多くのWebプロトコルと同様に、WebSocketでも通信を適切に保護するために、TLS、CORS、レート制限などの対策が必要です。これらはいずれもWebSocket固有のものではありませんが、不要なセキュリティ問題を避けるために理解しておくことが重要です。

無料のオンラインコードチェッカーツール

次のコミットの前に、コードを安全に保護しましょう。

コードをチェック

カテゴリー:
コードセキュリティアプリケーションセキュリティエンジニアリング

続きを読む

blog feature snyk container custom base recommendations
Blog

Evo Agentic AppSecのご紹介：AI開発のスピードに対応するセキュリティ

AI駆動型開発のスピードに、従来のセキュリティワークフローは追いついていません。SnykのAgentic AppSecは、セキュリティエージェントが脆弱性を継続的に検出、修正、検証し、人が基準を定められるようにします。

blog feature ai violet
Blog

VulnBump：AIエージェントはビルドを壊さずに脆弱な依存関係を修正できるか？

VulnBumpは、実際の依存関係アップグレード307件を対象にAIエージェントをテストします。脆弱性が解消され、ビルドとテストが成功した場合にのみ、修正としてカウントされます。

Blog

フロンティアモデルは脆弱性を発見した。攻撃者だけがエクスプロイトチェーンを見つけた。

静的解析で欠陥は見つかりましたが、ライブ攻撃テストで侵害につながる連鎖を実証できたのは唯一でした。Evo COS、Claude Security、Claude Code Securityを比較します。