connectiopn fix round 2 3

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,
+            "is_active": is_active,
-            "killable": r[0] in active_ids,
+            "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
 

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

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():

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)

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) {

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) {

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);
+  });
 })();

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();
+      // Robust gegen Nicht-JSON-Antworten (z.B. ein unbehandelter serverseitiger
-      if (!res.ok) throw new Error(data.detail || "Upload fehlgeschlagen");
+      // 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 = "";

templates/admin.html

@@ -752,6 +752,18 @@
     </div>
   </div>
 
+  <!-- ================= Aktive Session live ansehen (Overlay) ================= -->
+  <div class="ft-overlay hidden" id="watch-overlay">
+    <div class="ft-panel rec-panel">
+      <h2 id="watch-title">Live-Ansicht
+        <button type="button" class="btn-secondary btn-small" id="watch-close-btn">Schliessen</button>
+      </h2>
+      <p class="hint">Nur lesend -- Eingaben werden nicht an die Sitzung uebertragen.</p>
+      <p class="hint">Status: <span id="watch-status">verbinde ...</span></p>
+      <div id="watch-stage" class="rec-stage"></div>
+    </div>
+  </div>
+
   <script src="/static/js/vendor/xterm.js"></script>
   <script src="/static/js/vendor/xterm-addon-fit.js"></script>
   <script src="/static/js/vendor/guacamole-common.js"></script>

templates/rdp.html

@@ -13,6 +13,7 @@
       <button id="ctrlaltdel-btn">Strg+Alt+Entf</button>
       <span class="spacer"></span>
       <span class="status" id="status">Verbinde ...</span>
+      <button id="exit-btn" class="btn-danger btn-small" title="Sitzung beenden und Tab schliessen">Exit</button>
     </div>
     <div id="rdp-display"></div>
   </div>

templates/terminal.html

@@ -14,6 +14,7 @@
       <button id="filetransfer-btn">Dateitransfer</button>
       <span class="spacer"></span>
       <span class="status" id="status">Verbinde ...</span>
+      <button id="exit-btn" class="btn-danger btn-small" title="Sitzung beenden und Tab schliessen">Exit</button>
     </div>
     <div id="terminal"></div>
   </div>

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)