feat: Störungswache (pipe.monitor) + Verschlüsselung-at-rest dokumentiert

Dienste-Ampel (Telefon, Verarbeitung, Spracherkennung, Nextcloud) alarmierte
bisher nur im offenen Leitstand. pipe/monitor.py laeuft jetzt als eigener
Dauerprozess, prueft dieselbe Ampel (neu ausgelagert nach pipe/dienste.py)
periodisch und alarmiert bei Stoerung lokal per macOS-Benachrichtigung + Ton
(pipe/alarm.py) - unabhaengig davon, ob pipe.watch oder pipe.server laufen.
Wird ab jetzt von telefon/starten.sh mitgestartet.

Verschluesselung at rest: keine App-eigene Verschluesselung eingebaut,
stattdessen im README als Betreiber-Pflicht dokumentiert (FileVault aktivieren)
- App verlaesst sich auf Festplattenverschluesselung durch das Betriebssystem.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EbdWYnDtyMgxsq4eN1WnrQ
This commit is contained in:
Jeuner 2026-08-28 16:47:45 +02:00
parent be0252c907
commit 6e5fe59c53
8 changed files with 266 additions and 90 deletions

View file

@ -13,101 +13,15 @@ import base64
import hmac
import json
import re
import subprocess
import sys
import time
import urllib.error
import urllib.request
from datetime import datetime
from http.cookies import SimpleCookie
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, unquote, urlparse
from . import bewertung, config, dashboard, export, modellwahl, protokoll, stapel, store, testzugang
# Statusabfragen sind teuer (Netz, Unterprozesse) - kurz zwischenspeichern,
# damit haeufiges Nachfragen den Rechner nicht belastet.
_CACHE: dict[str, tuple[float, dict]] = {}
CACHE_S = 4.0
def _laeuft(muster: str) -> bool:
try:
return subprocess.run(["pgrep", "-f", muster], capture_output=True).returncode == 0
except Exception:
return False
def _http_ok(url: str, timeout: float = 5.0) -> bool:
"""Erreichbarkeitstest. Mit eigenem User-Agent - ein WAF vor Nextcloud
blockt den Standardnamen von urllib und meldete fälschlich 'nicht
erreichbar'."""
req = urllib.request.Request(url, headers={"User-Agent": "praxis-telefon-agent/1.0"})
try:
with urllib.request.urlopen(req, timeout=timeout, context=store._ssl_kontext()):
return True
except urllib.error.HTTPError:
return True # antwortet - reicht als Lebenszeichen
except Exception:
return False
def _trunk() -> tuple[str, str]:
"""(zustand, text) des SIP-Trunks."""
if not _laeuft("[f]reeswitch"):
return "aus", "FreeSWITCH läuft nicht"
try:
roh = subprocess.run(
["fs_cli", "-P", config.FS_PORT, "-x", "sofia status gateway plusnet"],
capture_output=True, text=True, timeout=8).stdout
for zeile in roh.splitlines():
if zeile.startswith("Status"):
wert = zeile.split()[-1]
return ("gut", "registriert") if wert == "UP" else ("schlecht", f"Trunk {wert}")
return "unklar", "Gateway unbekannt"
except Exception:
return "unklar", "fs_cli nicht erreichbar"
def status() -> dict:
jetzt = time.time()
gepuffert = _CACHE.get("status")
if gepuffert and jetzt - gepuffert[0] < CACHE_S:
return gepuffert[1]
trunk_zustand, trunk_text = _trunk()
watcher = _laeuft("[p]ipe.watch")
ollama = _http_ok(f"{config.OLLAMA_URL}/api/version")
nextcloud = (_http_ok(f"{config.NEXTCLOUD_URL}/status.php")
if config.nextcloud_aktiv() else None)
offen = 0
if config.TELEFON_EINGANG.is_dir():
offen = sum(1 for d in config.TELEFON_EINGANG.iterdir()
if d.is_file() and not d.name.startswith("."))
fehler = 0
if config.TELEFON_FEHLER.is_dir():
fehler = sum(1 for d in config.TELEFON_FEHLER.iterdir() if d.is_file())
daten = {
"dienste": [
{"name": "Telefon", "zustand": trunk_zustand, "text": trunk_text},
{"name": "Verarbeitung", "zustand": "gut" if watcher else "aus",
"text": "beobachtet Eingang" if watcher else "Watcher läuft nicht"},
{"name": "Spracherkennung", "zustand": "gut" if ollama else "schlecht",
"text": "Ollama bereit" if ollama else "Ollama nicht erreichbar"},
{"name": "Nextcloud", "zustand": "unklar" if nextcloud is None
else ("gut" if nextcloud else "schlecht"),
"text": "nicht konfiguriert" if nextcloud is None
else ("erreichbar" if nextcloud else "nicht erreichbar")},
],
"eingang_offen": offen,
"fehler": fehler,
"stand": datetime.now().strftime("%H:%M:%S"),
}
_CACHE["status"] = (jetzt, daten)
return daten
from . import bewertung, config, dashboard, dienste, export, modellwahl, protokoll, stapel, testzugang
def anrufe() -> list[dict]:
@ -325,7 +239,7 @@ class Handler(BaseHTTPRequestHandler):
if pfad == "/":
self._sende(SEITE.encode("utf-8"), "text/html; charset=utf-8")
elif pfad == "/api/status":
daten = status()
daten = dienste.status()
# Testzugang (per "agent-one test"-Link) -> Sicher-Modus im
# Frontend erzwingen, siehe sicherSetzen() im JS.
daten["testzugang"] = getattr(self, "_via_token", False)