🏡 index : quine-global/quine-anaconda-webui.git

#!/usr/bin/python3
#
# Copyright (C) 2026 Red Hat, Inc.
# SPDX-License-Identifier: LGPL-2.1-or-later
#
# TLS termination proxy for Anaconda Web UI remote installation.
#
# Terminates TLS on port 443 and forwards plaintext TCP to cockpit-ws
# on 127.0.0.1:80. Uses the certificate generated by
# cockpit-certificate-ensure.

import asyncio
import glob
import ssl
import sys

LISTEN_ADDRESS = "0.0.0.0"
LISTEN_PORT = 443
BACKEND_ADDRESS = "127.0.0.1"
BACKEND_PORT = 80
CERT_DIR = "/etc/cockpit/ws-certs.d"


async def pipe(reader, writer):
    """Forward data from reader to writer until EOF or connection error."""
    try:
        while data := await reader.read(8192):
            writer.write(data)
            await writer.drain()
    except (ConnectionResetError, BrokenPipeError, OSError):
        pass
    finally:
        writer.close()


async def handle(client_reader, client_writer):
    """Handle a TLS client by opening a plaintext connection to the backend and piping both directions."""
    try:
        backend_reader, backend_writer = await asyncio.open_connection(
            BACKEND_ADDRESS, BACKEND_PORT
        )
        await asyncio.gather(
            pipe(client_reader, backend_writer),
            pipe(backend_reader, client_writer),
        )
    except (ConnectionRefusedError, OSError) as e:
        print(f"Backend connection failed: {e}", file=sys.stderr)
        client_writer.close()


def create_ssl_context():
    """Load the certificate generated by cockpit-certificate-ensure."""
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    certs = glob.glob(f"{CERT_DIR}/*.cert")
    if not certs:
        print(f"No certificates found in {CERT_DIR}", file=sys.stderr)
        sys.exit(1)
    cert_file = certs[0]
    key_file = cert_file.rsplit(".", 1)[0] + ".key"
    ctx.load_cert_chain(certfile=cert_file, keyfile=key_file)
    return ctx


async def main():
    ssl_ctx = create_ssl_context()
    server = await asyncio.start_server(
        handle, LISTEN_ADDRESS, LISTEN_PORT, ssl=ssl_ctx
    )
    print(
        f"TLS proxy listening on {LISTEN_ADDRESS}:{LISTEN_PORT} -> "
        f"{BACKEND_ADDRESS}:{BACKEND_PORT}",
        file=sys.stderr,
    )
    async with server:
        await server.serve_forever()


asyncio.run(main())