Skip to content

Instantly share code, notes, and snippets.

@carlopires
Forked from vsajip/curio_tls.py
Created June 5, 2017 19:19
Show Gist options
  • Save carlopires/b9d0e0160132381b3948c4e1892d0e92 to your computer and use it in GitHub Desktop.
Save carlopires/b9d0e0160132381b3948c4e1892d0e92 to your computer and use it in GitHub Desktop.
Demonstrates wrapping of sockets for TLS using contexts which are not Curio's own.
from wsgiref.handlers import format_date_time
import curio
from curio import network
from curio import socket
from curio.io import Socket
import h11
import tls
async def client(host, port):
context = tls.Context(verify_cert=False, verify_name=False)
context.host = host
context.port = port
sock = await network.open_connection(host, port, ssl=context, server_hostname=host)
conn = h11.Connection(our_role=h11.CLIENT)
req = h11.Request(method='GET', target='/', headers=[('Host', host)])
data = conn.send(req)
await sock.sendall(data)
resp = await sock.recv(4096)
return resp
async def handler(client, addr):
data = await client.recv(1000)
assert data == b'GET / HTTP/1.1\r\nhost: localhost\r\n\r\n'
conn = h11.Connection(our_role=h11.SERVER)
body = b'Hello, world!'
headers = [
('Date', format_date_time(None).encode('ascii')),
('Server', 'Curio test'),
('Content-Type', 'text/plain'),
('Content-Length', str(len(body))),
]
resp = h11.Response(status_code=200, headers=headers)
data = conn.send(resp)
data += conn.send(h11.Data(data=body))
data += conn.send(h11.EndOfMessage())
await client.send(data)
def server(host, port):
certpath = 'servercert.pem'
keypath = 'serverkey.pem'
capath = 'cacert.pem'
context = tls.Context(server=True, capath=capath, certpath=certpath,
keypath=keypath)
context.host = host
# set up a listening socket
host = 'localhost'
port = 0
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
sock.bind((host, port))
sock.listen(100)
addr = sock.getsockname()
context.port = addr[-1]
return addr, network._run_server(sock, handler, context)
except Exception:
sock._socket.close()
raise
async def main():
addr, server_coro = server('localhost', 0)
server_task = await curio.spawn(server_coro)
resp = await client('localhost', addr[-1])
await server_task.cancel()
conn = h11.Connection(our_role=h11.CLIENT)
conn.receive_data(resp)
e = conn.next_event() # headers
e = conn.next_event() # body
print('Received: %s' % bytes(e.data))
if __name__ == '__main__':
try:
curio.run(main())
except KeyboardInterrupt:
pass
# ssl_echo
#
# An example of a simple SSL echo server. Use ssl_echo_client.py to test.
import os
import ssl
import curio
from curio import network
KEYFILE = 'serverkey.pem'
CERTFILE = 'servercert.pem'
async def handle(client, addr):
print('Connection from', addr)
async with client:
while True:
data = await client.recv(1000)
if not data:
break
await client.send(data)
print('Connection closed')
if __name__ == '__main__':
ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
ssl_context.load_cert_chain(certfile=CERTFILE, keyfile=KEYFILE)
try:
curio.run(network.tcp_server('', 10000, handle, ssl=ssl_context))
except KeyboardInterrupt:
pass
# ssl_echo
#
# An example of a simple SSL echo client. Use ssl_echo.py for the server.
import ssl
import curio
from curio import network
async def main(host, port):
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE
sock = await network.open_connection(
host, port, ssl=True, server_hostname=None)
for i in range(1000):
msg = ('Message %d' % i).encode('ascii')
print(msg)
await sock.sendall(msg)
resp = await sock.recv(1000)
assert msg == resp
await sock.close()
if __name__ == '__main__':
try:
curio.run(main('localhost', 10000))
except KeyboardInterrupt:
pass
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment