diff --git a/tests/socket_ssl.py b/tests/socket_ssl.py new file mode 100644 index 0000000..649fec4 --- /dev/null +++ b/tests/socket_ssl.py @@ -0,0 +1,75 @@ +from debugger import Debugger +import aiosched +import socket as sock +import ssl +import sys +import time + +_print = print + + +def print(*args, **kwargs): + sys.stdout.write(f"[{time.strftime('%H:%M:%S')}] ") + _print(*args, **kwargs) + + +async def test(host: str, port: int, bufsize: int = 4096): + socket = aiosched.socket.wrap_socket( + ssl.create_default_context().wrap_socket( + sock=sock.socket(), + # Note: do_handshake_on_connect MUST + # be set to False on the synchronous socket! + # The library handles the TLS handshake asynchronously + # and making the SSL library handle it blocks + # the entire event loop. To perform the TLS + # handshake upon connection, set this parameter + # in the AsyncSocket class instead + do_handshake_on_connect=False, + server_hostname=host, + ) + ) + print(f"Attempting a connection to {host}:{port}") + await socket.connect((host, port)) + print("Connected") + async with aiosched.with_context(): + async with socket: + # Closes the socket automatically + print("Entered socket context manager, sending request data") + await socket.send_all( + f"GET / HTTP/1.1\r\nHost: {host}\r\nUser-Agent: owo\r\nAccept: text/html\r\nConnection: keep-alive\r\nAccept: */*\r\n\r\n".encode() + ) + print("Data sent") + buffer = b"" + while not buffer.endswith(b"\r\n\r\n"): + print(f"Requesting up to {bufsize} bytes (current response size: {len(buffer)})") + data = await socket.receive(bufsize) + if data: + print(f"Received {len(data)} bytes") + buffer += data + else: + print("Received empty stream, closing connection") + break + if buffer: + data = buffer.decode().split("\r\n") + print(f"HTTP Response below:") + _print(f"Response: {data[0]}") + _print("Headers:") + content = False + for i, element in enumerate(data): + if i == 0: + continue + else: + if not element.strip() and not content: + # This only works because google sends a newline + # before the content + sys.stdout.write("\nContent:") + content = True + if not content: + _print(f"\t{element}") + else: + for line in element.split("\n"): + _print(f"\t{line}") + _print("Done!") + + +aiosched.run(test, "debian.org", 443, 256, debugger=())