Demonstrates wrapping of sockets for TLS using contexts which are not Curio's own.

curio_tls.py

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.py

# 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_client.py

# 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