Using TLS safely in Node.js and Python

tls/defaults.mjs

// defaults.mjs: what Node.js does with no options, and what switching verification off does. The impostor is a server that holds a
// certificate for the right name from a CA the client has never been told to trust.
import tls from 'node:tls';
import { session } from './pipe.mjs';

console.log(`defaults: minimum ${tls.DEFAULT_MIN_VERSION}, maximum ${tls.DEFAULT_MAX_VERSION}`);

const impostor = { serverChain: 'rogue_chain.pem', serverKey: 'rogue.key' };
const strict = await session(impostor);
console.log(`client with the defaults:            ${strict.clientError ? `refused, ${strict.clientError}` : `connected, reply ${strict.reply}`}`);

const lax = await session({ ...impostor, client: { rejectUnauthorized: false } });
console.log(`client with rejectUnauthorized false: ${lax.clientError ? `refused, ${lax.clientError}` : `connected, reply ${lax.reply}`}`);
console.log(`with verification off the socket still records authorized ${lax.socket.authorized} (${lax.socket.authorizationError}), and the connection carried on`);

const honest = await session();
console.log(`client of the real server:           connected, reply ${honest.reply}, authorized ${honest.socket.authorized}`);

defaults: minimum TLSv1.2, maximum TLSv1.3
client with the defaults:            refused, SELF_SIGNED_CERT_IN_CHAIN
client with rejectUnauthorized false: connected, reply pong
with verification off the socket still records authorized false (SELF_SIGNED_CERT_IN_CHAIN), and the connection carried on
client of the real server:           connected, reply pong, authorized true

tls/pipe.mjs

// pipe.mjs: a TLS client and a TLS server that talk through memory, so the examples need no network. The server is a real
// tls.Server that is handed a connection by hand, which the Node.js documentation allows for any Duplex stream, and the client
// is tls.connect over the other end of the pipe. The certificates are in ./pki, made by make_pki.sh.
import tls from 'node:tls';
import { Duplex, PassThrough } from 'node:stream';
import { readFileSync } from 'node:fs';

export const read = (file) => readFileSync(new URL(`./pki/${file}`, import.meta.url));

// The identity a service gives itself: the URI entry of the subject alternative names, as in a SPIFFE-style name.
export const identityOf = (peer) => /URI:([^,\s]+)/.exec(peer.subjectaltname ?? '')?.[1];

function pipe() {
  const toServer = new PassThrough();
  const toClient = new PassThrough();
  return [Duplex.from({ readable: toClient, writable: toServer }), Duplex.from({ readable: toServer, writable: toClient })];
}

// One connection. The client writes its request as soon as it thinks it is connected, and the server answers with whatever
// answer() returns. Under TLS 1.3 a client can finish its handshake before the server has judged the client's certificate, so
// only a reply shows that the connection was accepted.
export function session({ client = {}, server = {}, serverChain = 'server_chain.pem', serverKey = 'server.key', request = 'ping', answer = () => 'pong' } = {}) {
  const [clientEnd, serverEnd] = pipe();
  const result = { reply: null, clientError: null, serverVerdict: null, socket: null };
  const listener = tls.createServer({ key: read(serverKey), cert: read(serverChain), ...server }, (socket) => {
    const peer = socket.getPeerCertificate();
    const named = Object.keys(peer).length > 0 ? (identityOf(peer) ?? peer.subject.CN) : null;
    result.serverVerdict = socket.authorized ? `authorized as ${named}` : `not authorized (${named ? socket.authorizationError : 'no certificate presented'})`;
    socket.on('data', (data) => socket.write(answer(socket, String(data))));
    socket.on('error', () => {});
  });
  listener.on('tlsClientError', (error, socket) => {
    result.serverVerdict = `refused (${error.code === 'ABORT_ERR' ? socket.authorizationError : error.code})`;
  });
  listener.emit('connection', serverEnd);
  return new Promise((resolve) => {
    const socket = tls.connect({ socket: clientEnd, servername: 'orders.example.test', ca: read('root.pem'), ...client });
    result.socket = socket;
    let done = false;
    const finish = () => {
      if (done) return;
      done = true;
      setImmediate(() => {
        const outcome = { ...result };
        socket.destroy();
        listener.close();
        resolve(outcome);
      });
    };
    socket.on('secureConnect', () => socket.write(request));
    socket.on('data', (data) => {
      if (done) return;
      result.reply = String(data);
      result.negotiated = { protocol: socket.getProtocol(), cipher: socket.getCipher().name, group: socket.getEphemeralKeyInfo().name };
      finish();
    });
    socket.on('error', (error) => {
      if (done) return;
      result.clientError = error.code;
      result.clientMessage = error.message;
      finish();
    });
    socket.on('close', finish);
  });
}

tls/defaults.py

# defaults.py: what Python's ssl module does by default, and what switching verification off does. The impostor is a server that holds
# a certificate for the right name from a CA the client has not been told to trust.
import ssl

from membio import connect

ctx = ssl.create_default_context()
print("ssl.create_default_context():")
print(f"  verify_mode {ctx.verify_mode.name}, check_hostname {ctx.check_hostname}, minimum_version {ctx.minimum_version.name}")
print(f"  strict X.509 checks {bool(ctx.verify_flags & ssl.VERIFY_X509_STRICT)}, partial chains {bool(ctx.verify_flags & ssl.VERIFY_X509_PARTIAL_CHAIN)}")

impostor = dict(chain="rogue_chain.pem", key="rogue.key")
print()
try:
    connect(**impostor)
    print("impostor, client from create_default_context plus the CA: connected")
except ssl.SSLCertVerificationError as error:
    print(f"impostor, client from create_default_context plus the CA: refused, {error.verify_message} ({error.verify_code})")

off = ssl.create_default_context()
off.check_hostname = False
off.verify_mode = ssl.CERT_NONE
session = connect(client=off, **impostor)
print(f"impostor, client with check_hostname False and CERT_NONE: connected, {session.version()}")

ssl.create_default_context():
  verify_mode CERT_REQUIRED, check_hostname True, minimum_version TLSv1_2
  strict X.509 checks True, partial chains True

impostor, client from create_default_context plus the CA: refused, self-signed certificate in certificate chain (19)
impostor, client with check_hostname False and CERT_NONE: connected, TLSv1.3

tls/membio.py

# membio.py: a TLS client and a TLS server in one process, talking through memory buffers instead of a network. connect() makes one
# handshake between a client context and a server that holds a certificate chain from ./pki (made by make_pki.sh), and it raises
# what a real socket would raise.
import ssl
from pathlib import Path

PKI = Path(__file__).parent / "pki"


def trusting_context():
    """The client context a careful program builds: the defaults, plus the root of the private CA."""
    return ssl.create_default_context(cafile=str(PKI / "root.pem"))


def connect(chain="server_chain.pem", key="server.key", name="orders.example.test", client=None):
    server_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    server_ctx.load_cert_chain(PKI / chain, PKI / key)
    client_ctx = client or trusting_context()
    c_in, c_out, s_in, s_out = (ssl.MemoryBIO() for _ in range(4))
    client_side = client_ctx.wrap_bio(c_in, c_out, server_hostname=name)
    server_side = server_ctx.wrap_bio(s_in, s_out, server_side=True)
    pending = {"client": client_side, "server": server_side}
    for _ in range(10):
        for side in list(pending):
            try:
                pending[side].do_handshake()
                del pending[side]
            except ssl.SSLWantReadError:
                pass
        s_in.write(c_out.read())
        c_in.write(s_out.read())
        if not pending:
            return client_side
    raise RuntimeError("the handshake did not finish")