trio h2

trio_h2.py

import sys
import ssl
import json
import collections

import trio
import h2.config
import h2.connection
import h2.events

ReceivedData = collections.namedtuple("ReceivedData", ("headers", "data"))


class H2EchoServer:
    def __init__(self):
        config = h2.config.H2Configuration(client_side=False, header_encoding="utf-8")
        self.connection = h2.connection.H2Connection(config=config)
        self.received_data = {}
        self.flow_control_events = {}
        self.write_lock = trio.Lock()
        self.stream = None

    async def write_all_pending_data(self):
        async with self.write_lock:
            await self.stream.send_all(self.connection.data_to_send())

    def request_received(self, event):
        self.received_data[event.stream_id] = ReceivedData(event.headers, bytearray())

    def data_received(self, event):
        try:
            self.received_data[event.stream_id].data.extend(event.data)
        except KeyError:
            self.connection.reset_stream(event.stream_id,
                                         h2.errors.ErrorCodes.PROTOCOL_ERROR)
        else:
            self.connection.acknowledge_received_data(event.flow_controlled_length,
                                                      event.stream_id)

    async def reply_echo(self, stream_id):
        self.flow_control_events[stream_id] = trio.Event()
        response_body = json.dumps({
            "headers": collections.OrderedDict(self.received_data[stream_id].headers),
            "body": self.received_data[stream_id].data.decode("utf-8")
        }, indent=4).encode("utf-8")

        response_headers = (
            (":status", "200"),
            ("content-type", "application/json"),
            ("content-length", str(len(response_body))),
            ("server", "python-trio-h2"),
        )
        self.connection.send_headers(stream_id, response_headers)

        ptr = 0
        while ptr < len(response_body):
            while self.connection.local_flow_control_window(stream_id) == 0:
                await self.flow_control_events[stream_id]
            chunk_size = min(self.connection.max_outbound_frame_size,
                             self.connection.local_flow_control_window(stream_id))
            self.connection.send_data(stream_id, response_body[ptr:ptr+chunk_size])
            await self.write_all_pending_data()
            ptr += chunk_size
        self.connection.end_stream(stream_id)
        await self.write_all_pending_data()

    def window_updated(self, event):
        if event.stream_id == 0:
            for event in self.flow_control_events.values():
                event.set()
        else:
            try:
                self.flow_control_events[event.stream_id].set()
            except KeyError:
                self.connection.reset_stream(event.stream_id,
                                             h2.errors.ErrorCodes.PROTOCOL_ERROR)

    async def __call__(self, server_stream):
        self.stream = server_stream
        try:
            self.connection.initiate_connection()
            await self.write_all_pending_data()

            try:
                async with trio.open_nursery() as nursery:
                    while True:
                        data = await server_stream.receive_some(65536)
                        if not data:
                            return
                        events = self.connection.receive_data(data)
                        for event in events:
                            if isinstance(event, h2.events.RequestReceived):
                                self.request_received(event)
                            elif isinstance(event, h2.events.DataReceived):
                                self.data_received(event)
                            elif isinstance(event, h2.events.StreamEnded):
                                nursery.start_soon(self.reply_echo, event.stream_id)
                            elif isinstance(event, h2.events.WindowUpdated):
                                self.window_updated(event)
                            elif isinstance(event, h2.events.ConnectionTerminated):
                                return
                            await self.write_all_pending_data()
            finally:
                await self.write_all_pending_data()
        except:
            print("Got exception: {!r}".format(sys.exc_info()))


async def main(port):
    ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    ssl_context.options |= (
            ssl.OP_NO_TLSv1 | ssl.OP_NO_TLSv1_1 | ssl.OP_NO_COMPRESSION
    )
    ssl_context.set_ciphers("ECDHE+AESGCM")
    ssl_context.load_cert_chain(certfile="cert.crt", keyfile="cert.key")
    ssl_context.set_alpn_protocols(["h2"])

    await trio.serve_ssl_over_tcp(lambda stream: H2EchoServer()(stream),
                                  port, ssl_context)


if __name__ == "__main__":
    port = int(sys.argv[1])
    print("Try: $ curl --tlsv1.2 --http2 -k https://localhost:{}/path -d'data'"
          .format(port))
    print("Or open a browser to https://localhost:{}/ and accept all the warnings"
          .format(port))
    trio.run(main, port)

cert.crt

