I want to enable TLS encryption with a custom / self-signed PKI for a self-contained distributed system built with Python 3’s asyncio.
I’ve read Python’s ssl docs as well as several tutorials:
From all this, I extracted the following steps in order to create a minimal example for a simple client and server with a self-signed PKI.
It seems to work as expected, but I’m unsure if I missed anything or create non-obvious bugs which could reduce or nullify the security of the system.
Here is what I did so far:
I'll create one root CA which signs a CSR for each machine. Each machine will have its own certificate and key.
Considerations for the root CA:
- The key length should be 2048 bits, or better, 4096 bits.
- A strong passphrase should be used.
- The key should not leave the machine (except if you store it on an SD card and hide it somewhere save). Level of paranoia should depend on the security needs.
- The certificate should not use the same FQDN as any production system.
Generate the key and a self-signed certificate:
$ openssl genrsa -aes256 -out ca.key 4096 $ openssl req -new -x509 -nodes -key ca.key -out ca.pem -days 1000
Create a key and a certificate signing request (CSR) on a machine:
$ openssl genrsa -out server.key 4096 $ openssl genrsa -out client.key 4096 $ openssl req -new -key server.key -out server.csr $ openssl req -new -key client.key -out client.csr
Now you can sign the device key with your CA cert:
$ openssl x509 -CA ca.pem -CAkey ca.key -CAcreateserial -req -in server.csr -out server.pem -days 365 $ openssl x509 -CA ca.pem -CAkey ca.key -CAcreateserial -req -in client.csr -out client.pem -days 365
Minimal Python example. We need different SSLContexts for the client and server, because they do some things differently. Both, client and server, will use TLS 1.2 with ECDH-AESGCM. They will verify each other’s certificates against the root CA cert. The client also checks the server’s host name. ECDH keys are only used once per SSL session and compression is disabled:
import asyncio import ssl @asyncio.coroutine def handle_client(reader, writer): data = yield from reader.read(100) print(data) writer.write(b'cya') yield from writer.drain() writer.close() @asyncio.coroutine def client(addr, ssl_ctx): reader, writer = yield from asyncio.open_connection(*addr, ssl=ssl_ctx) writer.write(b'ohai') data = yield from reader.read(100) print(data) writer.close() addr = ('127.0.0.1', 5555) loop = asyncio.get_event_loop() # Setup server server_ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2) server_ctx.verify_mode = ssl.CERT_REQUIRED server_ctx.options |= ssl.OP_SINGLE_ECDH_USE server_ctx.options |= ssl.OP_NO_COMPRESSION server_ctx.load_cert_chain(certfile='server.pem', keyfile='server.key') server_ctx.load_verify_locations(cafile='ca.pem') server_ctx.set_ciphers('ECDH+AESGCM') coro = asyncio.start_server(handle_client, *addr, ssl=server_ctx) server = loop.run_until_complete(coro) # Run client client_ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2) client_ctx.verify_mode = ssl.CERT_REQUIRED client_ctx.check_hostname = True client_ctx.load_cert_chain(certfile='client.pem', keyfile='client.key') client_ctx.load_verify_locations(cafile='ca.pem') client_ctx.set_ciphers('ECDH+AESGCM') loop.run_until_complete(client(addr, client_ctx)) # Shutdown server.close() loop.run_until_complete(server.wait_closed()) loop.close()
The script will output
as expected and TLS seems to work. Are there any problems with it or things that could be done better?
Thanks for your feedback.