#!/usr/bin/env python3 """ Red Queen — PRE_HERMES_SECURITY_GATE AUTH.1 — Tolaria Write Auth Isolierte Test-Suite. Testet den Tolaria-Write-Auth-Contract (PRE_HERMES_TOLARIA_WRITE_AUTH_CONTRACT.md) ausschliesslich gegen ein isoliertes in-memory Fake-Fixture (FakeTolariaAuth). KEINE Verbindung zum produktiven Tolaria. KEINE produktiven Endpoints / POST-Probes / Mutationen. Nur synthetische Test-Credentials (NIE echte Tokens/Secrets). Abgedeckte Faelle (Mission PHASE 3-6): T1-T20 Auth-Matrix (READ/SAVE/RENAME/RENAME-FILENAME/DELETE, fail-closed) A-G Human-Approval-Composition (AUTH != APPROVAL, beide Ebenen unabhaengig noetig) ADV Adversarial (traversal, absolute path, malformed bearer, oversize, constant-time, body-token, cross-scope, replay, symlink-escape, leaks) SENS Test-Sensitivitaet (Mutationen der Enforcement-Logik -> Tests werden ROT) REGR Regression: bestehender C5-Verhalten (kein Produktionscode geaendert) Test-Credentials (synthetisch, KEINE echten Werte): SAVE_TOKEN = "auth1-save-synthetic-token-0000000000" DELETE_TOKEN = "auth1-delete-synthetic-token-0000000000" WRONG_TOKEN = "auth1-wrong-synthetic-token-9999999999" """ from __future__ import annotations import hmac import os import re import sys import tempfile import unittest from typing import Dict, List, Optional from urllib.parse import unquote sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) # --------------------------------------------------------------------------- # Synthetische Test-Credentials (NIE echte Werte) # --------------------------------------------------------------------------- SAVE_TOKEN = "auth1-save-synthetic-token-0000000000" DELETE_TOKEN = "auth1-delete-synthetic-token-0000000000" WRONG_TOKEN = "auth1-wrong-synthetic-token-9999999999" VAULT_ROOT = "/app/vault" # --------------------------------------------------------------------------- # Isoliertes Fake-Fixture: repliziert die AUTH-CONTRACT-Semantik des Servers. # Dies ist der REFERENZ-Enforcer fuer die Tests. In AUTH.2+ wird exakt diese # Logik produktiv implementiert; die Tests pruefen die CONTRACT-Regeln. # --------------------------------------------------------------------------- class TolariaAuthError(Exception): """Auth- oder Path-Safety-Fehler. Erzeugt NIE eine Mutation.""" def __init__(self, status: int, code: str, message: str): super().__init__(message) self.status = status self.code = code self.message = message class Approval: """Persistierte Human-Approval (C5 Invarianten A-N, ST_DELETE_APPROVED).""" def __init__(self, approval_id: str, workflow_commit: str, object_change_id: str, vault_path: str): self.approval_id = approval_id self.workflow_commit = workflow_commit self.object_change_id = object_change_id self.vault_path = vault_path self.used = False class FakeTolariaAuth: """ Isoliertes in-memory Fake, das die Auth-CONTRACT-Semantik erzwengt. Konfiguration: save_token / delete_token : erwartete Credentials (None = Token nicht konfiguriert) enforce_auth : False simuliert "Auth deaktiviert" (Sensitivitaet) enforce_scope_separation : False simuliert "ein gemeinsames Master-Token" enforce_fail_closed : False simuliert "fehlende fail-closed Rule" token_compare_always_true : False -> const-time; True simuliert "immer TRUE" secret_logging : True simuliert verbotenes Secret-Logging allow_path_traversal : False -> Path-Safety aktiv Mutationszaehler (save_count/delete_count) werden inkrementiert NUR bei erfolgreicher, autorisierter Mutation. Tests pruefen: DENIED => count unveraendert. """ def __init__(self, save_token: Optional[str] = SAVE_TOKEN, delete_token: Optional[str] = DELETE_TOKEN, enforce_auth: bool = True, enforce_scope_separation: bool = True, enforce_fail_closed: bool = True, token_compare_always_true: bool = False, secret_logging: bool = False, allow_path_traversal: bool = False): self.save_token = save_token self.delete_token = delete_token self.enforce_auth = enforce_auth self.enforce_scope_separation = enforce_scope_separation self.enforce_fail_closed = enforce_fail_closed self.token_compare_always_true = token_compare_always_true self.secret_logging = secret_logging self.allow_path_traversal = allow_path_traversal self.store: Dict[str, str] = {} # vault_path -> content self.save_count = 0 self.delete_count = 0 self.rename_count = 0 self.approvals: Dict[str, Approval] = {} self.log_entries: List[str] = [] # -- Logging-Helper (Secret-Rules) -- def _log(self, entry: str) -> None: self.log_entries.append(entry) # -- const-time Token-Vergleich -- def _tokens_match(self, provided: str, expected: Optional[str]) -> bool: if self.token_compare_always_true: return True if expected is None: return False return hmac.compare_digest(provided, expected) # -- Secret-Logging (nur als MUTATIONS-Flag fuer Sensitivitaetstests) -- def _log_auth_failure(self, header: Optional[str]) -> None: # Im korrekten System (secret_logging=False) wird der Header NIEMALS geloggt. # secret_logging=True simuliert die verbotene MUTATION "Auth-Fehler wird # mit Token geloggt" und wird von TestSensitivity nachgewiesen. if self.secret_logging and header is not None: self._log(f"auth fail header={header}") # -- Auth-Enforcement -- def _check_auth(self, header: Optional[str], scope: str, expected_token: Optional[str]) -> None: if not self.enforce_auth: return # simuliert "Auth deaktiviert" if not self.enforce_fail_closed: return # simuliert fehlende fail-closed rule if header is None: raise TolariaAuthError(401, "AUTH_MISSING", "missing authorization header") # nur Header-Auth, kein Body if not header.startswith("Bearer "): raise TolariaAuthError(400, "AUTH_MALFORMED", "malformed bearer") token = header[len("Bearer "):].strip() if not token: raise TolariaAuthError(400, "AUTH_EMPTY_TOKEN", "empty token") if len(token) > 128: raise TolariaAuthError(400, "AUTH_TOKEN_TOO_LARGE", "token too large") if not self.enforce_scope_separation: # ein gemeinsames Master-Token: beide Scopes teilen sich save_token scope_expected = self.save_token scope_ok = self._tokens_match(token, scope_expected) if not scope_ok: raise TolariaAuthError(401, "AUTH_INVALID", "invalid token") return # getrennte Scopes if scope == "SAVE": scope_expected = self.save_token elif scope == "DELETE": scope_expected = self.delete_token else: raise TolariaAuthError(500, "INTERNAL", "unknown scope") if self._tokens_match(token, scope_expected): return # PASS self._log_auth_failure(header) # nur bei secret_logging=True (Mutation) # falsches Token ODER falscher Scope (Token gehoert anderem Scope) if token == self.save_token or token == self.delete_token: raise TolariaAuthError(403, "AUTH_SCOPE", "token scope mismatch") raise TolariaAuthError(401, "AUTH_INVALID", "invalid token") # -- Path-Safety (Defense-in-Depth, Phase 2 Contract) -- def _normalize_vault_path(self, raw: str) -> str: # URL-decode, \ -> / ; collapse duplicate slashes; strip trailing p = unquote(raw) p = p.replace("\\", "/") p = re.sub(r"/{2,}", "/", p) p = p.rstrip("/") return p def _check_path(self, raw: str, *, must_exist: bool = False) -> str: if self.allow_path_traversal: return raw p = self._normalize_vault_path(raw) # absolute path must stay inside vault root if not p.startswith(VAULT_ROOT + "/") and p != VAULT_ROOT: raise TolariaAuthError(400, "PATH_OUTSIDE_VAULT", "path outside vault root") # .. traversal segments = p.split("/") if ".." in segments: raise TolariaAuthError(400, "PATH_TRAVERSAL", "path traversal blocked") # absolute foreign path (e.g. /etc) already rejected by vault-root check if must_exist and p not in self.store: raise TolariaAuthError(404, "PATH_NOT_EXISTS", "vault object does not exist") return p # -- Approval (Human Gate, Ebene B) -- def _check_delete_approval(self, vault_path: str, workflow_commit: str, object_change_id: str) -> None: for a in self.approvals.values(): if (a.vault_path == vault_path and a.workflow_commit == workflow_commit and a.object_change_id == object_change_id): if a.used: raise TolariaAuthError(403, "APPROVAL_USED", "approval already used") return # PASS raise TolariaAuthError(403, "APPROVAL_MISSING", "no matching human approval") def _mark_approval_used(self, vault_path: str, workflow_commit: str, object_change_id: str) -> None: for a in self.approvals.values(): if (a.vault_path == vault_path and a.workflow_commit == workflow_commit and a.object_change_id == object_change_id): a.used = True def add_approval(self, approval_id: str, workflow_commit: str, object_change_id: str, vault_path: str) -> None: self.approvals[approval_id] = Approval( approval_id, workflow_commit, object_change_id, vault_path) # -- Mutierende Endpoints -- def save(self, header: Optional[str], raw_path: str, content: str) -> str: self._check_auth(header, "SAVE", self.save_token) p = self._check_path(raw_path) self.store[p] = content self.save_count += 1 return p def rename(self, header: Optional[str], src_raw: str, dst_raw: str) -> str: self._check_auth(header, "SAVE", self.save_token) src = self._check_path(src_raw, must_exist=True) dst = self._check_path(dst_raw) self.store[dst] = self.store.pop(src) self.rename_count += 1 return dst def rename_filename(self, header: Optional[str], raw_path: str, new_name: str) -> str: self._check_auth(header, "SAVE", self.save_token) p = self._check_path(raw_path, must_exist=True) # new_name applied within same directory; validate target stays in vault parent = p.rsplit("/", 1)[0] dst = parent + "/" + new_name dst = self._check_path(dst) self.store[dst] = self.store.pop(p) self.rename_count += 1 return dst def delete(self, header: Optional[str], raw_path: str, *, workflow_commit: str, object_change_id: str) -> None: # Ebene A: technisches DELETE-Credential self._check_auth(header, "DELETE", self.delete_token) # Ebene B: Human Approval (unabhaengig, zwingend) self._check_delete_approval(raw_path, workflow_commit, object_change_id) p = self._check_path(raw_path, must_exist=True) del self.store[p] self.delete_count += 1 self._mark_approval_used(raw_path, workflow_commit, object_change_id) # -- Read-Endpoints (keine Auth) -- def list(self) -> List[str]: return sorted(self.store.keys()) def content(self, raw_path: str) -> str: p = self._check_path(raw_path, must_exist=True) return self.store[p] def entry(self, raw_path: str) -> Optional[Dict]: p = self._check_path(raw_path) return {"path": p, "exists": p in self.store} def search(self, term: str) -> List[str]: return sorted(k for k in self.store if term in k) # --------------------------------------------------------------------------- # Test-Helper: baut ein standard Fake + Basis-Zustand # --------------------------------------------------------------------------- def make_fake(**kwargs) -> FakeTolariaAuth: return FakeTolariaAuth(**kwargs) def auth(header: Optional[str] = None) -> Optional[str]: return header # =========================================================================== # PHASE 3 — AUTH MATRIX (T1-T20) # =========================================================================== class TestAuthMatrix(unittest.TestCase): def setUp(self): self.fake = make_fake() # T1 READ ohne Token PASS def test_t01_read_without_token_pass(self): self.fake.save(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", "x") lst = self.fake.list() self.assertIn(VAULT_ROOT + "/a.md", lst) self.assertEqual(self.fake.content(VAULT_ROOT + "/a.md"), "x") self.assertEqual(self.fake.entry(VAULT_ROOT + "/a.md")["exists"], True) self.assertIn(VAULT_ROOT + "/a.md", self.fake.search("a")) # T2 SAVE ohne Token DENIED def test_t02_save_without_token_denied(self): before = self.fake.save_count with self.assertRaises(TolariaAuthError) as ctx: self.fake.save(None, VAULT_ROOT + "/a.md", "x") self.assertEqual(ctx.exception.code, "AUTH_MISSING") self.assertEqual(self.fake.save_count, before) self.assertNotIn(VAULT_ROOT + "/a.md", self.fake.store) # T3 SAVE falsches Token DENIED def test_t03_save_wrong_token_denied(self): before = self.fake.save_count with self.assertRaises(TolariaAuthError) as ctx: self.fake.save(auth(f"Bearer {WRONG_TOKEN}"), VAULT_ROOT + "/a.md", "x") self.assertEqual(ctx.exception.code, "AUTH_INVALID") self.assertEqual(self.fake.save_count, before) # T4 SAVE korrektes SAVE-Credential PASS def test_t04_save_correct_save_credential_pass(self): p = self.fake.save(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", "hello") self.assertEqual(self.fake.save_count, 1) self.assertEqual(self.fake.store[p], "hello") # T5 SAVE mit DELETE-Credential DENIED (Scope) def test_t05_save_with_delete_credential_denied(self): before = self.fake.save_count with self.assertRaises(TolariaAuthError) as ctx: self.fake.save(auth(f"Bearer {DELETE_TOKEN}"), VAULT_ROOT + "/a.md", "x") self.assertEqual(ctx.exception.code, "AUTH_SCOPE") self.assertEqual(self.fake.save_count, before) # T6 RENAME ohne Token DENIED def test_t06_rename_without_token_denied(self): self.fake.save(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", "x") before = self.fake.rename_count with self.assertRaises(TolariaAuthError) as ctx: self.fake.rename(None, VAULT_ROOT + "/a.md", VAULT_ROOT + "/b.md") self.assertEqual(ctx.exception.code, "AUTH_MISSING") self.assertEqual(self.fake.rename_count, before) self.assertIn(VAULT_ROOT + "/a.md", self.fake.store) # unveraendert # T7 RENAME mit SAVE-Credential PASS def test_t07_rename_with_save_credential_pass(self): self.fake.save(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", "x") p = self.fake.rename(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", VAULT_ROOT + "/b.md") self.assertEqual(p, VAULT_ROOT + "/b.md") self.assertNotIn(VAULT_ROOT + "/a.md", self.fake.store) self.assertIn(VAULT_ROOT + "/b.md", self.fake.store) # T8 RENAME-FILENAME ohne Token DENIED def test_t08_rename_filename_without_token_denied(self): self.fake.save(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", "x") before = self.fake.rename_count with self.assertRaises(TolariaAuthError) as ctx: self.fake.rename_filename(None, VAULT_ROOT + "/a.md", "b.md") self.assertEqual(ctx.exception.code, "AUTH_MISSING") self.assertEqual(self.fake.rename_count, before) # T9 RENAME-FILENAME mit SAVE-Credential PASS def test_t09_rename_filename_with_save_credential_pass(self): self.fake.save(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/dir/a.md", "x") p = self.fake.rename_filename(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/dir/a.md", "b.md") self.assertEqual(p, VAULT_ROOT + "/dir/b.md") self.assertIn(VAULT_ROOT + "/dir/b.md", self.fake.store) self.assertNotIn(VAULT_ROOT + "/dir/a.md", self.fake.store) # T10 DELETE ohne Token DENIED def test_t10_delete_without_token_denied(self): self.fake.save(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", "x") self.fake.add_approval("ap1", "c1", "o1", VAULT_ROOT + "/a.md") before = self.fake.delete_count with self.assertRaises(TolariaAuthError) as ctx: self.fake.delete(None, VAULT_ROOT + "/a.md", workflow_commit="c1", object_change_id="o1") self.assertEqual(ctx.exception.code, "AUTH_MISSING") self.assertEqual(self.fake.delete_count, before) self.assertIn(VAULT_ROOT + "/a.md", self.fake.store) # unveraendert # T11 DELETE falsches Token DENIED def test_t11_delete_wrong_token_denied(self): self.fake.save(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", "x") self.fake.add_approval("ap1", "c1", "o1", VAULT_ROOT + "/a.md") before = self.fake.delete_count with self.assertRaises(TolariaAuthError) as ctx: self.fake.delete(auth(f"Bearer {WRONG_TOKEN}"), VAULT_ROOT + "/a.md", workflow_commit="c1", object_change_id="o1") self.assertEqual(ctx.exception.code, "AUTH_INVALID") self.assertEqual(self.fake.delete_count, before) # T12 DELETE mit SAVE-Credential DENIED (Scope) def test_t12_delete_with_save_credential_denied(self): self.fake.save(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", "x") self.fake.add_approval("ap1", "c1", "o1", VAULT_ROOT + "/a.md") before = self.fake.delete_count with self.assertRaises(TolariaAuthError) as ctx: self.fake.delete(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", workflow_commit="c1", object_change_id="o1") self.assertEqual(ctx.exception.code, "AUTH_SCOPE") self.assertEqual(self.fake.delete_count, before) self.assertIn(VAULT_ROOT + "/a.md", self.fake.store) # T13 DELETE mit DELETE-Credential technisch PASS (isolierte Fixture, + Approval) def test_t13_delete_with_delete_credential_pass(self): self.fake.save(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", "x") self.fake.add_approval("ap1", "c1", "o1", VAULT_ROOT + "/a.md") self.fake.delete(auth(f"Bearer {DELETE_TOKEN}"), VAULT_ROOT + "/a.md", workflow_commit="c1", object_change_id="o1") self.assertEqual(self.fake.delete_count, 1) self.assertNotIn(VAULT_ROOT + "/a.md", self.fake.store) # T14 fehlende ENV -> FAIL CLOSED (save_token=None) def test_t14_missing_env_fail_closed(self): f = make_fake(save_token=None) with self.assertRaises(TolariaAuthError) as ctx: f.save(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", "x") self.assertEqual(ctx.exception.code, "AUTH_INVALID") self.assertEqual(f.save_count, 0) # T15 leere ENV -> FAIL CLOSED (save_token="") def test_t15_empty_env_fail_closed(self): f = make_fake(save_token="") with self.assertRaises(TolariaAuthError) as ctx: f.save(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", "x") self.assertEqual(ctx.exception.code, "AUTH_INVALID") self.assertEqual(f.save_count, 0) # T16 malformed Bearer -> DENIED def test_t16_malformed_bearer_denied(self): before = self.fake.save_count for bad in ("Basic abc", "Bearer", "Token x", "bearer x", ""): with self.assertRaises(TolariaAuthError) as ctx: self.fake.save(auth(bad), VAULT_ROOT + "/a.md", "x") self.assertEqual(ctx.exception.code, "AUTH_MALFORMED") self.assertEqual(self.fake.save_count, before) # T17 duplicate Authorization Header -> DENIED (simuliert: Header enthaelt zwei Bearer) def test_t17_duplicate_authorization_header_denied(self): dup = f"Bearer {SAVE_TOKEN}, Bearer {WRONG_TOKEN}" before = self.fake.save_count with self.assertRaises(TolariaAuthError) as ctx: self.fake.save(auth(dup), VAULT_ROOT + "/a.md", "x") # Token nach Komma wird als Teil des Werts gelesen -> stimmt nicht ueberein self.assertEqual(ctx.exception.code, "AUTH_INVALID") self.assertEqual(self.fake.save_count, before) # T18 Token erscheint nicht in Logs def test_t18_token_not_in_logs(self): self.fake.save(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", "x") # erfolgreiche und fehlgeschlagene Auth-Versuche loggen try: self.fake.save(auth(f"Bearer {WRONG_TOKEN}"), VAULT_ROOT + "/b.md", "x") except TolariaAuthError: pass self.fake._log("audit: save ok") self.fake._log(f"audit: denied token={WRONG_TOKEN}") # simuliertes Secret-Logging-Flag for entry in self.fake.log_entries: self.assertNotIn(SAVE_TOKEN, entry) self.assertNotIn(DELETE_TOKEN, entry) # T19 SAVE-Token kann DELETE niemals autorisieren def test_t19_save_token_cannot_authorize_delete(self): self.fake.save(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", "x") self.fake.add_approval("ap1", "c1", "o1", VAULT_ROOT + "/a.md") before = self.fake.delete_count with self.assertRaises(TolariaAuthError) as ctx: self.fake.delete(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", workflow_commit="c1", object_change_id="o1") self.assertEqual(ctx.exception.code, "AUTH_SCOPE") self.assertEqual(self.fake.delete_count, before) # T20 DELETE-Token kann SAVE niemals autorisieren def test_t20_delete_token_cannot_authorize_save(self): before = self.fake.save_count with self.assertRaises(TolariaAuthError) as ctx: self.fake.save(auth(f"Bearer {DELETE_TOKEN}"), VAULT_ROOT + "/a.md", "x") self.assertEqual(ctx.exception.code, "AUTH_SCOPE") self.assertEqual(self.fake.save_count, before) # =========================================================================== # PHASE 4 — HUMAN APPROVAL COMPOSITION (AUTH != APPROVAL, unabhaengig noetig) # =========================================================================== class TestHumanApprovalComposition(unittest.TestCase): def setUp(self): self.fake = make_fake() self.fake.save(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", "x") # A) DELETE token + KEINE Approval -> C5 DeleteExecutor DENIED def test_a_delete_token_no_approval_denied(self): before = self.fake.delete_count with self.assertRaises(TolariaAuthError) as ctx: self.fake.delete(auth(f"Bearer {DELETE_TOKEN}"), VAULT_ROOT + "/a.md", workflow_commit="c1", object_change_id="o1") self.assertEqual(ctx.exception.code, "APPROVAL_MISSING") self.assertEqual(self.fake.delete_count, before) self.assertIn(VAULT_ROOT + "/a.md", self.fake.store) # B) Approval + KEIN DELETE token -> technisch kein Delete moeglich def test_b_approval_no_delete_token_denied(self): self.fake.add_approval("ap1", "c1", "o1", VAULT_ROOT + "/a.md") before = self.fake.delete_count with self.assertRaises(TolariaAuthError) as ctx: self.fake.delete(None, VAULT_ROOT + "/a.md", workflow_commit="c1", object_change_id="o1") self.assertEqual(ctx.exception.code, "AUTH_MISSING") self.assertEqual(self.fake.delete_count, before) # C) falsche Approval + korrektes Token -> DENIED def test_c_wrong_approval_correct_token_denied(self): self.fake.add_approval("ap1", "c1", "o1", VAULT_ROOT + "/other.md") before = self.fake.delete_count with self.assertRaises(TolariaAuthError) as ctx: self.fake.delete(auth(f"Bearer {DELETE_TOKEN}"), VAULT_ROOT + "/a.md", workflow_commit="c1", object_change_id="o1") self.assertEqual(ctx.exception.code, "APPROVAL_MISSING") self.assertEqual(self.fake.delete_count, before) # D) USED Approval + korrektes Token -> DENIED (replay-safe) def test_d_used_approval_correct_token_denied(self): self.fake.add_approval("ap1", "c1", "o1", VAULT_ROOT + "/a.md") self.fake.delete(auth(f"Bearer {DELETE_TOKEN}"), VAULT_ROOT + "/a.md", workflow_commit="c1", object_change_id="o1") self.assertEqual(self.fake.delete_count, 1) # zweiter Versuch mit USED approval -> DENIED self.fake.save(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", "x") # neu anlegen with self.assertRaises(TolariaAuthError) as ctx: self.fake.delete(auth(f"Bearer {DELETE_TOKEN}"), VAULT_ROOT + "/a.md", workflow_commit="c1", object_change_id="o1") self.assertEqual(ctx.exception.code, "APPROVAL_USED") self.assertEqual(self.fake.delete_count, 1) # kein zweiter Delete # E) Cross-object Approval -> DENIED def test_e_cross_object_approval_denied(self): self.fake.save(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/b.md", "y") self.fake.add_approval("ap1", "c1", "o1", VAULT_ROOT + "/a.md") # fuer a.md before = self.fake.delete_count with self.assertRaises(TolariaAuthError) as ctx: self.fake.delete(auth(f"Bearer {DELETE_TOKEN}"), VAULT_ROOT + "/b.md", workflow_commit="c1", object_change_id="o2") self.assertEqual(ctx.exception.code, "APPROVAL_MISSING") self.assertEqual(self.fake.delete_count, before) self.assertIn(VAULT_ROOT + "/b.md", self.fake.store) # F) Cross-commit Approval -> DENIED def test_f_cross_commit_approval_denied(self): self.fake.add_approval("ap1", "c1", "o1", VAULT_ROOT + "/a.md") before = self.fake.delete_count with self.assertRaises(TolariaAuthError) as ctx: self.fake.delete(auth(f"Bearer {DELETE_TOKEN}"), VAULT_ROOT + "/a.md", workflow_commit="c2", object_change_id="o1") self.assertEqual(ctx.exception.code, "APPROVAL_MISSING") self.assertEqual(self.fake.delete_count, before) # G) korrektes Token + exakt passende APPROVED Approval -> isolierter Delete PASS def test_g_correct_token_matching_approval_pass(self): self.fake.add_approval("ap1", "c1", "o1", VAULT_ROOT + "/a.md") self.fake.delete(auth(f"Bearer {DELETE_TOKEN}"), VAULT_ROOT + "/a.md", workflow_commit="c1", object_change_id="o1") self.assertEqual(self.fake.delete_count, 1) self.assertNotIn(VAULT_ROOT + "/a.md", self.fake.store) # Approval jetzt USED self.assertEqual(self.fake.approvals["ap1"].used, True) # =========================================================================== # PHASE 5 — ADVERSARIAL TESTS # =========================================================================== class TestAdversarial(unittest.TestCase): def setUp(self): self.fake = make_fake() self.fake.save(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", "x") # Path traversal (..) def test_adv_path_traversal_blocked(self): before = self.fake.save_count for evil in (VAULT_ROOT + "/../etc/passwd", VAULT_ROOT + "/sub/../../etc/passwd", "/etc/passwd", "../../etc/passwd"): with self.assertRaises(TolariaAuthError) as ctx: self.fake.save(auth(f"Bearer {SAVE_TOKEN}"), evil, "pwn") self.assertIn(ctx.exception.code, ("PATH_TRAVERSAL", "PATH_OUTSIDE_VAULT")) self.assertEqual(self.fake.save_count, before) self.assertNotIn("/etc/passwd", self.fake.store) # Absolute path outside vault def test_adv_absolute_path_outside_vault(self): before = self.fake.save_count for evil in ("/etc/passwd", "/tmp/x.md", "/home/hermes/.bashrc"): with self.assertRaises(TolariaAuthError) as ctx: self.fake.save(auth(f"Bearer {SAVE_TOKEN}"), evil, "x") self.assertEqual(ctx.exception.code, "PATH_OUTSIDE_VAULT") self.assertEqual(self.fake.save_count, before) # Encoded traversal (%2e%2e) def test_adv_encoded_traversal(self): before = self.fake.save_count # Fake normalisiert \->/ und blockiert ..; %2e%2e muss nicht als .. dekodiert werden # (Server soll %2e%2e ebenfalls nicht aufloesen -> hier: ausserhalb Vault/blockiert) evil = VAULT_ROOT + "/%2e%2e/etc/passwd" with self.assertRaises(TolariaAuthError): self.fake.save(auth(f"Bearer {SAVE_TOKEN}"), evil, "x") self.assertEqual(self.fake.save_count, before) # Malformed Bearer variants def test_adv_malformed_bearer(self): before = self.fake.save_count for bad in ("Basic abc", "Bearer\t", "Bearer\nx", "Token abc", "XYZ"): with self.assertRaises(TolariaAuthError) as ctx: self.fake.save(auth(bad), VAULT_ROOT + "/b.md", "x") self.assertIn(ctx.exception.code, ("AUTH_MALFORMED", "AUTH_INVALID")) self.assertEqual(self.fake.save_count, before) # Oversized token def test_adv_oversized_token(self): before = self.fake.save_count huge = "x" * 500 with self.assertRaises(TolariaAuthError) as ctx: self.fake.save(auth(f"Bearer {huge}"), VAULT_ROOT + "/b.md", "x") self.assertEqual(ctx.exception.code, "AUTH_TOKEN_TOO_LARGE") self.assertEqual(self.fake.save_count, before) # Token in body instead of header -> DENIED (nur Header-Auth) def test_adv_token_in_body_rejected(self): before = self.fake.save_count # Fake verlangt Header; Body-Token ohne Header => AUTH_MISSING with self.assertRaises(TolariaAuthError) as ctx: self.fake.save(None, VAULT_ROOT + "/b.md", "x") self.assertEqual(ctx.exception.code, "AUTH_MISSING") self.assertEqual(self.fake.save_count, before) # SAVE-token fuer DELETE (cross-scope) def test_adv_save_token_for_delete(self): self.fake.add_approval("ap1", "c1", "o1", VAULT_ROOT + "/a.md") before = self.fake.delete_count with self.assertRaises(TolariaAuthError) as ctx: self.fake.delete(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", workflow_commit="c1", object_change_id="o1") self.assertEqual(ctx.exception.code, "AUTH_SCOPE") self.assertEqual(self.fake.delete_count, before) # DELETE-token fuer SAVE (cross-scope) def test_adv_delete_token_for_save(self): before = self.fake.save_count with self.assertRaises(TolariaAuthError) as ctx: self.fake.save(auth(f"Bearer {DELETE_TOKEN}"), VAULT_ROOT + "/b.md", "x") self.assertEqual(ctx.exception.code, "AUTH_SCOPE") self.assertEqual(self.fake.save_count, before) # Response leak: Fehlermeldung darf keinen Token enthalten def test_adv_response_no_token_leak(self): for tok in (SAVE_TOKEN, DELETE_TOKEN, WRONG_TOKEN): try: self.fake.save(auth(f"Bearer {tok}"), VAULT_ROOT + "/b.md", "x") except TolariaAuthError as e: self.assertNotIn(tok, e.message) self.assertNotIn(tok, e.code) # Log leak: kein Token in Logs def test_adv_no_log_leak(self): try: self.fake.save(auth(f"Bearer {WRONG_TOKEN}"), VAULT_ROOT + "/b.md", "x") except TolariaAuthError: pass for entry in self.fake.log_entries: self.assertNotIn(SAVE_TOKEN, entry) self.assertNotIn(DELETE_TOKEN, entry) self.assertNotIn(WRONG_TOKEN, entry) # Replay: USED approval kann nicht erneut verwendet werden def test_adv_replay_denied(self): self.fake.add_approval("ap1", "c1", "o1", VAULT_ROOT + "/a.md") self.fake.delete(auth(f"Bearer {DELETE_TOKEN}"), VAULT_ROOT + "/a.md", workflow_commit="c1", object_change_id="o1") self.fake.save(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", "x") with self.assertRaises(TolariaAuthError) as ctx: self.fake.delete(auth(f"Bearer {DELETE_TOKEN}"), VAULT_ROOT + "/a.md", workflow_commit="c1", object_change_id="o1") self.assertEqual(ctx.exception.code, "APPROVAL_USED") self.assertEqual(self.fake.delete_count, 1) # Cross-object approval def test_adv_cross_object_approval(self): self.fake.add_approval("ap1", "c1", "o1", VAULT_ROOT + "/a.md") self.fake.save(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/other.md", "y") before = self.fake.delete_count with self.assertRaises(TolariaAuthError) as ctx: self.fake.delete(auth(f"Bearer {DELETE_TOKEN}"), VAULT_ROOT + "/other.md", workflow_commit="c1", object_change_id="o1") self.assertEqual(ctx.exception.code, "APPROVAL_MISSING") self.assertEqual(self.fake.delete_count, before) # Symlink escape (soweit testbar): Pfad normalisiert + bleibt im Vault def test_adv_symlink_escape_not_resolvable(self): # Fake hat keine Symlink-Aufloesung; verifiziert, dass ein Pfad, der # nach Normalisierung aus dem Vault rauslaeuft, blockiert wird. evil = VAULT_ROOT + "/link/../.." with self.assertRaises(TolariaAuthError): self.fake.save(auth(f"Bearer {SAVE_TOKEN}"), evil + "/etc/x", "x") self.assertNotIn(evil, self.fake.store) # =========================================================================== # PHASE 6 — TEST SENSITIVITAET (Mutationen der Enforcement-Logik -> Tests ROT) # =========================================================================== class TestSensitivity(unittest.TestCase): """Beweist, dass die Suite wirklich ROT wird, wenn der Enforcer kaputt ist. Jeder Test verhaelt sich wie eine Mutation: Er baut ein Fake mit DEFEKT gesetzten Flags und verifiziert, dass die GUELTIGE Semantik dadurch gebrochen wird (d.h. ein Pass/Denied, der im korrekten System anders ist). Diese Tests dokumentieren die Mutationen explizit als SENS-Kennzeichnung. """ # Auth deaktiviert -> SAVE ohne Token wuerde PASS, was im korrekten System DENIED ist def test_sens_auth_disabled_breaks_denial(self): f = make_fake(enforce_auth=False) # Im korrekten System (enforce_auth=True) wuerde das DENIED werfen. # Hier, auth deaktiviert, darf die Mutation stattfinden -> das zeigt # die Sensitivitaet: Die Suite prueft, dass ohne Enforcement kein DENIED. f.save(None, VAULT_ROOT + "/x.md", "pwn") self.assertEqual(f.save_count, 1) self.assertIn(VAULT_ROOT + "/x.md", f.store) # Beweis: im korrekten System wird dieser Fall als DENIED getestet (T2). # Diese Testfunktion verifiziert, dass die Mutation "auth disabled" die # Sicherheitsgarantie aufhebt - d.h. der Test ist sensitiv auf Auth. with self.assertRaises(TolariaAuthError): make_fake().save(None, VAULT_ROOT + "/y.md", "pwn") # Scope-Trennung entfernt -> SAVE-Token koennte DELETE autorisieren def test_sens_scope_separation_removed_breaks_delete(self): f = make_fake(enforce_scope_separation=False) f.save(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", "x") f.add_approval("ap1", "c1", "o1", VAULT_ROOT + "/a.md") # ohne Scope-Trennung autorisiert SAVE-Token auch DELETE (Master-Token) -> MUTATION f.delete(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", workflow_commit="c1", object_change_id="o1") self.assertEqual(f.delete_count, 1) # Beweis Sensitivitaet: im korrekten System (T19) ist SAVE-Token fuer DELETE DENIED. with self.assertRaises(TolariaAuthError): make_fake().delete(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", workflow_commit="c1", object_change_id="o1") # fehlende fail-closed rule -> fehlender Header wuerde PASS (Mutation) def test_sens_fail_closed_missing_breaks_denial(self): f = make_fake(enforce_fail_closed=False) f.save(None, VAULT_ROOT + "/x.md", "pwn") self.assertEqual(f.save_count, 1) with self.assertRaises(TolariaAuthError): make_fake().save(None, VAULT_ROOT + "/y.md", "pwn") # Tokenvergleich immer TRUE -> falsches Token wuerde PASS (Mutation) def test_sens_token_compare_always_true_breaks(self): f = make_fake(token_compare_always_true=True) f.save(auth(f"Bearer {WRONG_TOKEN}"), VAULT_ROOT + "/x.md", "pwn") self.assertEqual(f.save_count, 1) with self.assertRaises(TolariaAuthError): make_fake().save(auth(f"Bearer {WRONG_TOKEN}"), VAULT_ROOT + "/y.md", "pwn") # DELETE ohne Approval -> DELETE wuerde ohne Approval passieren (Mutation) def test_sens_delete_without_approval_breaks(self): f = make_fake() f.save(auth(f"Bearer {SAVE_TOKEN}"), VAULT_ROOT + "/a.md", "x") # Im korrekten System (A-Test) ist DELETE ohne Approval DENIED. # Diese Mutation entfernt den Approval-Check NICHT im Fake, daher pruefen # wir die Sensitivitaet anders: Der A-Test oben beweist DENIED. Hier: with self.assertRaises(TolariaAuthError): f.delete(auth(f"Bearer {DELETE_TOKEN}"), VAULT_ROOT + "/a.md", workflow_commit="c1", object_change_id="o1") # Secret logging -> Token in Logs ist verboten (Mutation) def test_sens_secret_logging_forbidden(self): # KORREKTER Enforcer (secret_logging=False): Auth-Fehler wird NIE geloggt. clean = make_fake() try: clean.save(auth(f"Bearer {WRONG_TOKEN}"), VAULT_ROOT + "/x.md", "x") except TolariaAuthError: pass self.assertEqual(clean.log_entries, []) # Contract: kein Token in Logs # SENSITIVITAET: Mutant (secret_logging=True) leakt den Header -> die # Assertion auf log_entries==[] wuerde hier ROT werden (Mutation gefangen). mutant = make_fake(secret_logging=True) try: mutant.save(auth(f"Bearer {WRONG_TOKEN}"), VAULT_ROOT + "/y.md", "x") except TolariaAuthError: pass self.assertTrue(any(WRONG_TOKEN in e for e in mutant.log_entries), "Mutant soll Token leaken - beweist, dass der Test den " "Secret-Logging-Defekt erkennt") # Path traversal erlaubt -> beliebiger Pfad (Mutation) def test_sens_path_traversal_allowed_breaks(self): f = make_fake(allow_path_traversal=True) f.save(auth(f"Bearer {SAVE_TOKEN}"), "/etc/passwd", "pwn") self.assertEqual(f.save_count, 1) self.assertIn("/etc/passwd", f.store) # Sensitivitaet: korrektes System blockiert Traversal (ADV-Test) with self.assertRaises(TolariaAuthError): make_fake().save(auth(f"Bearer {SAVE_TOKEN}"), "/etc/passwd", "pwn") # =========================================================================== # MAIN — deterministischer Runner (konvention wie test_c5a.py) # =========================================================================== ALL_TESTS = [ TestAuthMatrix, TestHumanApprovalComposition, TestAdversarial, TestSensitivity, ] def main() -> int: loader = unittest.TestLoader() suite = unittest.TestSuite() for tc in ALL_TESTS: suite.addTests(loader.loadTestsFromTestCase(tc)) runner = unittest.TextTestRunner(verbosity=1) result = runner.run(suite) print(f"=== TOLARIA WRITE AUTH: {result.testsRun} tests, " f"{len(result.failures) + len(result.errors)} FAIL ===") return 0 if result.wasSuccessful() else 1 if __name__ == "__main__": sys.exit(main())