"""CluPilot Terminal-Bruecke.

Loest ein einmaliges Ticket ein, baut damit eine SSH-Sitzung auf und verbindet
deren PTY mit einem WebSocket. Mehr tut dieser Dienst nicht: keine Datenbank,
kein Laravel, keine eigene Anmeldung.

Das Ticket ist die einzige Tuer. Es lebt dreissig Sekunden, traegt genau eine
Sitzung, und `GETDEL` loescht es im selben Zug, in dem es gelesen wird.

DAS TICKET STEHT NICHT IN DER ADRESSZEILE. Es kommt als `Sec-WebSocket-Protocol`
— derselbe Weg, den Kubernetes fuer `exec` benutzt, und aus demselben Grund: die
Adresszeile eines Upgrade-Antrags schreibt jeder Reverse Proxy auf der Strecke
mit, und auf dieser Strecke stehen zwei (nginx im app-Container, Caddy davor),
von denen nur einer aus diesem Repo konfiguriert wird. Dreissig Sekunden und
eine einzige Einloesung begrenzen den Schaden, aber ein Geheimnis, das gar nicht
erst in ein Protokoll geraet, braucht diese Begrenzung nicht.

DIE AUSGABE GEHT IMMER ALS BINAERRAHMEN raus, nie als Text — siehe to_browser().
Was hereinkommt, sind Tastenanschlaege des Browsers als Textrahmen; diese
Asymmetrie ist Absicht: xterm.js liefert Strings, ein PTY liefert Bytes, und
was ein PTY schickt, ist nicht zwingend gueltiges UTF-8.
"""
import asyncio
import base64
import hashlib
import http
import io
import json
import logging
import os
import re
import socket

import paramiko
import redis.asyncio as redis
from websockets.asyncio.server import serve as websocket_server
from websockets.exceptions import InvalidStatus

REDIS_URL = os.environ.get("REDIS_URL", "redis://redis:6379/1")

# phpredis stellt jedem Schluessel dieser Verbindung `database.redis.options.
# prefix` voran — auf dieser Installation `clupilot-database-`, abgeleitet aus
# APP_NAME. Fuer PHP ist das unsichtbar, fuer jeden anderen Client ist es Teil
# des Schluessels. Wer APP_NAME aendert, ohne REDIS_PREFIX zu setzen, findet
# hier kein Ticket mehr — deshalb steht der benutzte Wert beim Start im Log.
REDIS_PREFIX = os.environ.get("REDIS_PREFIX", "clupilot-database-")

IDLE_SECONDS = int(os.environ.get("IDLE_SECONDS", "900"))
PORT = int(os.environ.get("PORT", "8082"))

# 32 Byte Zufall als Hex, so wie TerminalTicket::issue() sie bildet. Was diese
# Form nicht hat, ist kein Ticket dieser Anwendung und braucht Redis gar nicht
# erst zu behelligen.
TICKET_PATTERN = re.compile(r"\A[0-9a-f]{64}\Z")

log = logging.getLogger("terminal")


async def redeem_payload(ticket: str) -> dict | None:
    """Liest das Ticket und loescht es im selben Zug.

    Der Inhalt ist reines JSON, weil TerminalTicket ihn ueber `Redis::
    connection('cache')` schreibt und nicht ueber die Cache-Fassade — die
    verpackte den JSON-Text noch einmal PHP-serialisiert (`s:412:"{...}"`).
    Nachgemessen, nicht angenommen. Faellt das je zurueck, wirft json.loads
    hier, die Sitzung kommt nicht zustande und es steht im Log: besser als ein
    Parser, der beide Formen frisst und den Rueckfall verdeckt.
    """
    client = redis.from_url(REDIS_URL)
    try:
        raw = await client.getdel(f"{REDIS_PREFIX}terminal:ticket:{ticket}")
    finally:
        await client.aclose()

    if not raw:
        return None

    return json.loads(raw.decode() if isinstance(raw, bytes) else raw)


