# -*- coding: utf-8 -*- """RC V7 (real GUI Runde 2): A) WebView-SSO blieb FAIL, weil ein VERALTETER Proxy-Prozess (alter Code im Speicher) denselben Port bediente; SO_REUSEADDR erlaubte unter Windows ein Co-Binding. Fix: kein Co-Bind (_pick_free_port SO_EXCLUSIVEADDRUSE + _TestProxyServer.allow_reuse_address=False auf Windows); Startskript beendet veraltete --serve-Proxys; secret-freier SSO-Diagnoselog (nur Cookie-NAMEN). HINWEIS: Diese mechanischen Tests sind KEIN Beweis fuer WebView-SSO. Der verbindliche Beweis ist der reale GUI-Lauf (siehe Abschlussbericht). B) Mini-Diktat hatte keinen Stop-Pfad. Fix: "Weiterfahren" wirkt waehrend der Aufnahme als "Stoppen" (bestehender toggle_diktat-Stop/Finalize); der Stop setzt _diktat_neu_busy und setzt es IMMER (finally) zurueck. """ from __future__ import annotations import inspect import json import os import re import socket import sys import threading import unittest import urllib.error import urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path ROOT = Path(__file__).resolve().parent _SESSION_VALUE = "sess-rcv7r2-secret-value" _USER = {"user_id": "u-andre-001", "practice_id": "p-lindengut-007", "display_name": "Andre M. Surovy"} _used = set() _used_lock = threading.Lock() def _func_body(src: str, name: str) -> str: m = re.search(r"\n def " + re.escape(name) + r"\(", src) if not m: m = re.search(r"\n def " + re.escape(name) + r"\(", src) if not m: return "" start = m.start() rest = src[start + 1:] nxt = re.search(r"\n def |\n def ", rest) end = (start + 1 + nxt.start()) if nxt else len(src) return src[start:end] class _FakeUpstream(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def log_message(self, *a): pass def _json(self, code, payload): b = json.dumps(payload).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(b))) self.end_headers() self.wfile.write(b) def do_GET(self): if self.path.startswith("/empfang/shell/launch"): tok = self.path.split("token=", 1)[1].split("&", 1)[0] if "token=" in self.path else "" with _used_lock: if not tok or tok in _used: self._json(401, {"detail": "verbraucht"}) return _used.add(tok) self.send_response(302) self.send_header("Location", "/empfang/?empfang_chat_shell=1&shell_source=empfang_chat_shell") self.send_header("Set-Cookie", f"aza_session={_SESSION_VALUE}; HttpOnly; Path=/; SameSite=Lax; Secure; Domain=api.aza-medwork.ch") self.send_header("Content-Length", "0") self.end_headers() return if self.path.startswith("/empfang/auth/me") or self.path.startswith("/empfang/shell/context/me"): ck = self.headers.get("Cookie") or "" if f"aza_session={_SESSION_VALUE}" in ck: self._json(200, {"authenticated": True, "success": True, **_USER}) else: self._json(401, {"detail": "nicht angemeldet"}) return self._json(404, {"detail": "nf"}) def _no_redirect_opener(): class _NR(urllib.request.HTTPRedirectHandler): def redirect_request(self, *a, **k): return None return urllib.request.build_opener(_NR) class ProxyAntiCoBindTests(unittest.TestCase): def setUp(self): os.environ["AZA_DOKU_PROMPT_TEST"] = "1" import aza_empfang_test_html_proxy as px self.px = px def test_server_no_reuse_addr_on_windows(self): self.assertEqual(self.px._TestProxyServer.allow_reuse_address, sys.platform != "win32") def test_pick_free_port_skips_occupied(self): occ = socket.socket(socket.AF_INET, socket.SOCK_STREAM) occ.bind(("127.0.0.1", self.px._PORT_RANGE_START)) occ.listen(1) try: port = self.px._pick_free_port() self.assertNotEqual(port, self.px._PORT_RANGE_START, "Proxy darf nicht an belegten Port co-binden") finally: occ.close() class ProxySSOMechanicalTests(unittest.TestCase): """Mechanisch (Fake-Upstream) — KEIN WebView-SSO-Beweis, nur Proxy-Verhalten.""" @classmethod def setUpClass(cls): os.environ["AZA_DOKU_PROMPT_TEST"] = "1" _used.clear() cls.up = ThreadingHTTPServer(("127.0.0.1", 0), _FakeUpstream) threading.Thread(target=cls.up.serve_forever, daemon=True).start() os.environ["AZA_EMPFANG_TEST_UPSTREAM"] = f"http://127.0.0.1:{cls.up.server_address[1]}" import aza_empfang_test_html_proxy as px cls.px = px cls.srv, cls.port = px._start_server_on_port(0) cls.base = f"http://127.0.0.1:{cls.port}" @classmethod def tearDownClass(cls): for s in (getattr(cls, "srv", None), getattr(cls, "up", None)): try: s.shutdown(); s.server_close() except Exception: pass def test_302_not_followed_cookie_localized(self): op = _no_redirect_opener() try: r = op.open(f"{self.base}/empfang/shell/launch?token=t-a&target=empfang_chat_shell", timeout=8) status, headers = r.status, r.headers except urllib.error.HTTPError as e: status, headers = e.code, e.headers self.assertEqual(status, 302) cookies = headers.get_all("Set-Cookie") or [] joined = " || ".join(cookies) self.assertIn("aza_session=", joined) self.assertNotIn("secure", joined.lower()) self.assertNotIn("domain=", joined.lower()) def test_cookie_roundtrip_auth_me_200(self): op = _no_redirect_opener() try: r = op.open(f"{self.base}/empfang/shell/launch?token=t-b&target=empfang_chat_shell", timeout=8) headers = r.headers except urllib.error.HTTPError as e: headers = e.headers sc = "" for c in headers.get_all("Set-Cookie") or []: if c.startswith("aza_session="): sc = c.split(";", 1)[0] self.assertTrue(sc) req = urllib.request.Request(f"{self.base}/empfang/auth/me", headers={"Cookie": sc}) with urllib.request.urlopen(req, timeout=8) as r2: self.assertEqual(r2.status, 200) d = json.loads(r2.read().decode("utf-8")) self.assertEqual(d.get("user_id"), _USER["user_id"]) self.assertEqual(d.get("practice_id"), _USER["practice_id"]) def test_sso_diag_logs_names_only_no_values(self): log = ROOT / ".aza_test_proxy_logs" / "sso_diag.log" try: if log.exists(): log.unlink() except Exception: pass op = _no_redirect_opener() try: op.open(f"{self.base}/empfang/shell/launch?token=t-diag&target=empfang_chat_shell", timeout=8) except urllib.error.HTTPError: pass self.assertTrue(log.exists(), "Diagnoselog fehlt") content = log.read_text(encoding="utf-8", errors="replace") self.assertIn("set_cookie=aza_session", content) # nur Name self.assertIn("status=302", content) self.assertNotIn(_SESSION_VALUE, content) # niemals Wert class MiniDiktatStopCycleTests(unittest.TestCase): @classmethod def setUpClass(cls): import aza_mini_diktat_window as md import aza_diktat_mixin as dm cls.md_open = inspect.getsource(md.open_mini_diktat_window) cls.md_sync = inspect.getsource(md.sync_mini_diktat_window) cls.md_close = inspect.getsource(md.close_mini_diktat_window) cls.dm_src = inspect.getsource(dm.AzaDiktatMixin.open_diktat_window) def test_weiterfahren_stops_during_recording(self): body = _func_body(self.md_open, "_on_mini_stop") self.assertTrue(body) self.assertIn("_aza_diktat_toggle", body) self.assertIn("_diktat_neu_busy", body) sync_body = self.md_sync self.assertIn("command=stop_cmd", sync_body) def test_weiter_button_reference_stored(self): self.assertIn("_mini_diktat_weiter_btn", self.md_open) def test_sync_button_reflects_state(self): self.assertIn('"Stoppen"', self.md_sync) self.assertIn('"Weiterfahren"', self.md_sync) self.assertIn("_diktat_phase", self.md_sync) self.assertIn("command=stop_cmd", self.md_sync) def test_close_during_recording_stops(self): self.assertIn("_aza_diktat_toggle", self.md_close) self.assertIn("_diktat_phase", self.md_close) self.assertIn("_skip_stop", self.md_close) def test_toggle_stop_sets_and_resets_busy(self): self.assertIn("self._diktat_neu_busy = True", self.dm_src) self.assertIn("def _after_finalize", self.dm_src) self.assertIn("finally:", self.dm_src) self.assertIn("_safe_after(_after_finalize)", self.dm_src) def test_busy_guard_on_entrypoints(self): for name in ("toggle_diktat", "_diktat_weiterfahren", "_diktat_neu_von_logo", "do_neu"): body = _func_body(self.dm_src, name) self.assertTrue(body, f"{name} fehlt") self.assertIn("_diktat_neu_busy", body, f"{name} ohne Guard") if __name__ == "__main__": unittest.main()