Secure Communication With Python SSL Certificate and Asyncio

The documentation for using ssl certificates with asyncio streams is distributed over a number of packages, and can be confusing at first. In this post I’ll take a simple network service with no encryption to an HTTP service with both server and client certificate security.

You can find the source code for this post here.

To start we need a network service. The standard library documentation provides a simple implementation of an echo service.

Step 1 — An Echo Service

Client

Here is the client.

import asyncio

async def tcp_echo_client(message):
    reader, writer = await asyncio.open_connection(
        '127.0.0.1',
        8888
    )

    print(f'Send: {message!r}')
    writer.write(message.encode())

    data = await reader.read(100)
    print(f'Received: {data.decode()!r}')

    print('Close the connection')
    writer.close()

asyncio.run(tcp_echo_client('Hello World!'))

The client connects to the server, sends a message, reads the returned message, prints it out and exits.

Server

Here is the server.

import asyncio

async def handle_echo(reader, writer):
    data = await reader.read(100)
    message = data.decode()
    addr = writer.get_extra_info('peername')

    print(f"Received {message!r} from {addr!r}")

    print(f"Send: {message!r}")
    writer.write(data)
    await writer.drain()

    print("Close the connection")
    writer.close()

async def main():
    server = await asyncio.start_server(
        handle_echo,
        '127.0.0.1',
        8888
    )

    addrs = ', '.join(
        str(sock.getsockname())
        for sock in server.sockets
    )
    print(f'Serving on {addrs}')

    async with server:
        await server.serve_forever()

asyncio.run(main())

The server waits for a connection. When a connection is established it reads the sent message, and writes the same message back. This is known as an echo server.

Step 2 — Server Certificates

First we need to get a server certificate. If this is for internal use we can generate a self signed certificate. If we need the certificate to be trusted by clients external to our domain we might choose to buy one from a certificate provider, or use LetsEncrypt to get a free one, (or share our self-signed certificates with the external clients to be added to their “trust stores”).

I’m going to use self-signed certificates that have been generated using a technique from a previous post. Whichever way you do this three or more items will be produced:

1. A server certificate,
2. A server key,
3. One or many certificates for the certificate authority (or CA certificates).

The CA certificate can cause a little confusion. The CA certificate is used to issue new certificates. If you’ve purchased a certificate or are using LetsEncrypt you may not appear to have one of these. This is because it will be included in the “trust store” of the operating system. On Unix this is often just a file (e.g. /etc/ssl/certs/ca-certificates.crt) which is often called a “bundle”. Mac and Windows provide a set of tools for storing the certificates.

In my self-signed certificate example, I created two CA’s: an intermediate and a root. This is a common security measure. Certificates are issued from the intermediate CA. If the key to one of these certificates was stolen, the intermediate CA can be removed from the trust chain, a new one created, and new certificates issued.

In order to create our own bundle from the self signed certificates we can do the following.

cat intermediate-ca.pem root-ca.pem > cacerts.pem

This just makes a single file of the two files. You could use an editor like notepad to do the same. Note the order where the intermediate is first, and the root last. This is the “trust chain”

Client

Now we can write some code! Here’s the client.

import asyncio
from os.path import expanduser
import socket
import ssl

async def tcp_echo_client(message):

    # We need a fully qualified domain name for the server.
    host = socket.gethostname()

    # We need an SSL context.
    context = ssl.SSLContext(ssl.PROTOCOL_TLS)
    context.load_verify_locations(
        expanduser("~/.keys/cacerts.pem")
    )
    context.verify_mode = ssl.CERT_REQUIRED
    context.check_hostname = True

    reader, writer = await asyncio.open_connection(
        host, # use the real host name
        8888,
        ssl=context # pass in the context.
    )

    # We will get a certificate from the server.
    peercert = writer.get_extra_info('peercert')
    print(f'Peer cert: {peercert!r}')

    print(f'Send: {message!r}')
    writer.write(message.encode())

    data = await reader.read(100)
    print(f'Received: {data.decode()!r}')

    print('Close the connection')
    writer.close()

asyncio.run(tcp_echo_client('Hello World!'))