def load_key(text: str) -> paramiko.PKey:
    """Der Schluessel aus dem Tresor, in der Form, in der er dort liegt.

    App\\Services\\Ssh\\Keypair legt Ed25519 im OpenSSH-Format an; ein Betreiber
    kann aber auch einen eigenen Schluessel eingesetzt haben. paramiko hat kein
    "lade, was auch immer das ist" — jede Klasse liest genau ihr Format.
    """
    for cls in (paramiko.Ed25519Key, paramiko.ECDSAKey, paramiko.RSAKey):
        try:
            return cls.from_private_key(io.StringIO(text))
        except (paramiko.SSHException, ValueError):
            continue

    raise RuntimeError("Der hinterlegte SSH-Schluessel ist in keinem unterstuetzten Format lesbar")


def fingerprints(key: paramiko.PKey) -> set[str]:
    """Die Fingerabdruecke SO, WIE CLUPILOT SIE BILDET — nicht wie OpenSSH.

    PhpseclibRemoteShell::hostKeyFingerprint() hasht nicht den rohen
    Schluessel-Blob (das waere der bekannte OpenSSH-Fingerabdruck), sondern die
    Zeichenkette `<algorithmus> <base64-blob>`, die phpseclib'
    getServerPublicHostKey() zurueckgibt:

        'SHA256:' . base64_encode(hash('sha256', $key, true))

    Wer das hier mit dem OpenSSH-Fingerabdruck vergleicht, bekommt bei JEDEM
    Host eine Abweichung und faellt in die Versuchung, die Pruefung fuer kaputt
    zu halten und wegzulassen. Deshalb steht es hier so ausfuehrlich.

    Mehrere Kandidaten, weil `<algorithmus>` bei phpseclib aus der SIGNATUR
    stammt: bei einem RSA-Hostschluessel steht dort je nach Aushandlung
    `rsa-sha2-256`/`rsa-sha2-512`, waehrend der Blob selbst `ssh-rsa` heisst
    (Net/SSH2.php, key_exchange). Bei Ed25519 und ECDSA sind beide gleich.
    Mehrere Kandidaten schwaechen nichts ab: der base64-Blob des Hostschluessels
    steckt in jedem einzelnen von ihnen, ein anderer Host hat also auch unter
    jedem Namen einen anderen Fingerabdruck.
    """
    blob = base64.b64encode(key.asbytes()).decode()

    names = [key.get_name()]
    if key.get_name() == "ssh-rsa":
        names += ["rsa-sha2-256", "rsa-sha2-512"]

    return {
        "SHA256:" + base64.b64encode(hashlib.sha256(f"{name} {blob}".encode()).digest()).decode()
        for name in names
    }


