Skip to content

Instantly share code, notes, and snippets.

@flano-yuki flano-yuki/quic_transport.py Secret
Created Jan 13, 2020

Embed
What would you like to do?
WebTransport over QUIC mock client using aioquic
import argparse
import asyncio
import json
import logging
import ssl
from typing import Optional, cast
from aioquic.asyncio.client import connect
from aioquic.asyncio.protocol import QuicConnectionProtocol
from aioquic.quic.configuration import QuicConfiguration
from aioquic.quic.events import QuicEvent, StreamDataReceived
from aioquic.quic.events import DatagramFrameReceived, QuicEvent
from aioquic.quic.logger import QuicLogger
logger = logging.getLogger("client")
class QuicTransportClient(QuicConnectionProtocol):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._ack_waiter: Optional[asyncio.Future[None]] = None
async def quack(self) -> None:
assert self._ack_waiter is None, "Only one quack at a time."
# client indication
logger.info("Sent Client Indication")
b = b'\x00\x00' + b'\x00\x17' + b'https://example.com:443'
b += b'\x00\x01' + b'\x00\x05' + b'/echo'
self._quic.send_stream_data(2, b, True)
self._quic.send_datagram_frame(b)
self.transmit()
# say hello
logger.info("Say Hello")
b = b"hello world"
self._quic.send_stream_data(6, b, True)
self._quic.send_datagram_frame(b)
waiter = self._loop.create_future()
self._ack_waiter = waiter
self.transmit()
return await asyncio.shield(waiter)
def quic_event_received(self, event: QuicEvent) -> None:
if self._ack_waiter is not None:
if isinstance(event, DatagramFrameReceived) or isinstance(event,StreamDataReceived):
logger.info("Echo is received: ")
logger.info(event.data)
waiter = self._ack_waiter
self._ack_waiter = None
waiter.set_result(None)
async def run(configuration: QuicConfiguration, host: str, port: int) -> None:
async with connect(
host, port, configuration=configuration, create_protocol=QuicTransportClient
) as client:
client = cast(QuicTransportClient, client)
await client.quack()
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="SiDUCK client")
parser.add_argument(
"host", type=str, help="The remote peer's host name or IP address"
)
parser.add_argument("port", type=int, help="The remote peer's port number")
parser.add_argument(
"-k",
"--insecure",
action="store_true",
help="do not validate server certificate",
)
parser.add_argument(
"-q", "--quic-log", type=str, help="log QUIC events to a file in QLOG format"
)
parser.add_argument(
"-l",
"--secrets-log",
type=str,
help="log secrets to a file, for use with Wireshark",
)
parser.add_argument(
"-v", "--verbose", action="store_true", help="increase logging verbosity"
)
args = parser.parse_args()
logging.basicConfig(
format="%(asctime)s %(levelname)s %(name)s %(message)s",
level=logging.DEBUG if args.verbose else logging.INFO,
)
configuration = QuicConfiguration(
alpn_protocols=["wq-vvv-01"], is_client=True, max_datagram_frame_size=65536
)
if args.insecure:
configuration.verify_mode = ssl.CERT_NONE
if args.quic_log:
configuration.quic_logger = QuicLogger()
if args.secrets_log:
configuration.secrets_log_file = open(args.secrets_log, "a")
loop = asyncio.get_event_loop()
try:
loop.run_until_complete(
run(configuration=configuration, host=args.host, port=args.port)
)
finally:
if configuration.quic_logger is not None:
with open(args.quic_log, "w") as logger_fp:
json.dump(configuration.quic_logger.to_dict(), logger_fp, indent=4)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.