Most of the extra stuff is at the top. First we need to get the real name of the server. The name is embedded in the server’s certificate, and it must match the name we use to connect. It’s usually something like “app.example.com”.

Next we make the SSL context specifying the TLS protocol.

Then we load the CA bundle for the self signed certificates. Note I’ve put all mine in a folder underneath my home folder called “.keys”. For a trusted certificate we could read the OS bundle or just call load_default_certs, which knows where they are. Regardless of the method, the client needs to know the CA’s it can trust in order to verify the certificate.

A couple of properties get set on the context. The verify_modeis set to CERT_REQUIRED, as we want to get to check the certificate, and we also set check_hostnameto true, to ensure the host we’re talking to really owns the certificate.

Finally we pass in the ssl context when opening the connection. The returned write has a method get_extra_infoto get information from the connection. We should see the certificate from the server being printed.

Server

Here is the server.

import asyncio
from os.path import expanduser
import socket
import ssl

async def handle_echo(reader, writer):

    data = await reader.read(100)
    message = data.decode()
    addr = writer.get_extra_info('peername')
    # This will be None as the child has no certificate.
    peercert = writer.get_extra_info('peercert')
    print(f"Received {message!r} from {addr!r} with cert {peercert!r}")

    print(f"Send: {message!r}")
    writer.write(data)
    await writer.drain()

    print("Close the connection")
    writer.close()

async def main():

    # We could have "used 0.0.0.0"
    host = socket.gethostname()

    # We need an SSL context.
    context = ssl.SSLContext(ssl.PROTOCOL_TLS)
    context.load_cert_chain(
        expanduser("~/.keys/server.crt"),
        expanduser("~/.keys/server.key")
    )

    server = await asyncio.start_server(
        handle_echo,
        host,
        8888,
        ssl=context # pass in the SSL context.
    )

    addrs = ', '.join(
        str(sock.getsockname())
        for sock in server.sockets
    )
    print(f'Serving on {addrs}')

    async with server:
        await server.serve_forever()

asyncio.run(main())

The hostname is not so important here as the server doesn’t validate itself; it’s just used for specifying the network interface to listen on. We could have used “0.0.0.0” to listen to all interfaces.

The context starts by specifying the TLS protocol. However it also loads the certificate and key for the server. It doesn’t need the CA bundle as the client doesn’t send a certificate.

We don’t use the verify mode flag or check hostname, as the client doesn’t have a certificate to send. The get_extra_info method will return None as the client has no certificate.

What’s going on?

When the client connects the “TLS handshake is performed”. To establish trust the server sends it’s certificate. The client checks the certificate with it’s known certificate authorities to make sure the certificate is valid.

Step 3 — Client Certificates

Server authentication is the most common scenario when using SSL certificates. This is what happens when a browser connects to a web server on the internet.

Client certificates are used when a server needs to establish trust with a client. For example nodes in a distributed service would need to trust their peers. It can also be used instead of a password to provide authentication.

Client

Here is the client.

import asyncio
from os.path import expanduser
import socket
import ssl

async def tcp_echo_client(message):

    host = socket.gethostname()

    context = ssl.SSLContext(ssl.PROTOCOL_TLS)
    # Load the client certificate.
    context.load_cert_chain(
        expanduser("~/.keys/client.crt"),
        expanduser("~/.keys/client.key")
    )
    context.load_verify_locations(
        expanduser("~/.keys/cacerts.pem")
    )
    context.verify_mode = ssl.CERT_REQUIRED
    context.check_hostname = True

    reader, writer = await asyncio.open_connection(
        host,
        8888,
        ssl=context
    )

    peercert = writer.get_extra_info('peercert')
    print(f'Peer cert: {peercert!r}')

    print(f'Send: {message!r}')
    writer.write(message.encode())

    data = await reader.read(100)
    print(f'Received: {data.decode()!r}')

    print('Close the connection')
    writer.close()

asyncio.run(tcp_echo_client('Hello World!'))