def connect(payload: dict) -> tuple[paramiko.Transport, paramiko.Channel]:
    """SSH aufbauen — Fingerabdruck geprueft, BEVOR der Schluessel benutzt wird.

    Bewusst ueber Transport statt ueber SSHClient: SSHClient prueft den
    Hostschluessel gegen seine known_hosts-Sammlung, und dieser Container hat
    keine. Er kann auch keine haben — CluPilot kennt vom Host nur den
    Fingerabdruck, also einen Hash, aus dem sich kein Schluessel zurueckrechnen
    laesst. Uebrig blieben zwei schlechte Wege: RejectPolicy (verbindet nie) oder
    AutoAddPolicy (vertraut blind und meldet sich mit einem Root-Schluessel an,
    bevor irgendetwas geprueft ist). Der Transport erlaubt die dritte, richtige
    Reihenfolge: Handschlag, Fingerabdruck vergleichen, erst dann anmelden.
    """
    key = load_key(payload["private_key"])

    expected = (payload.get("fingerprint") or "").strip()
    if not expected:
        # TerminalTicket::issue() laesst das gar nicht erst entstehen. Wenn hier
        # doch nichts steht, ist die Antwort trotzdem nicht "dann eben ohne".
        raise RuntimeError("Ticket ohne Fingerabdruck")

    # Eigener Socket mit Zeitgrenze: paramiko.Transport(("ip", 22)) verbindet im
    # Konstruktor ohne eine, und ein Host im Tunnel, der gerade nicht da ist,
    # haengt dann am TCP-Standard von zwei Minuten — laenger, als das Ticket
    # ueberhaupt gilt.
    sock = socket.create_connection((payload["ip"], 22), timeout=10)
    sock.settimeout(None)

    transport = paramiko.Transport(sock)
    transport.banner_timeout = 15

    try:
        transport.start_client(timeout=10)

        remote = transport.get_remote_server_key()
        if expected not in fingerprints(remote):
            raise RuntimeError("Fingerabdruck weicht ab")

        transport.auth_publickey(payload["user"], key)

        # Damit eine tote Gegenstelle auffaellt, bevor der Leerlauf zuschlaegt:
        # ein Fenster, das nach einem Netzausfall stumm offen bleibt, sieht aus
        # wie eine haengende Sitzung.
        transport.set_keepalive(30)

        channel = transport.open_session(timeout=10)
        channel.get_pty(term="xterm-256color")
        channel.invoke_shell()
        channel.settimeout(0.0)
    except Exception:
        transport.close()
        raise

    return transport, channel


async def bridge(websocket, payload: dict) -> None:
    """Bytes in beide Richtungen, bis eine der drei Grenzen greift."""
    loop = asyncio.get_running_loop()

    try:
        transport, channel = await loop.run_in_executor(None, connect, payload)
    except Exception as exc:
        # Die Meldung ins Log, nicht an den Browser: was genau schiefging, ist
        # eine Auskunft ueber den Host, und die hat auf der anderen Seite eines
        # WebSockets nichts verloren.
        log.warning("Kein SSH zu %s: %s", payload.get("ip"), exc)
        await websocket.close(code=4502)
        return

    log.info(
        "Sitzung offen — Host %s (%s), Betreiber %s",
        payload.get("host_uuid"), payload.get("ip"), payload.get("operator_id"),
    )

    # Ein gemeinsamer Zeitstempel fuer beide Richtungen: Leerlauf heisst, dass
    # WEDER getippt noch ausgegeben wurde.
    last = loop.time()

    async def to_ssh():
        nonlocal last
        async for message in websocket:
            last = loop.time()
            data = message.encode() if isinstance(message, str) else message
            await loop.run_in_executor(None, channel.sendall, data)

    async def to_browser():
        nonlocal last
        while True:
            if channel.recv_ready():
                data = channel.recv(32768)
                if not data:
                    return
                last = loop.time()
                # bytes, also ein BINAERRAHMEN — immer. Ein gemischter Betrieb
                # aus Text- und Binaerrahmen haette hier keinen Nutzen und im
                # Browser zwei Fehlerbilder.
                await websocket.send(data)
                continue

            if channel.closed or channel.exit_status_ready() or channel.eof_received:
                return

            await asyncio.sleep(0.02)

    async def idle_watch():
        while True:
            await asyncio.sleep(5)
            if loop.time() - last > IDLE_SECONDS:
                log.info("Sitzung nach %s s Leerlauf beendet", IDLE_SECONDS)
                return

    tasks = [asyncio.create_task(task()) for task in (to_ssh, to_browser, idle_watch)]

    try:
        # Was zuerst endet, beendet alles: das Fenster zu, die Sitzung tot oder
        # der Leerlauf abgelaufen.
        await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
    finally:
        for task in tasks:
            task.cancel()
        channel.close()
        transport.close()
        await websocket.close()
        log.info("Sitzung geschlossen — Host %s", payload.get("host_uuid"))


