Python Websockets SSL with Let's Encrypt

This tutorial is an explanation of my gist Python Websockets SSL with Let's Encrypt .

With the launch of HTML5 in 2008, a technology that immediately took off in popularity was WebSockets. According to W3C, the basic definition of a Websocket is - an interface that enables web applications to maintain bidirectional communications with server-side processes.

In this short tutorial, I'll be showing you how you can host a WebSocket server with SSL enabled on it. This allows your socket server to run on an HTTPS address.

Setup

We'll be working with asyncio library and for WebSocket server implementation will be using the websockets library.

The asyncio library comes pre-packaged with Python distributions since Python 3.4. To install websockets library, you can use the following command:

pip install websockets

Next, we'll be looking to how to generate the SSL certificate files.

Generate certificate and keyfile using Let's Encrypt

Before you can enable SSL on your WebSocket being run by a Python script, you'll have to generate certificate files for your domain.

The basic gist of this step is to fetch Let's Encrypt signed certficiates for your domain and store them on your server where the Python WebSocket script is running.

Here's a great quick tutorial on How To Use Certbot Standalone Mode to Retrieve Let's Encrypt SSL Certificates on Ubuntu 20.04.

Make certificate files accessible

After generating the files correctly, you need to make them accessible to the current user who runs the Python WebSocket script.

In the previous step, if certbot stored your certificate files at /etc/letsencrypt/live/your_domain location, you should be able to see 4 files when you perform an ls on the folder -

~$ ls /etc/letsencrypt/live/your_domain
cert.pem  chain.pem  fullchain.pem  privkey.pem  README

To change the owner of the certificate files, use the following command:

~$ sudo chown -R $(id -u):$(id -g) /etc/letsencrypt/live/your_domain

Next, ensure that the right permissions are applied to the folder:

~$ sudo chmod -R 400 /etc/letsencrypt/live/your_domain

We're now good to read these files from the Python WebSocket script.

Create server script

While your WebSocket server script will differ from the most barebones implementation, here's one for you -

Create a file named socket_server.py. Then, make all the necessary imports.

#!/usr/bin/env python

# WS server example that synchronizes state across clients

import asyncio
import json
import logging
import websockets
import ssl

logging.basicConfig()

Next, let us configure the SSL for this script, as shown below -

ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)

# Generate with Lets Encrypt, chown to current user and 400 permissions
ssl_cert = "/etc/letsencrypt/live/your_domain/fullchain.pem"
ssl_key = "/etc/letsencrypt/live/your_domain/privkey.pem"

ssl_context.load_cert_chain(ssl_cert, keyfile=ssl_key)

The above code configures the script to run a TLS server with the certificates available in the folder generated by certbot.

Next, we define the functions that shall be used to notify clients of the socket server and to get information from them.

STATE = {"value": 0}

USERS = set()


def state_event():
    return json.dumps({"type": "state", **STATE})


def users_event():
    return json.dumps({"type": "users", "count": len(USERS)})


async def notify_state():
    if USERS:  # asyncio.wait doesn't accept an empty list
        message = state_event()
        await asyncio.wait([user.send(message) for user in USERS])


async def notify_users():
    if USERS:  # asyncio.wait doesn't accept an empty list
        message = users_event()
        await asyncio.wait([user.send(message) for user in USERS])

After that, we need to add functions that register and unregister clients from the WebSocket.

async def register(websocket):
    USERS.add(websocket)
    await notify_users()


async def unregister(websocket):
    USERS.remove(websocket)
    await notify_users()

This done, let us implement a function that gets state update requests from clients and performs it.

async def counter(websocket, path):
    # register(websocket) sends user_event() to websocket
    await register(websocket)
    try:
        await websocket.send(state_event())
        async for message in websocket:
            data = json.loads(message)
            if data["action"] == "minus":
                STATE["value"] -= 1
                await notify_state()
            elif data["action"] == "plus":
                STATE["value"] += 1
                await notify_state()
            else:
                logging.error("unsupported event: {}", data)
    finally:
        await unregister(websocket)

Finally, we run the server.

start_server = websockets.serve(counter, "0.0.0.0", 6789, ssl=ssl_context)

asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

How you expose the server to the internet, I shall leave that on your use case. However, you can explore this tutorial on Websockets - Deploy behind nginx.

Implement a client page

To test the above WebSocket script, you can spin up your own client script or feel free to use the socket_client.html file I've provided.

Make sure to update the following line in the client file to point to your live server -

websocket = new WebSocket("wss://localhost:6789/");

Note that we're using the wss:// protocol here instead of ws:// protocol which is popularly found on other tutorials on the internet.

Conclusion

This was my attempt at explaining the gist I put out more than an year back. Hope it helps you go through the process easier than how I had first written it.

Make sure to leave me feedback on how this blog went!