diff --git a/app/admin/routes.py b/app/admin/routes.py index 3309043..4102aeb 100644 --- a/app/admin/routes.py +++ b/app/admin/routes.py @@ -1921,13 +1921,20 @@ async def list_sessions( result = [] for r in rows: recording_path = r[11] + is_active = r[8] is None + killable = r[0] in active_ids result.append({ "id": r[0], "user_id": r[1], "username": r[2], "host_id": r[3], "hostname": r[4], "host_group_name": r[5], "protocol": r[6], "started_at": r[7], "ended_at": r[8], "client_ip": r[9], "end_reason": r[10], - "is_active": r[8] is None, - "killable": r[0] in active_ids, + "is_active": is_active, + "killable": killable, "has_recording": bool(recording_path) and Path(recording_path).exists(), + # Live-Mitschau (GET /ws/sessions/{id}/watch): wie 'killable' nur + # moeglich, wenn die Sitzung auf DIESEM Worker-Prozess laeuft -- + # und bisher nur fuer SSH umgesetzt (RDP haette dafuer eine eigene + # Multiplexing-Loesung fuer den guacd-Binaer-Tunnel noetig). + "watchable": is_active and killable and r[6] == "ssh", }) return result diff --git a/app/security/active_sessions.py b/app/security/active_sessions.py index f36b732..ab72b16 100644 --- a/app/security/active_sessions.py +++ b/app/security/active_sessions.py @@ -14,17 +14,27 @@ uvicorn-Worker (Standard-Deployment dieses Projekts, siehe ansible/) sieht und beendet jeder Superadmin-Request alle laufenden Sitzungen. Bei mehreren Worker-Prozessen sind nur die Sitzungen DIESES Workers 'killable' -- die Sessionview zeigt das ueber das 'killable'-Feld pro Sitzung an, statt einen -Beenden-Versuch fehlschlagen zu lassen.""" +Beenden-Versuch fehlschlagen zu lassen. + +Seit der Live-Mitschau ("Aktive Session live ansehen", GET /ws/sessions/ +{id}/watch, app/ssh_proxy/terminal_ws.py) traegt jeder Registry-Eintrag +zusaetzlich eine Menge von Beobachter-Queues: app/ssh_proxy/terminal_ws.py:: +_pump_ssh_to_ws() speist jeden vom Zielsystem gelesenen Byte-Chunk sowohl an +den eigentlichen Sitzungsinhaber als auch (per broadcast()) an alle gerade +zuschauenden Superadmins. Genau wie 'killable' gilt: nur auf demselben +Worker-Prozess moeglich -- ein Beobachter kann sich nur an eine Sitzung +haengen, die in DIESER Prozess-Registry steht.""" from __future__ import annotations import asyncio -from dataclasses import dataclass +from dataclasses import dataclass, field @dataclass class ActiveSession: session_id: int task: asyncio.Task + watchers: set[asyncio.Queue] = field(default_factory=set) _active: dict[int, ActiveSession] = {} @@ -44,3 +54,38 @@ def get(session_id: int) -> ActiveSession | None: def all_ids() -> set[int]: return set(_active.keys()) + + +def add_watcher(session_id: int) -> asyncio.Queue | None: + """Meldet einen Beobachter fuer eine laufende Sitzung an. Gibt None + zurueck, wenn die Sitzung nicht (mehr) auf diesem Prozess laeuft -- + der Aufrufer (WS-Route) beendet die Beobachter-Verbindung dann sofort.""" + entry = _active.get(session_id) + if entry is None: + return None + queue: asyncio.Queue = asyncio.Queue(maxsize=1000) + entry.watchers.add(queue) + return queue + + +def remove_watcher(session_id: int, queue: asyncio.Queue) -> None: + entry = _active.get(session_id) + if entry is not None: + entry.watchers.discard(queue) + + +def broadcast(session_id: int, data: bytes) -> None: + """Verteilt einen Ausgabe-Chunk an alle aktuell zuschauenden Beobachter + dieser Sitzung. Bewusst best-effort wie log_stream.py: ein langsamer + oder haengender Beobachter darf weder die eigentliche Sitzung + verlangsamen noch blockieren -- volle Queues verwerfen den Chunk + stattdessen (put_nowait/QueueFull), statt den Sitzungsinhaber warten zu + lassen.""" + entry = _active.get(session_id) + if entry is None or not entry.watchers: + return + for queue in list(entry.watchers): + try: + queue.put_nowait(data) + except asyncio.QueueFull: + pass diff --git a/app/ssh_proxy/sftp.py b/app/ssh_proxy/sftp.py index 8f0dcca..8e27b4f 100644 --- a/app/ssh_proxy/sftp.py +++ b/app/ssh_proxy/sftp.py @@ -8,6 +8,7 @@ from __future__ import annotations import hashlib import logging +import asyncssh from fastapi import APIRouter, Depends, HTTPException, Query, Request, UploadFile, status from fastapi.responses import StreamingResponse @@ -111,10 +112,33 @@ async def upload_file( # erreichbar): 400 mit Klartext statt eines unbehandelten 500. logger.warning("Dateitransfer fuer Host %s nicht moeglich: %s", host_id, exc) raise HTTPException(status.HTTP_400_BAD_REQUEST, describe_connection_error(exc)) + except asyncssh.Error as exc: + # Bugfix: SSH_SETUP_ERRORS deckt nur Fehler VOR der Anmeldung ab + # (Pinning/Konfiguration). asyncssh.Error ist asyncssh's gemeinsame + # Basisklasse -- das schliesst sowohl einen echten Verbindungsfehler + # WAEHREND asyncssh.connect() ein (z.B. Verbindung abgelehnt/abgebrochen, + # von connect_to_host() bewusst unuebersetzt weitergereicht) als auch + # SFTP-Fehler NACH erfolgreicher Anmeldung (z.B. SFTPNoSuchFile, wenn + # das Zielverzeichnis nicht existiert, oder SFTPPermissionDenied). Beides + # war hier bisher NICHT gefangen und lief unbehandelt bis Starlette + # durch, das bei einer unbehandelten Exception (debug=False) eine + # KLARTEXT-500-Antwort "Internal Server Error" liefert statt JSON -- + # daher der Frontend-Fehler "Unexpected token 'I', 'Internal S'... is + # not valid JSON" beim Upload. + logger.warning("SFTP-Fehler bei Host %s: %s", host_id, exc) + raise HTTPException(status.HTTP_400_BAD_REQUEST, f"SFTP-Fehler: {exc}") + except HTTPException: + raise + except Exception: + # Letztes Auffangnetz (gleiches Muster wie terminal_ws.py): niemals + # eine unbehandelte Ausnahme bis zu Starlettes Klartext-500 durchreichen + # -- das Frontend erwartet hier immer eine JSON-Antwort. + logger.exception("Unerwarteter Fehler beim Datei-Upload fuer Host %s", host_id) + raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, "Unerwarteter Fehler beim Dateitransfer") await _log_transfer( conn, user=user, host_id=host_id, client_ip=_client_ip(request), direction="upload", - filename=file.filename or remote_path, size=len(data), sha256=sha256, av_scan_result=av_result, + filename=file.filename or remote_path, size=len(data), sha256=sha256, av_result=av_result, ) return {"status": "ok", "sha256": sha256, "size": len(data), "av_scan_result": av_result} @@ -146,12 +170,24 @@ async def download_file( # erreichbar): 400 mit Klartext statt eines unbehandelten 500. logger.warning("Dateitransfer fuer Host %s nicht moeglich: %s", host_id, exc) raise HTTPException(status.HTTP_400_BAD_REQUEST, describe_connection_error(exc)) + except asyncssh.Error as exc: + # Siehe ausfuehrlicher Kommentar in upload_file() weiter oben -- gleicher + # Bugfix: SFTP-Fehler nach erfolgreicher Anmeldung (z.B. Datei nicht + # gefunden, keine Leseberechtigung) liefen bisher unbehandelt bis zu + # Starlettes Klartext-500 durch. + logger.warning("SFTP-Fehler bei Host %s: %s", host_id, exc) + raise HTTPException(status.HTTP_400_BAD_REQUEST, f"SFTP-Fehler: {exc}") + except HTTPException: + raise + except Exception: + logger.exception("Unerwarteter Fehler beim Datei-Download fuer Host %s", host_id) + raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, "Unerwarteter Fehler beim Dateitransfer") sha256 = hashlib.sha256(data).hexdigest() filename = remote_path.rsplit("/", 1)[-1] await _log_transfer( conn, user=user, host_id=host_id, client_ip=_client_ip(request), direction="download", - filename=filename, size=len(data), sha256=sha256, av_scan_result="not_applicable_download", + filename=filename, size=len(data), sha256=sha256, av_result="not_applicable_download", ) def _iter(): diff --git a/app/ssh_proxy/terminal_ws.py b/app/ssh_proxy/terminal_ws.py index 798dd6c..a78b193 100644 --- a/app/ssh_proxy/terminal_ws.py +++ b/app/ssh_proxy/terminal_ws.py @@ -56,7 +56,9 @@ async def _reject(websocket: WebSocket, code: int, reason: str, *, accepted: boo await websocket.close(code=code, reason=reason_bytes.decode("utf-8", errors="ignore")) -async def _pump_ssh_to_ws(process: asyncssh.SSHClientProcess, websocket: WebSocket, recorder: SessionRecorder): +async def _pump_ssh_to_ws( + process: asyncssh.SSHClientProcess, websocket: WebSocket, recorder: SessionRecorder, session_id: int +): try: while True: data = await process.stdout.read(65536) @@ -65,6 +67,11 @@ async def _pump_ssh_to_ws(process: asyncssh.SSHClientProcess, websocket: WebSock if isinstance(data, str): data = data.encode("utf-8", errors="replace") recorder.record("output", base64.b64encode(data).decode()) + # Live-Mitschau (GET /ws/sessions/{id}/watch, siehe unten): jeder + # Chunk geht zusaetzlich an alle aktuell zuschauenden Superadmins. + # Best-effort/read-only -- ein Beobachter kann diese Sitzung nicht + # beeinflussen, und sein Fehlen/Trennen hat keinen Einfluss hierauf. + active_sessions.broadcast(session_id, data) await websocket.send_json({"type": "output", "data": base64.b64encode(data).decode()}) except (asyncssh.Error, ConnectionResetError): pass @@ -133,7 +140,7 @@ async def ssh_terminal(websocket: WebSocket, host_id: int): ssh_conn = await connect_to_host(conn, host_id) logger.debug("SSH-Sitzung %s: Verbindung zu %s hergestellt", session_id, host["hostname"]) process = await ssh_conn.create_process(term_type="xterm-256color") - pump_task = asyncio.create_task(_pump_ssh_to_ws(process, websocket, recorder)) + pump_task = asyncio.create_task(_pump_ssh_to_ws(process, websocket, recorder, session_id)) while True: try: @@ -237,3 +244,60 @@ async def ssh_terminal(websocket: WebSocket, host_id: int): await websocket.close() except Exception: logger.debug("WebSocket war beim Schliessen bereits getrennt", exc_info=True) + + +@router.websocket("/ws/sessions/{session_id}/watch") +async def watch_ssh_session(websocket: WebSocket, session_id: int): + """Live-Mitschau einer laufenden SSH-Sitzung fuer Superadmins ('Aktive + Session mitschauen', Sessionview im Adminbereich). + + Rein lesend: der Beobachter bekommt exakt dieselben "output"-Frames wie + der Sitzungsinhaber selbst (siehe _pump_ssh_to_ws()/active_sessions. + broadcast()), kann aber keine Eingaben in die fremde Sitzung schicken -- + dafuer gibt es hier bewusst keinen "input"-Pfad. Eigener Top-Level-Router + unter /ws/... (siehe app/admin/log_ws.py fuer die ausfuehrliche + Begruendung: der nginx-Reverse-Proxy setzt WebSocket-Upgrade-Header nur + fuer /ws/-Pfade). Nur auf demselben Server-Prozess moeglich wie + 'Beenden' -- active_sessions.add_watcher() gibt None zurueck, wenn die + Sitzung hier nicht (mehr) laeuft.""" + user = await get_current_user_ws(websocket) + if user is None or not user.is_admin: + await websocket.close(code=4403) + return + + conn = get_db() + row = await (await conn.execute( + "SELECT protocol, ended_at FROM sessions WHERE id = ?", (session_id,) + )).fetchone() + if row is None or row[0] != "ssh" or row[1] is not None: + await websocket.close(code=4404) + return + + queue = active_sessions.add_watcher(session_id) + if queue is None: + # Sitzung existiert, laeuft aber nicht (mehr) auf diesem Worker- + # Prozess -- analog zum 'killable'-Feld bei 'Beenden'. + await websocket.close(code=4409) + return + + await websocket.accept() + client_ip = websocket.client.host if websocket.client else "unknown" + await write_audit_event( + conn, event_type="session_watch_started", user_id=user.id, client_ip=client_ip, + details={"session_id": session_id}, + ) + await conn.commit() + logger.debug("Superadmin %s beobachtet SSH-Sitzung %s live", user.username, session_id) + + try: + while True: + data = await queue.get() + await websocket.send_json({"type": "output", "data": base64.b64encode(data).decode()}) + except WebSocketDisconnect: + pass + finally: + active_sessions.remove_watcher(session_id, queue) + try: + await websocket.close() + except Exception: + logger.debug("Watch-WebSocket war beim Schliessen bereits getrennt", exc_info=True) diff --git a/static/js/admin.js b/static/js/admin.js index a1df2cb..4cc162d 100644 --- a/static/js/admin.js +++ b/static/js/admin.js @@ -1364,6 +1364,11 @@ } else if (s.is_active) { actions.appendChild(el("span", { className: "hint", textContent: "anderer Prozess" })); } + if (s.watchable) { + actions.appendChild( + actionButton("Live ansehen", "btn-secondary", () => watchSession(s)) + ); + } if (s.has_recording) { actions.appendChild( actionButton("Wiedergabe", "btn-secondary", () => showSessionRecording(s)) @@ -1388,6 +1393,75 @@ ); } + // --- Live-Mitschau einer aktiven SSH-Sitzung (nur lesend) -------------- + // + // Verbindet sich mit GET /ws/sessions/{id}/watch (app/ssh_proxy/ + // terminal_ws.py) und schreibt jedes "output"-Frame in ein eigenes, + // schreibgeschuetztes xterm.js-Terminal -- exakt dieselben Frames, die + // auch der Sitzungsinhaber selbst bekommt (active_sessions.broadcast()). + // Es gibt hier bewusst KEINEN Eingabepfad: ein Beobachter kann die + // fremde Sitzung nicht beeinflussen. + + let activeWatch = null; + + function closeWatchOverlay() { + document.getElementById("watch-overlay").classList.add("hidden"); + if (activeWatch) { + activeWatch.destroy(); + activeWatch = null; + } + document.getElementById("watch-stage").replaceChildren(); + } + + document.getElementById("watch-close-btn").addEventListener("click", closeWatchOverlay); + + function watchSession(session) { + closeWatchOverlay(); + const overlay = document.getElementById("watch-overlay"); + const title = document.getElementById("watch-title"); + const statusEl = document.getElementById("watch-status"); + const stage = document.getElementById("watch-stage"); + overlay.classList.remove("hidden"); + title.textContent = `Live-Ansicht -- Sitzung #${session.id} (${session.username} @ ${session.hostname})`; + statusEl.textContent = "verbinde ..."; + + const container = el("div", { className: "rec-terminal" }); + stage.appendChild(container); + const term = new Terminal({ + cursorBlink: false, disableStdin: true, convertEol: false, + fontFamily: "Menlo, Consolas, monospace", fontSize: 13, + theme: { background: "#000000" }, + }); + const fitAddon = new FitAddon.FitAddon(); + term.loadAddon(fitAddon); + term.open(container); + fitAddon.fit(); + + const proto = window.location.protocol === "https:" ? "wss:" : "ws:"; + const ws = new WebSocket(`${proto}//${window.location.host}/ws/sessions/${session.id}/watch`); + + ws.addEventListener("open", () => { statusEl.textContent = "verbunden (nur lesend)"; }); + ws.addEventListener("close", (ev) => { + statusEl.textContent = ev.reason + ? `getrennt: ${ev.reason}` + : "getrennt (Sitzung vermutlich beendet)"; + }); + ws.addEventListener("error", () => { statusEl.textContent = "Verbindungsfehler"; }); + ws.addEventListener("message", (ev) => { + const msg = JSON.parse(ev.data); + if (msg.type === "output") { + term.write(recB64decode(msg.data)); + } + }); + + activeWatch = { + destroy() { + try { ws.close(); } catch (_) { /* bereits getrennt */ } + term.dispose(); + }, + }; + } + // --- Wiedergabe (SSH: Terminal-Replay ueber xterm.js; RDP: volle // grafische Wiedergabe ueber Guacamole.SessionRecording) -------------- // @@ -1400,8 +1474,28 @@ // die "output"-Eintraege deshalb exakt das Format, das die Bibliothek // erwartet -- kein eigener Player noetig. + // Bugfix ("SSH-Wiedergabe funktioniert nicht"): NICHT als JS-String ueber + // decodeURIComponent(escape(atob(...))) dekodieren. Jeder aufgezeichnete + // "output"-Eintrag entspricht genau einem process.stdout.read()-Chunk auf + // dem Server (app/ssh_proxy/terminal_ws.py) -- ein Mehrbyte-UTF-8-Zeichen + // (Umlaute/Sonderzeichen in Banner, Prompt, htop, ls --color etc.) kann + // dabei GENAU an der Chunk-Grenze zerschnitten sein. decodeURIComponent() + // wirft dann "URI malformed" fuer diesen Eintrag -- und weil term.write() + // in scheduleNext()/seekTo() innerhalb eines try-losen setTimeout-Callbacks + // steht, bricht genau dort die gesamte Wiedergabe-Kette endgueltig ab + // (idx wird nie mehr erhoeht, scheduleNext() nie mehr aufgerufen): die + // Wiedergabe blieb an der ersten betroffenen Stelle fuer immer haengen, + // oft schon beim Login-Banner. term.write() nimmt daher direkt die rohen + // Bytes als Uint8Array entgegen -- xterm.js ist ein echter + // Terminal-Emulator mit eigenem, chunk-uebergreifend korrektem + // UTF-8-Parser, genau fuer diesen Fall gebaut (siehe auch + // static/js/terminal.js::b64decodeBytes fuer dieselbe Klasse Bug in der + // Live-Ansicht). function recB64decode(b64) { - return decodeURIComponent(escape(atob(b64))); + const bin = atob(b64); + const bytes = new Uint8Array(bin.length); + for (let i = 0; i < bin.length; i++) bytes[i] = bin.charCodeAt(i); + return bytes; } function formatMs(ms) { diff --git a/static/js/dashboard.js b/static/js/dashboard.js index e7b303b..9bf5e91 100644 --- a/static/js/dashboard.js +++ b/static/js/dashboard.js @@ -54,6 +54,11 @@ const connect = document.createElement("a"); connect.href = host.protocol === "ssh" ? `/terminal/${host.id}` : `/rdp/${host.id}`; connect.textContent = "Verbinden"; + // Oeffnet die Sitzung in einem eigenen Tab statt im Dashboard-Tab zu + // navigieren -- das Dashboard bleibt dadurch erreichbar, waehrend die + // Sitzung laeuft, und der Exit-Button der Sitzung (session-toolbar in + // terminal.html/rdp.html) kann genau diesen Tab wieder schliessen. + connect.target = "_blank"; actions.appendChild(connect); if (host.can_view_credentials) { diff --git a/static/js/rdp.js b/static/js/rdp.js index 10f20b1..cc692fa 100644 --- a/static/js/rdp.js +++ b/static/js/rdp.js @@ -107,4 +107,13 @@ for (const keysym of keys) client.sendKeyEvent(1, keysym); for (const keysym of keys.slice().reverse()) client.sendKeyEvent(0, keysym); }); + + document.getElementById("exit-btn").addEventListener("click", () => { + try { client.disconnect(); } catch (_) { /* bereits getrennt */ } + window.close(); + // Siehe terminal.js fuer die ausfuehrliche Begruendung: window.close() + // schliesst den Tab nur unter bestimmten Voraussetzungen lautlos -- + // deshalb zusaetzlich immer ein Fallback zum Dashboard. + window.setTimeout(() => { window.location.href = "/dashboard"; }, 300); + }); })(); diff --git a/static/js/terminal.js b/static/js/terminal.js index ad589ba..fe0f666 100644 --- a/static/js/terminal.js +++ b/static/js/terminal.js @@ -206,8 +206,16 @@ `/ssh/${hostId}/files/upload?remote_path=${encodeURIComponent(remotePath)}`, { method: "POST", credentials: "same-origin", body: formData } ); - const data = await res.json(); - if (!res.ok) throw new Error(data.detail || "Upload fehlgeschlagen"); + // Robust gegen Nicht-JSON-Antworten (z.B. ein unbehandelter serverseitiger + // Fehler, den Starlette als Klartext "Internal Server Error" statt JSON + // ausliefert) -- sonst crasht hier res.json() mit einer verwirrenden + // "Unexpected token 'I' ..."-Meldung statt der eigentlichen Fehlerursache + // (siehe GET .../files/download, das dasselbe Muster schon nutzt). + let data = null; + try { + data = await res.json(); + } catch (_) { /* Antwort war kein JSON */ } + if (!res.ok) throw new Error((data && data.detail) || `Upload fehlgeschlagen (HTTP ${res.status})`); ftStatus.textContent = `Upload abgeschlossen (AV: ${data.av_scan_result})`; recordTransfer("upload", file.name, data.size, true, ""); uploadFileInput.value = ""; diff --git a/templates/admin.html b/templates/admin.html index 480c415..7b82698 100644 --- a/templates/admin.html +++ b/templates/admin.html @@ -752,6 +752,18 @@ + + + diff --git a/templates/rdp.html b/templates/rdp.html index ba7201b..fdb2961 100644 --- a/templates/rdp.html +++ b/templates/rdp.html @@ -13,6 +13,7 @@ Verbinde ... +
diff --git a/templates/terminal.html b/templates/terminal.html index e7f91fa..8567a86 100644 --- a/templates/terminal.html +++ b/templates/terminal.html @@ -14,6 +14,7 @@ Verbinde ... +
diff --git a/tests/test_phase15.py b/tests/test_phase15.py new file mode 100644 index 0000000..b2fd3a0 --- /dev/null +++ b/tests/test_phase15.py @@ -0,0 +1,415 @@ +""" +Tests fuer Phase 15 (2026-08-21) -- Dateitransfer-500er, SSH-Wiedergabe- +Haenger und Live-Mitschau aktiver Sitzungen. + +Drei voneinander unabhaengige Bugs/Features, gemeldet in derselben Session: + +1) Dateitransfer-Upload/-Download endete bei einem Fehler als nackter + Klartext-500 ("Internal Server Error") statt JSON -- das Frontend crashte + beim res.json() mit "Unexpected token 'I' ...". Zwei voneinander + UNABHAENGIGE Ursachen in app/ssh_proxy/sftp.py: + a) _log_transfer() wurde mit dem falschen Keyword-Argumentnamen + aufgerufen (av_scan_result= statt av_result=) -- das feuerte bei + JEDEM erfolgreichen Transfer, ausserhalb jedes try/except. + b) except SSH_SETUP_ERRORS deckte nur Fehler VOR der Anmeldung ab; + asyncssh.Error (echter Verbindungsfehler waehrend asyncssh.connect(), + oder ein SFTP-Fehler NACH der Anmeldung wie SFTPPermissionDenied) + lief unbehandelt durch. + Siehe project memory: dateitransfer_upload_500_phase15.md. + +2) "SSH-Wiedergabe funktioniert nicht": static/js/admin.js dekodierte jeden + aufgezeichneten output-Chunk einzeln ueber + decodeURIComponent(escape(atob(...))) als JS-String. Ein Mehrbyte-UTF-8- + Zeichen (Umlaute etc.), das GENAU an einer process.stdout.read()- + Chunk-Grenze zerschnitten wurde, liess das an dieser Stelle mit + "URI malformed" abstuerzen -- innerhalb eines try-losen setTimeout, was + die gesamte Wiedergabe fuer immer haengen liess. Reiner JS-Bug (kein + Python-Aequivalent hier direkt testbar) -- separat mit einem Node-Skript + verifiziert (siehe Session-Notizen); dieser Testfall sichert stattdessen + die BACKEND-Seite ab, die diesen Chunk ueberhaupt erst erzeugt. + +3) "Aktive Session kann man nicht mitschauen": neue read-only Live-Mitschau + fuer Superadmins, GET /ws/sessions/{id}/watch (app/ssh_proxy/ + terminal_ws.py) + app/security/active_sessions.py (Beobachter-Queues pro + Sitzung, dieselbe best-effort/put_nowait-Philosophie wie log_stream.py). +""" +from __future__ import annotations + +import base64 + +import asyncssh +import pytest +from fastapi import HTTPException, WebSocketDisconnect + +from app.security import active_sessions +from app.ssh_proxy import sftp as sftp_module +from app.ssh_proxy import terminal_ws as terminal_ws_module + + +class _FakeUser: + def __init__(self, *, is_admin=True, id=1, username="admin"): + self.is_admin = is_admin + self.id = id + self.username = username + + +class _FakeConn: + """Minimaler aiosqlite-Ersatz fuer diese Tests -- genug fuer die + INSERT/UPDATE/SELECT-Aufrufe aus sftp.py/terminal_ws.py, ohne echte + SQLite-Semantik nachzubilden (das leisten test_phase8/9/12/13 bereits + fuer die jeweils betroffenen Endpunkte).""" + + def __init__(self, select_result=None): + self._select_result = select_result + + async def execute(self, sql, params=()): + class _Cursor: + lastrowid = 1 + + async def fetchone(self_inner): + return self._select_result + + return _Cursor() + + async def commit(self): + pass + + +# -------------------------------------------------------------------------- +# 1a) Regression: _log_transfer() muss mit dem PARAMETERNAMEN aufgerufen +# werden, den es tatsaechlich definiert (av_result), nicht mit einem +# Namen, der zufaellig plausibel klingt (av_scan_result). Ein erfolg- +# reicher Transfer darf NIE eine TypeError werfen. +# -------------------------------------------------------------------------- + +async def test_upload_erfolgreicher_transfer_wirft_keine_typeerror(monkeypatch): + fake_conn = _FakeConn() + monkeypatch.setattr(sftp_module, "get_db", lambda: fake_conn) + + async def fake_write_audit_event(*a, **kw): + pass + + monkeypatch.setattr(sftp_module, "write_audit_event", fake_write_audit_event) + + class _OkRemoteFile: + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + async def write(self, data): + self.written = data + + class _OkSftp: + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + def open(self, path, mode): + return _OkRemoteFile() + + class _OkSshConn: + def start_sftp_client(self): + return _OkSftp() + + def close(self): + pass + + async def fake_connect_to_host(conn, host_id): + return _OkSshConn() + + monkeypatch.setattr(sftp_module, "connect_to_host", fake_connect_to_host) + + request = sftp_module.Request() + upload = sftp_module.UploadFile("test.txt", b"hallo welt") + + result = await sftp_module.upload_file( + host_id=1, request=request, remote_path="/tmp/test.txt", file=upload, + host={"file_transfer_enabled": True}, user=_FakeUser(), + ) + assert result["status"] == "ok" + assert result["size"] == len(b"hallo welt") + + +# -------------------------------------------------------------------------- +# 1b) asyncssh.Error (Verbindungsfehler NACH SSH_SETUP_ERRORS-Pruefung, oder +# ein SFTP-Fehler nach erfolgreicher Anmeldung) muss als HTTPException +# ankommen -- NIE unbehandelt durchlaufen. +# -------------------------------------------------------------------------- + +async def test_upload_asyncssh_error_wird_zu_http_exception_400(monkeypatch): + fake_conn = _FakeConn() + monkeypatch.setattr(sftp_module, "get_db", lambda: fake_conn) + + async def fake_connect_to_host(conn, host_id): + raise asyncssh.Error(reason="Connection refused") + + monkeypatch.setattr(sftp_module, "connect_to_host", fake_connect_to_host) + + request = sftp_module.Request() + upload = sftp_module.UploadFile("test.txt", b"hallo welt") + + with pytest.raises(HTTPException) as excinfo: + await sftp_module.upload_file( + host_id=1, request=request, remote_path="/tmp/test.txt", file=upload, + host={"file_transfer_enabled": True}, user=_FakeUser(), + ) + assert excinfo.value.status_code == 400 + + +async def test_upload_sftp_permission_denied_wird_zu_http_exception_400(monkeypatch): + """SFTP-Fehler NACH erfolgreicher Anmeldung (z.B. Zielverzeichnis nicht + vorhanden/keine Berechtigung) sind asyncssh.SFTPError -- eine Unterklasse + von asyncssh.Error, KEINE der SSH_SETUP_ERRORS.""" + fake_conn = _FakeConn() + monkeypatch.setattr(sftp_module, "get_db", lambda: fake_conn) + + class _DeniedSftp: + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + def open(self, path, mode): + raise asyncssh.SFTPPermissionDenied(reason="Permission denied") + + class _SshConn: + def start_sftp_client(self): + return _DeniedSftp() + + def close(self): + pass + + async def fake_connect_to_host(conn, host_id): + return _SshConn() + + monkeypatch.setattr(sftp_module, "connect_to_host", fake_connect_to_host) + + request = sftp_module.Request() + upload = sftp_module.UploadFile("test.txt", b"hallo welt") + + with pytest.raises(HTTPException) as excinfo: + await sftp_module.upload_file( + host_id=1, request=request, remote_path="/root/blocked.txt", file=upload, + host={"file_transfer_enabled": True}, user=_FakeUser(), + ) + assert excinfo.value.status_code == 400 + + +async def test_upload_unerwarteter_fehler_wird_zu_http_exception_500_nicht_unbehandelt(monkeypatch): + """Letztes Auffangnetz: selbst ein voellig unverwandter Bug (hier + simuliert per AttributeError) darf NIE unbehandelt bis Starlettes + Klartext-500 durchlaufen -- das Frontend braucht immer JSON.""" + fake_conn = _FakeConn() + monkeypatch.setattr(sftp_module, "get_db", lambda: fake_conn) + + async def fake_connect_to_host(conn, host_id): + raise AttributeError("simuliert einen unverwandten kuenftigen Bug") + + monkeypatch.setattr(sftp_module, "connect_to_host", fake_connect_to_host) + + request = sftp_module.Request() + upload = sftp_module.UploadFile("test.txt", b"hallo welt") + + with pytest.raises(HTTPException) as excinfo: + await sftp_module.upload_file( + host_id=1, request=request, remote_path="/tmp/test.txt", file=upload, + host={"file_transfer_enabled": True}, user=_FakeUser(), + ) + assert excinfo.value.status_code == 500 + + +async def test_download_datei_zu_gross_bleibt_413_und_wird_nicht_zu_500(monkeypatch): + """Regressionsschutz fuer den neuen Catch-all: die bereits bestehende + HTTPException(413) aus dem inneren try-Block darf NICHT vom neuen + 'except Exception' verschluckt und zu einem 500 umgemuenzt werden.""" + fake_conn = _FakeConn() + monkeypatch.setattr(sftp_module, "get_db", lambda: fake_conn) + + class _Stat: + size = sftp_module.MAX_UPLOAD_BYTES + 1 + + class _HugeSftp: + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + async def stat(self, path): + return _Stat() + + class _SshConn: + def start_sftp_client(self): + return _HugeSftp() + + def close(self): + pass + + async def fake_connect_to_host(conn, host_id): + return _SshConn() + + monkeypatch.setattr(sftp_module, "connect_to_host", fake_connect_to_host) + + request = sftp_module.Request() + with pytest.raises(HTTPException) as excinfo: + await sftp_module.download_file( + host_id=1, request=request, remote_path="/tmp/huge.bin", + host={"file_transfer_enabled": True}, user=_FakeUser(), + ) + assert excinfo.value.status_code == 413 + + +# -------------------------------------------------------------------------- +# 3) Live-Mitschau: app/security/active_sessions.py Beobachter-Queues. +# -------------------------------------------------------------------------- + +async def test_active_sessions_broadcast_erreicht_alle_beobachter(event_loop=None): + import asyncio + + task = asyncio.current_task() + active_sessions.register(session_id=4242, task=task) + try: + q1 = active_sessions.add_watcher(4242) + q2 = active_sessions.add_watcher(4242) + assert q1 is not None and q2 is not None and q1 is not q2 + + active_sessions.broadcast(4242, b"chunk-1") + assert q1.get_nowait() == b"chunk-1" + assert q2.get_nowait() == b"chunk-1" + + active_sessions.remove_watcher(4242, q1) + active_sessions.broadcast(4242, b"chunk-2") + assert q2.get_nowait() == b"chunk-2" + assert q1.empty() + finally: + active_sessions.unregister(4242) + + +async def test_active_sessions_add_watcher_auf_unbekannter_sitzung_gibt_none(): + assert active_sessions.get(999999) is None + assert active_sessions.add_watcher(999999) is None + + +# -------------------------------------------------------------------------- +# 3) Live-Mitschau: WS-Route -- Zugriffskontrolle + Frame-Weiterleitung. +# -------------------------------------------------------------------------- + +class _FakeWebSocket: + def __init__(self): + self.accepted = False + self.sent = [] + self.closes = [] + self.client = type("C", (), {"host": "10.0.0.5"})() + + async def accept(self): + self.accepted = True + + async def send_json(self, obj): + self.sent.append(obj) + + async def close(self, code=1000, reason=""): + self.closes.append((code, reason)) + + +async def test_watch_ssh_session_lehnt_nicht_admin_ab(monkeypatch): + async def as_normal_user(ws): + return _FakeUser(is_admin=False) + + monkeypatch.setattr(terminal_ws_module, "get_current_user_ws", as_normal_user) + + ws = _FakeWebSocket() + await terminal_ws_module.watch_ssh_session(ws, 1) + assert ws.closes == [(4403, "")] + assert ws.accepted is False + + +async def test_watch_ssh_session_lehnt_beendete_sitzung_ab(monkeypatch): + async def as_admin(ws): + return _FakeUser(is_admin=True) + + monkeypatch.setattr(terminal_ws_module, "get_current_user_ws", as_admin) + monkeypatch.setattr( + terminal_ws_module, "get_db", + lambda: _FakeConn(select_result=("ssh", "2026-08-21T10:00:00Z")), + ) + + ws = _FakeWebSocket() + await terminal_ws_module.watch_ssh_session(ws, 1) + assert ws.closes == [(4404, "")] + assert ws.accepted is False + + +async def test_watch_ssh_session_lehnt_rdp_ab(monkeypatch): + async def as_admin(ws): + return _FakeUser(is_admin=True) + + monkeypatch.setattr(terminal_ws_module, "get_current_user_ws", as_admin) + monkeypatch.setattr( + terminal_ws_module, "get_db", lambda: _FakeConn(select_result=("rdp", None)) + ) + + ws = _FakeWebSocket() + await terminal_ws_module.watch_ssh_session(ws, 1) + assert ws.closes == [(4404, "")] + assert ws.accepted is False + + +async def test_watch_ssh_session_leitet_frames_weiter_und_raeumt_beobachter_auf(monkeypatch): + import asyncio + + async def as_admin(ws): + return _FakeUser(is_admin=True) + + monkeypatch.setattr(terminal_ws_module, "get_current_user_ws", as_admin) + monkeypatch.setattr( + terminal_ws_module, "get_db", lambda: _FakeConn(select_result=("ssh", None)) + ) + + audit_calls = [] + + async def fake_write_audit_event(conn, **kw): + audit_calls.append(kw) + + monkeypatch.setattr(terminal_ws_module, "write_audit_event", fake_write_audit_event) + + active_sessions.register(session_id=1, task=asyncio.current_task()) + try: + class _WsThenDisconnect(_FakeWebSocket): + def __init__(self): + super().__init__() + self._n = 0 + + async def send_json(self, obj): + await super().send_json(obj) + self._n += 1 + if self._n >= 2: + raise WebSocketDisconnect() + + ws = _WsThenDisconnect() + + async def feeder(): + await asyncio.sleep(0.01) + active_sessions.broadcast(1, b"frame-1") + await asyncio.sleep(0.01) + active_sessions.broadcast(1, b"frame-2") + + feeder_task = asyncio.create_task(feeder()) + await terminal_ws_module.watch_ssh_session(ws, 1) + await feeder_task + + assert ws.accepted is True + assert [f["data"] for f in ws.sent] == [ + base64.b64encode(b"frame-1").decode(), + base64.b64encode(b"frame-2").decode(), + ] + # Beobachter muss nach dem Trennen entfernt sein (kein Leak). + assert len(active_sessions.get(1).watchers) == 0 + assert len(audit_calls) == 1 + assert audit_calls[0]["event_type"] == "session_watch_started" + finally: + active_sessions.unregister(1)