async def serve(websocket) -> None:
    ticket = websocket.subprotocol or ""

    payload = await redeem_payload(ticket) if TICKET_PATTERN.match(ticket) else None

    if payload is None:
        # Kein Grund, keine Meldung mit Inhalt: ein unbekanntes Ticket darf
        # nicht verraten, ob es je eines gab. Und nie das Ticket selbst ins Log
        # — es ist genau so lange gueltig wie das, was jemand mitliest.
        log.info("Ticket abgewiesen (unbekannt, abgelaufen oder schon eingeloest)")
        await websocket.close(code=4401)
        return

    await bridge(websocket, payload)


class HealthNoiseFilter(logging.Filter):
    """Die Lebendpruefung ist kein Fehler, auch wenn die Bibliothek das so sieht.

    websockets protokolliert JEDEN Handschlag, der nicht mit 101 endet, als
    "opening handshake failed" samt Stapelabzug — auch den, den wir selbst mit
    200 beantwortet haben (siehe process_request). Alle 30 Sekunden ein Fehler
    plus eine Abweisung im Log macht das Log wertlos, und "einfach ignorieren"
    ist keine Anleitung.

    Deshalb genau dieser eine Fall: abgewiesen mit 200, also unsere eigene
    Antwort auf /healthz — kein anderer Weg durch diesen Dienst endet mit 200.
    Ein echter Fehlschlag (abgebrochene Verbindung, kaputter Antrag, 4xx) steht
    weiterhin da.
    """

    def filter(self, record: logging.LogRecord) -> bool:
        exception = record.exc_info[1] if record.exc_info else None

        if isinstance(exception, InvalidStatus) and exception.response.status_code == 200:
            return False

        # Vorsichtig ausgepackt: ein Filter, der selbst wirft, reisst die
        # Protokollierung mit — und damit die Meldung, die gerade gebraucht wird.
        arguments = record.args if isinstance(record.args, tuple) else ()

        return not (str(record.msg).startswith("connection rejected") and arguments[:1] == (200,))


def process_request(connection, request):
    """Ein Ort fuer die Lebendpruefung — mit Antwort, ohne Handschlag.

    Die naheliegende Pruefung (TCP aufmachen, wieder zumachen) laesst diesen
    Dienst alle 30 Sekunden einen abgebrochenen WebSocket-Handschlag als
    Fehlermeldung mit vollem Stapelabzug protokollieren. Ein Log, das im
    Ruhezustand Fehler schreibt, ist ein Log, das im Ernstfall niemand liest.
    """
    if request.path == "/healthz":
        return connection.respond(http.HTTPStatus.OK, "ok\n")

    return None


def select_subprotocol(connection, subprotocols):
    """Das angebotene Ticket wird zurueckgespiegelt, sonst bricht der Browser ab.

    Hier wird NICHT geprueft, ob es ein gutes Ticket ist: ein abgelehntes
    Unterprotokoll laesst den Handschlag scheitern, und der Browser bekaeme
    einen nackten Verbindungsfehler statt der Schliessung mit 4401, an der die
    Seite "abgewiesen" von "kein Weg dorthin" unterscheidet. Die Pruefung steht
    in serve(), eine Zeile spaeter.

    Die websockets-Bibliothek hat den Wert vorher als HTTP-Token geparst; CR/LF
    kaeme hier also gar nicht erst an.
    """
    return subprotocols[0] if subprotocols else None


async def main() -> None:
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    log.info("Terminal-Bruecke lauscht auf %s, Redis-Praefix %r, Leerlauf %s s", PORT, REDIS_PREFIX, IDLE_SECONDS)

    # Ein eigener Logger fuer die Bibliothek, damit der Filter oben nur ihre
    # Meldungen sieht und nicht die dieses Moduls.
    server_log = logging.getLogger("terminal.server")
    server_log.addFilter(HealthNoiseFilter())

    async with websocket_server(
        serve,
        "0.0.0.0",
        PORT,
        logger=server_log,
        process_request=process_request,
        select_subprotocol=select_subprotocol,
        ping_interval=20,
    ):
        await asyncio.Future()


if __name__ == "__main__":
    asyncio.run(main())