The only addition to the client is the loading of the client certificate. Note that the key and certificate are client.{crt,key} not server.{crt,key}. Part of a certificate specifies its usage which includes whether it can authenticate as a client or a server or both. It is possible to create a peer certificate that can be used for both client and server authentication.

Server

Here is the server.

import asyncio
from os.path import expanduser
import socket
import ssl

async def handle_echo(reader, writer):

    data = await reader.read(100)
    message = data.decode()
    addr = writer.get_extra_info('peername')
    # We should get the child certificate.
    peercert = writer.get_extra_info('peercert')
    print(f"Received {message!r} from {addr!r} with cert {peercert!r}")

    print(f"Send: {message!r}")
    writer.write(data)
    await writer.drain()

    print("Close the connection")
    writer.close()

async def main():

    host = socket.gethostname()

    context = ssl.SSLContext(ssl.PROTOCOL_TLS)
    context.load_cert_chain(
        expanduser("~/.keys/server.crt"),
        expanduser("~/.keys/server.key")
    )
    context.load_verify_locations(
        expanduser("~/.keys/cacerts.pem")
    )
    # Check the client certificate
    context.verify_mode = ssl.CERT_REQUIRED
    context.check_hostname = True

    server = await asyncio.start_server(
        handle_echo,
        host,
        8888,
        ssl=context
    )

    addrs = ', '.join(
        str(sock.getsockname())
        for sock in server.sockets
    )
    print(f'Serving on {addrs}')

    async with server:
        await server.serve_forever()

asyncio.run(main())

The server has to load the CA bundle to check the client certificate. We also need to set verify_mode to CERT_REQUIRED and check_hostname to true.

What’s going on?

The ssl handshake now involves the client’s certificate as well as the server’s. The server will check its bundle to ensure the client certificate has a valid CA.

Step 4 — Web Services

We can apply the information we have gained above to writing a full trust web service.

Client

I’m going to stick with asyncio and use the httpx client instead of requests.

pip install httpx

Here is the client.

import asyncio
from os.path import expanduser
import socket
import ssl

import httpx

async def http_echo_client(message: str):
    host = socket.gethostname()

    context = ssl.SSLContext(ssl.PROTOCOL_TLS)
    context.check_hostname = True
    context.verify_mode = ssl.CERT_REQUIRED
    context.load_verify_locations(expanduser("~/.keys/cacerts.pem"))
    context.load_cert_chain(
        certfile=expanduser("~/.keys/client.crt"),
        keyfile=expanduser("~/.keys/client.key"),
    )

    async with httpx.AsyncClient(verify=context) as client:
        response = await client.post(
            f"https://{host}:8888/echo",
            content=message.encode()
        )
        print(response.content.decode())

asyncio.run(http_echo_client('Hello World!'))

The SSL context gets set up in exactly the same manner as before, and is simply passed to the httpx client’s verify argument. The client then hits the /echo endpoint with a POST request where the body is the binary encoded text message.

Server

For the server I’m going to use hypercorn as the AGSI web server and bareASGI as the ASGI web framework.

pip install hypercorn bareASGI

Here is the server.

import asyncio
from os.path import expanduser
import socket
import ssl

from hypercorn.asyncio import serve
from hypercorn.config import Config
from bareasgi import Application, HttpRequest, HttpResponse, text_reader

async def echo(request: HttpRequest) -> HttpResponse:
    body = await text_reader(request.body)
    return HttpResponse.from_text(body)

async def main():
    host = socket.gethostname()

    app = Application()
    app.http_router.add({'POST'}, '/echo', echo)

    config = Config()
    config.bind = [f"{host}:8888"]
    config.certfile = expanduser("~/.keys/server.crt")
    config.keyfile = expanduser("~/.keys/server.key")
    config.ca_certs = expanduser("~/.keys/cacerts.pem")
    config.verify_mode = ssl.CERT_REQUIRED
    await serve(app, config)

asyncio.run(main())

The most simple way to pass the SSL configuration to hypercorn is through the Config object.

The echo function is registered as a handler for POST requests to the /echo endpoint. It reads the body and returns it as text.

Thoughts

At the start this felt like quite a complex problem, but in the end the solution was only a few lines of code.

I hope you find it useful.