Run Websockets with SSL for a Production Application (py2exe)

Question

I am building a python application that is packaged as an EXE. This application runs as a Websockets server and communicates with a web app hosted on Firebase. However, the issue is that the web app on firebase requires WSS with an SSL certificate. Is there a way to create and sign a certificate that can work on any computer where the exe is installed?

Answer 1

You'll need to generate the certificate and keyfile using Let's Encrypt and for each system you want to run this exe file on it, you must have the same paths for the certificate and keyfile.

After generating the files correctly, you need to make them accessible to the current user who runs the script, my way of doing this was to copy it to the home directory of the current user and change the owner to the current user, set the permissions of the files to 400.

The following code snippet is an example that uses the python-websockets library to create a websocket and you can see how you can have ssl certificate for your websocket:

socket_server.py

#!/usr/bin/env python

# WS server example that synchronizes state across clients

import asyncio
import json
import logging
import websockets
import ssl

logging.basicConfig()
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)

# Generate with Lets Encrypt, copied to this location, chown to current user and 400 permissions
ssl_cert = "/home/username/fullchain.pem"
ssl_key = "/home/username/privkey.pem"

ssl_context.load_cert_chain(ssl_cert, keyfile=ssl_key)

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])


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


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


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)


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()