QUICで純粋なechoサーバーを実装する(Python)

QUICとは

QUICとはTCPの接続をUDPで実現し、より安全で早いものにしようという考えから生まれたUDPの通信プロトコルです。2021年にIETFでRFC 9000として正式に標準化され現在ではhttp/3やSSH over QUICなどで知られています。

本稿ではそんなモダンなプロトコルを用いた簡易的なメッセージの送受信のテストを行います。

実装

今回はQUICのPython版の実装であるaioquicというライブラリを用いた実装を行いました。最初はChatGPTの出力をもとに、最後の修正は公式ドキュメントと睨めっこしながら書きました。

Server.py

# server.py
import asyncio
from aioquic.asyncio import serve
from aioquic.quic.configuration import QuicConfiguration

async def handle_stream(reader, writer):
    print("New stream opened.")
    data = await reader.read(1024)
    while not reader.at_eof():
        print(f"Received: {data.decode()}")
        writer.write_eof()
        data = await reader.read(1024)
    print("Stream closed.")

async def run_server(host, port):
    configuration = QuicConfiguration(is_client=False)

    configuration.load_cert_chain(certfile="cert.pem", keyfile="key.pem")

    def handle_stream_awaited(reader, writer):
        asyncio.create_task(handle_stream(reader, writer))
    await serve(host, port, configuration=configuration, stream_handler=handle_stream_awaited)

def main():
    loop = asyncio.get_event_loop()
    loop.run_until_complete(run_server('localhost', 4433))
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        pass
    finally:
        loop.close()

if __name__ == "__main__":
    main()
    

client.py

# client.py
import asyncio
from aioquic.asyncio import connect
from aioquic.quic.configuration import QuicConfiguration

async def run_client(host, port, ca_cert_path):
    configuration = QuicConfiguration(is_client=True)
    configuration.load_verify_locations(cafile=ca_cert_path)

    async with connect(host, port, configuration=configuration) as protocol:
        print("Connecting to server...",end="")
        await protocol.wait_connected()
        reader, writer = await protocol.create_stream()
        print("Connected")

        while not reader.at_eof():
            text = input("Message: ")
            writer.write((text).encode())
            print("Sending to server...",end="")
            await reader.read(1024)

        print("done")
        print("Sending eof...",end="")
        writer.write_eof()
        protocol.close()
        await protocol.wait_closed()
        print("done")

def main():
    asyncio.run(run_client('localhost', 4433, 'ca_cert.pem'))

if __name__ == "__main__":
    main()

プログラムの概要は以下です。

1. サーバーは待機状態になる
2. クライアントから接続し、streamを作成する
3. ユーザーからの入力をクライアントで受け付ける
4. サーバーはメッセージの受信を待機、クライアントからサーバーにメッセージを送信する
5. サーバーはメッセージを受け取ったらeofを送信する、クラインとはeofを受け取るまで待機
6. クライアントがeofを受け取ったらstreamを終了

オレオレ証明書の発行にはmkcertを使います。これは簡単に信頼されているオレオレ証明書を発行するためのツールです。

shell

mkcert -install
Created a new local CA 💥
Sudo password:
The local CA is now installed in the system trust store! ⚡️

shell

mkcert localhost 127.0.0.1 ::1

Created a new certificate valid for the following names 📜
 - "localhost"
 - "127.0.0.1"
 - "::1"

The certificate is at "./localhost+2.pem" and the key at "./localhost+2-key.pem" ✅

It will expire on 11 May 2026 🗓

それぞれのファイル名をプログラムのcert.pem、key.pemに変更してプログラムを実行します。また、マシンからルート証明書を保存してきて、そちらもca_cert.pemとして保存しておきました。

結果

プログラムの実行結果です。

server side

python3 server.py
New stream opened.
Received: this is a message
Stream closed.

client side

python3  client.py
Connecting to server...Connected
Message: this is a message
Sending to server...done
Sending eof...done

またwiresharkでは以下のようなパケットが観測されました。

実装は以下のGithubで公開しています。

まとめ

本稿では純粋なQUICでechoサーバーを作ることができました。証明書を利用して通信することがデフォルトになっているのはとても安心ですね。世間一般ではHTTP/3に乗っけた上での通信を行ったりすることがデフォルトだと思うので、純粋なQUICを使ってみたい人の参考になればと思います。

また、関連してこの参考になるような記事・プログラムは確認する範囲では公開されていませんでした。もし類似のことをやりたい場合はしっかりとライブラリの公式ドキュメントを読むことをお勧めするのと、部分的に実装がまだであったりすることがあるのでご注意ください。