-----BEGIN CERTIFICATE-----
MIIDhTCCAm2gAwIBAgIJAOrxh0dOYJLdMA0GCSqGSIb3DQEBCwUAMFkxCzAJBgNV
BAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEwHwYDVQQKDBhJbnRlcm5ldCBX
aWRnaXRzIFB0eSBMdGQxEjAQBgNVBAMMCWxvY2FsaG9zdDAeFw0xNTA5MTkxNDE2
NDRaFw0xNTEwMTkxNDE2NDRaMFkxCzAJBgNVBAYTAkFVMRMwEQYDVQQIDApTb21l
LVN0YXRlMSEwHwYDVQQKDBhJbnRlcm5ldCBXaWRnaXRzIFB0eSBMdGQxEjAQBgNV
BAMMCWxvY2FsaG9zdDCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAMqt
A1iu8EN00FU0eBcBGlLVmNEgV7Jkbukra+kwS8j/U2y50QPGJc/FiIVDfuBqk5dL
ACTNc6A/FQcXvWmOc5ixmC3QKKasMpuofqKz0V9C6irZdYXZ9rcsW0gHQIr989yd
R+N1VbIlEVW/T9FJL3B2UD9GVIkUELzm47CSOWZvAxQUlsx8CUNuUCWqyZJoqTFN
j0LeJDOWGCsug1Pkj0Q1x+jMVL6l6Zf6vMkLNOMsOsWsxUk+0L3tl/OzcTgUOCsw
UzY59RIi6Rudrp0oaU8NuHr91yiSqPbKFlX10M9KwEEdnIpcxhND3dacrDycj3ux
eWlqKync2vOFUkhwiaMCAwEAAaNQME4wHQYDVR0OBBYEFA0PN+PGoofZ+QIys2Jy
1Zz94vBOMB8GA1UdIwQYMBaAFA0PN+PGoofZ+QIys2Jy1Zz94vBOMAwGA1UdEwQF
MAMBAf8wDQYJKoZIhvcNAQELBQADggEBAEplethBoPpcP3EbR5Rz6snDDIcbtAJu
Ngd0YZppGT+P0DYnPJva4vRG3bb84ZMSuppz5j67qD6DdWte8UXhK8BzWiHzwmQE
QmbKyzzTMKQgTNFntpx5cgsSvTtrHpNYoMHzHOmyAOboNeM0DWiRXsYLkWTitLTN
qbOpstwPubExbT9lPjLclntShT/lCupt+zsbnrR9YiqlYFY/fDzfAybZhrD5GMBY
XdMPItwAc/sWvH31yztarjkLmld76AGCcO5r8cSR/cX98SicyfjOBbSco8GkjYNY
582gTPkKGYpStuN7GNT5tZmxvMq935HRa2XZvlAIe8ufp8EHVoYiF3c=
-----END CERTIFICATE-----

cert.key

-----BEGIN RSA PRIVATE KEY-----
MIIEpAIBAAKCAQEAyq0DWK7wQ3TQVTR4FwEaUtWY0SBXsmRu6Str6TBLyP9TbLnR
A8Ylz8WIhUN+4GqTl0sAJM1zoD8VBxe9aY5zmLGYLdAopqwym6h+orPRX0LqKtl1
hdn2tyxbSAdAiv3z3J1H43VVsiURVb9P0UkvcHZQP0ZUiRQQvObjsJI5Zm8DFBSW
zHwJQ25QJarJkmipMU2PQt4kM5YYKy6DU+SPRDXH6MxUvqXpl/q8yQs04yw6xazF
ST7Qve2X87NxOBQ4KzBTNjn1EiLpG52unShpTw24ev3XKJKo9soWVfXQz0rAQR2c
ilzGE0Pd1pysPJyPe7F5aWorKdza84VSSHCJowIDAQABAoIBACp+nh4BB/VMz8Wd
q7Q/EfLeQB1Q57JKpoqTBRwueSVai3ZXe4CMEi9/HkG6xiZtkiZ9njkZLq4hq9oB
2z//kzMnwV2RsIRJxI6ohGy+wR51HD4BvEdlTPpY/Yabpqe92VyfSYxidKZWaU0O
QMED1EODOw4ZQ+4928iPrJu//PMB4e7TFao0b9Fk/XLWtu5/tQZz9jsrlTi1zthh
7n+oaGNhfTeIJJL4jrhTrKW1CLHXATtr9SJlfZ3wbMxQVeyj2wUlP1V0M6kBuhNj
tbGbMpixD5iCNJ49Cm2PHg+wBOfS3ADGIpi3PcGw5mb8nB3N9eGBRPhLShAlq5Hi
Lv4tyykCgYEA8u3b3xJ04pxWYN25ou/Sc8xzgDCK4XvDNdHVTuZDjLVA+VTVPzql
lw7VvJArsx47MSPvsaX/+4hQXYtfnR7yJpx6QagvQ+z4ludnIZYrQwdUmb9pFL1s
8UNj+3j9QFRPenIiIQ8qxxNIQ9w2HsVQ8scvc9CjYop/YYAPaQyHaL8CgYEA1ZSz
CR4NcpfgRSILdhb1dLcyw5Qus1VOSAx3DYkhDkMiB8XZwgMdJjwehJo9yaqRCLE8
Sw5znMnkfoZpu7+skrjK0FqmMpXMH9gIszHvFG8wSw/6+2HIWS19/wOu8dh95LuC
0zurMk8rFqxgWMWF20afhgYrUz42cvUTo10FVB0CgYEAt7mW6W3PArfUSCxIwmb4
VmXREKkl0ATHDYQl/Cb//YHzot467TgQll883QB4XF5HzBFurX9rSzO7/BN1e6I0
52i+ubtWC9xD4fUetXMaQvZfUGxIL8xXgVxDWKQXfLiG54c8Mp6C7s6xf8kjEUCP
yR1F0SSA/Pzb+8RbY0p7eocCgYA+1rs+SXtHZev0KyoYGnUpW+Uxqd17ofOgOxqj
/t6c5Z+TjeCdtnDTGQkZlo/rT6XQWuUUaDIXxUbW+xEMzj4mBPyXBLS1WWFvVQ5q
OpzO9E/PJeqAH6rkof/aEelc+oc/zvOU1o9uA+D3kMvgEm1psIOq2RHSMhGvDPA0
NmAk+QKBgQCwd1681GagdIYSZUCBecnLtevXmIsJyDW2yR1NNcIe/ukcVQREMDvy
5DDkhnGDgnV1D5gYcXb34g9vYvbfTnBMl/JXmMAAG1kIS+3pvHyN6f1poVe3yJV1
yHVuvymnJxKnyaV0L3ntepVvV0vVNIkA3oauoUTLto6txBI+b/ImDA==
-----END RSA PRIVATE KEY-----