【Python】asyncio を使って TLS 対応の HTTP/1 サーバーをつくる

次のコマンドでサーバーを起動させて https//localhost:8000 に curl かブラウザーでアクセスする

python server.py

curl のコマンドは次のとおり

curl -v https://localhost:8000

TLS 証明書と秘密鍵は mkcert で生成した

mkcert localhost

server.py

import asyncio
import ssl

# https://docs.python.org/ja/3/library/asyncio-stream.html#tcp-echo-server-using-streams
# https://stackoverflow.com/a/76933377/531320


async def handle_echo(reader, writer):
    data = (await reader.read(1024)).decode()
    addr = writer.get_extra_info('peername')

    print(f"Received {data!r} from {addr!r}")

    msg =(
      'HTTP/1.1 200 OK\r\n' \
      'Content-Type: text/html; charset=UTF-8\r\n' \
      'Content-Length: 13\r\n' \
      'Connection: Close\r\n' \
      '\r\n' \
      'Hello World\r\n' \
    ).encode()

    writer.write(msg)
    await writer.drain()

    writer.close()
    await writer.wait_closed()

async def main():
    ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    ctx.load_cert_chain('localhost.pem', 'localhost-key.pem')
    ctx.set_alpn_protocols(['http/1.1'])

    server = await asyncio.start_server(
        handle_echo, '127.0.0.1', 8000, ssl=ctx
    )

    addrs = ', '.join(str(sock.getsockname()) for sock in server.sockets)
    print(f'Serving on {addrs}')

    async with server:
        await server.serve_forever()

asyncio.run(main())

writer.start_tls を使って TLS に対応させることもできるが、Chrome でアクセスするといろいろなエラーメッセージを吐き出した。サーバーそのものは止まらず動いた。

async def handle_echo(reader, writer):
    ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    ctx.load_cert_chain('localhost.pem', 'localhost-key.pem')
    ctx.set_alpn_protocols(['http/1.1'])

    await writer.start_tls(ctx)

    data = (await reader.read(1024)).decode()
    addr = writer.get_extra_info('peername')

サーバーを起動させていて出力されたエラーメッセージは次のとおり

Unhandled exception in client_connected_cb
transport: <_SelectorSocketTransport closed fd=7>
Traceback (most recent call last):
  File "/home/masakielastic/asyncio-project/server.py", line 9, in handle_echo
    await writer.start_tls(ctx)
  File "/home/masakielastic/.local/share/mise/installs/python/3.13.0b3/lib/python3.13/asyncio/streams.py", line 396, in start_tls
    new_transport = await self._loop.start_tls(  # type: ignore
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    ...<3 lines>...
        ssl_shutdown_timeout=ssl_shutdown_timeout)
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/masakielastic/.local/share/mise/installs/python/3.13.0b3/lib/python3.13/asyncio/base_events.py", line 1338, in start_tls
    await waiter
  File "/home/masakielastic/.local/share/mise/installs/python/3.13.0b3/lib/python3.13/asyncio/sslproto.py", line 578, in _on_handshake_complete
    raise handshake_exc
  File "/home/masakielastic/.local/share/mise/installs/python/3.13.0b3/lib/python3.13/asyncio/sslproto.py", line 560, in _do_handshake
    self._sslobj.do_handshake()
    ~~~~~~~~~~~~~~~~~~~~~~~~~^^
  File "/home/masakielastic/.local/share/mise/installs/python/3.13.0b3/lib/python3.13/ssl.py", line 952, in do_handshake
    self._sslobj.do_handshake()
    ~~~~~~~~~~~~~~~~~~~~~~~~~^^

ほかに asyncio.Protocol を継承するクラスを使ったやり方もある。

server.py

import asyncio
import ssl

# https://docs.python.org/ja/3/library/asyncio-protocol.html#tcp-echo-server
# https://github.com/python/cpython/issues/109534

class EchoServerProtocol(asyncio.Protocol):

    def connection_made(self, transport):
        peername = transport.get_extra_info('peername')
        print('Connection from {}'.format(peername))
        self.transport = transport

    def data_received(self, data):
        msg =(
          'HTTP/1.1 200 OK\r\n' \
          'Content-Type: text/html; charset=UTF-8\r\n' \
          'Content-Length: 13\r\n' \
          'Connection: Close\r\n' \
          '\r\n' \
          'Hello World\r\n'
        ).encode()

        self.transport.write(msg)
        self.transport.close()


async def main():
    ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    ctx.load_cert_chain('localhost.pem', 'localhost-key.pem')
    ctx.set_alpn_protocols(['http/1.1'])

    loop = asyncio.get_running_loop()

    server = await loop.create_server(
        lambda: EchoServerProtocol(),
        '127.0.0.1', 8000, ssl=ctx)

    async with server:
        await server.serve_forever()


asyncio.run(main())