From 40bbc40a49016dbb6f07ea6d82db1459d02b6db5 Mon Sep 17 00:00:00 2001 From: Red Queen Date: Thu, 27 Aug 2026 05:57:22 +0000 Subject: [PATCH] PRE_HERMES_SECURITY_GATE AUTH.1: Tolaria Write Auth Contract + isolierte Test-Suite - Contract (ADR): Auth-Modell (SAVE/DELETE-Scope getrennt), Credential-Modell (RQ/Hermes NO SAVE/DELETE; C5C Writer SAVE only; DeleteExecutor DELETE only), FAIL-CLOSED-Regeln, Human-DELETE-Gate (2 unabhaengige Ebenen AUTH+APPROVAL), Logging/Secret-Regeln, Rotation/Revocation, Path-Safety-Contract (Defense-in-Depth). - Isolierte Test-Suite (47 Tests): T1-T20 Auth-Matrix, Human-Approval-Composition A-G, Adversarial, Sensitivitaet (Mutationen -> ROT). KEIN Produktionscode geaendert. - KEINE produktive Mutation, KEINE echten Tokens, nur synthetische Fixture-Werte. --- .../PRE_HERMES_TOLARIA_WRITE_AUTH_CONTRACT.md | 194 +++++ .../test_tolaria_write_auth.py | 811 ++++++++++++++++++ 2 files changed, 1005 insertions(+) create mode 100644 tolaria/tolaria-write-auth/PRE_HERMES_TOLARIA_WRITE_AUTH_CONTRACT.md create mode 100644 tolaria/tolaria-write-auth/test_tolaria_write_auth.py diff --git a/tolaria/tolaria-write-auth/PRE_HERMES_TOLARIA_WRITE_AUTH_CONTRACT.md b/tolaria/tolaria-write-auth/PRE_HERMES_TOLARIA_WRITE_AUTH_CONTRACT.md new file mode 100644 index 0000000..c6d1fcc --- /dev/null +++ b/tolaria/tolaria-write-auth/PRE_HERMES_TOLARIA_WRITE_AUTH_CONTRACT.md @@ -0,0 +1,194 @@ +# PRE_HERMES_SECURITY_GATE — Tolaria Write Authentication Contract (AUTH) + +> **Status:** VERBINDLICHER SECURITY-CONTRACT (Entwurf, AUTH.1) +> **Klassifikation:** PRE_HERMES_SECURITY_GATE §5/§31 — CRITICAL-Gap-Closure +> **Mission:** PRE_HERMES_SECURITY_GATE AUTH.1 — Contract + isolierte Tests +> **Sprache:** Deutsch (verbindlich) · Code-/API-/Schema-Identifier englisch +> **Verwandt:** `C5_SYNC_ARCHITECTURE_DESIGN.md` (§5, §31), `C5_DELETE_EXECUTION_ARCHITECTURE_DECISION.md`, `ROOT_SSH_GATE.md`, `SAFETY_CONTRACT.md` + +--- + +## 1. Zweck & Geltungsbereich + +Dieser Contract definiert die **authentifizierte Write-Zugriffskontrolle** für das Tolaria +Vault-API. Er adressiert die verifizierte CRITICAL Gap aus PRE_HERMES_SECURITY_GATE Discovery: +`POST /api/vault/save` und `/api/vault/delete` (sowie `rename`/`rename-filename`) sind +aktuell **unauthentifiziert** über das interne Netz erreichbar. + +**Dieses Dokument ist ein DESIGN-Contract. Es implementiert nichts.** AUTH.1 liefert nur +Contract + isolierte Tests (Test-Harness). Die tatsächliche Server-Enforcement ist AUTH.2+ +und wird separat freigegeben. + +### Grundsatz +- **READ** bleibt intern **ohne** Write-Credential erreichbar (harmlos, öffentlich im internen Netz). +- **WRITE** (save/rename/rename-filename/delete) erfordert **authentifizierte Credentials**. +- **SAVE-Scope und DELETE-Scope sind getrennte Credentials.** Es gibt **KEIN** gemeinsames Master-Write-Token. +- **Hermes/RQ erhalten KEIN direktes Tolaria-Write-Credential.** Writes laufen ausschließlich über den isolierten C5-Executor. +- **FAIL CLOSED:** Fehlende/leere/falsche Auth-Konfiguration → mutierende Endpoints **DENIED** (keinerlei Mutation). + +--- + +## 2. AUTHENTICATION MODEL (Scope-Definition) + +| Scope | Endpoints | Auth | Hinweis | +|---|---|---|---| +| **READ** | `POST /api/vault/list` · `POST /api/vault/content` · `POST /api/vault/entry` · `POST /api/vault/search` | **KEINE** (öffentlich intern) | Lesen ist sicher; keine unnötige Auth | +| **SAVE** | `POST /api/vault/save` · `POST /api/vault/rename` · `POST /api/vault/rename-filename` | **SAVE-Credential** erforderlich | Reguläre Writes/Kanonalisierung | +| **DELETE** | `POST /api/vault/delete` | **DELETE-Credential** erforderlich (getrennt) | Destruktiv, höchste Schutzstufe | + +### Scope-Regeln +- Ein **SAVE-Credential autorisiert NIE** DELETE (und NIE umgekehrt). +- Ein **DELETE-Credential autorisiert NIE** SAVE. +- Ein Request mit **falschem Scope-Credential** → `403 DENIED`, keine Mutation. + +--- + +## 3. CREDENTIAL MODEL + +| Rolle / Caller | SAVE-Credential | DELETE-Credential | Begründung | +|---|---|---|---| +| **Red Queen (Hermes)** | **NO** | **NO** | Orchestrator schreibt nicht direkt; kein direkter Tolaria-Grund | +| **Hermes (generisch)** | **NO** | **NO** | Kein Tolaria-Write-Zweck | +| **Rain (OpenClaw)** | **NO** | **NO** | kein direktes Tolaria-Write | +| **C5C Writer (Executor)** | **YES** (nur SAVE) | **NO** | isolierter Propagator, schreibt Vault-Inhalt | +| **C5 DeleteExecutor (privilegiert)** | **NO** | **YES** (nur DELETE) | einziger DELETE-Ausführer, + Human Approval | +| **Tauri-UI (lokal, gleiche Origin)** | YES | YES | lokale Admin-Oberfläche (gleiche Origin-Vertrauensgrenze) | +| **C5 CLI / Migrationen / Admin** | YES | YES (nur via separatem Pfad) | kontrollierte manuelle Pfade | + +### Least-Privilege +- Jeder Caller erhält **nur die Credentials, die er für seine dokumentierte Aufgabe benötigt**. +- Kein Caller erhält beide Credentials, außer die lokale UI / manuelle Admin-Pfade (gleiche Origin-Vertrauensgrenze). +- **C5C Writer und C5 DeleteExecutor sind strikt getrennte Prozesse mit getrennten Credentials.** + +--- + +## 4. FAIL-CLOSED RULES + +Folgende Bedingungen führen zwingend zu **DENIED** (HTTP 401/403), **nie** zu einer Mutation: + +| Bedingung | Ergebnis | +|---|---| +| Fehlende Token-Konfiguration (ENV nicht gesetzt) | **DENIED** (401) | +| Leeres Token (ENV = `""`) | **DENIED** (401) | +| Falsches Token (Wert ≠ erwartet) | **DENIED** (401/403) | +| Falscher Scope (SAVE-Token auf DELETE-Endpoint o. umgekehrt) | **DENIED** (403) | +| Malformed `Authorization`-Header (kein gültiges `Bearer `) | **DENIED** (400/401) | +| Duplicate `Authorization`-Header | **DENIED** (400) | +| Token im Request-Body statt im Header | **DENIED** (kein Body-Token akzeptiert) | +| Überlanges Token / Oversized-Header | **DENIED** (400) | + +**Regel:** Ein Auth-Fehler darf **niemals** zu einer teilweisen/partiellen Mutation führen. +Auth-Check erfolgt **vor** jeder Payload-Verarbeitung und **vor** jeder Schreiboperation. + +**Server-Start-Fail-Closed:** Wenn die für einen mutierenden Endpoint erforderliche +Auth-Konfiguration beim Start fehlt, startet der Endpoint **nicht** unauthentifiziert. +Er verweigert alle Write-Requests bis eine gültige Konfiguration vorhanden ist. + +--- + +## 5. HUMAN DELETE GATE (zwei unabhängige Ebenen) + +**DELETE benötigt zwei unabhängige Bedingungen, die sich NICHT ersetzen:** + +``` +EBENE A (AUTHORIZATION/AUTHENTICATION): Der Aufrufer besitzt ein gültiges DELETE-Credential. + → technische Fähigkeit, POST /api/vault/delete auszuführen. + +UND (logisches UND) + +EBENE B (HUMAN APPROVAL): Für genau diese konkrete DELETE-Aktion existiert eine gültige, + persistierte, commit- und objekt-spezifische Human-Approval (C5 Invarianten A–N, + ST_DELETE_APPROVED). + → menschliche Freigabe durch Christian. +``` + +- **Keine Ebene ersetzt die andere.** DELETE-Credential ohne Human Approval → C5-DeleteExecutor verweigert. Human Approval ohne DELETE-Credential → technisch kein Delete möglich. +- Hermes/RQ besitzt **nie** die Kombination aus beidem; insbesondere **nie** ein DELETE-Credential. + +--- + +## 6. LOGGING / SECRET RULES + +- Der **`Authorization`-Header wird niemals geloggt** (weder Wert noch Präfix). +- Der **Token-Wert erscheint niemals** in: + - Error-Responses + - Stacktraces + - Audit-Reports + - Telegram-Meldungen + - Repository (committed Code/Docs/Fixtures) + - Test-Fixtures mit echtem Wert (nur synthetische Testwerte) +- Fehlermeldungen bei Auth-Failure enthalten **keine** Token-Teile oder Header-Rohdaten. +- Secret-Scanner (C5B `detect_secret`-Muster: `sk-`, `Bearer `, `-----BEGIN ... PRIVATE KEY-----` etc.) laufen vor jedem Commit/Report. + +--- + +## 7. ROTATION / REVOCATION + +- **SAVE- und DELETE-Credentials sind getrennt rotierbar** — Rotation des einen berührt den anderen nicht. +- **DELETE-Credential ist sofort widerrufbar:** Widerruf → sofort keine Deletes mehr möglich (unabhängig von Approval). +- Rotations- und Revocation-Verfahren sind getrennt dokumentiert; kein gemeinsamer Rotations-Zyklus. +- Token werden über Secret-Manager/Container-ENV injiziert, **nie** im Repo, in Logs oder Reports. + +--- + +## 8. PATH SAFETY CONTRACT (Defense-in-Depth) + +Der Discovery-Incident zeigte: `DELETE` akzeptiert offenbar frei übergebenen Pfad. Diese Regeln +sind **Security-Anforderung für spätere AUTH-Phasen** (AUTH.2+). AUTH.1 implementiert sie noch nicht, +dokumentiert sie aber als verbindliche Anforderung: + +| Regel | Beschreibung | +|---|---| +| **Vault-Root-Beschränkung** | Ausschließlich Pfade **innerhalb** des Vault-Roots (`/app/vault/`). | +| **Kein absoluter Fremdpfad** | Pfade außerhalb des Vault-Roots → **DENIED**. | +| **Kein `..`-Traversal** | `..`-Segmente werden normalisiert und **blockiert** (FAIL CLOSED). | +| **Kein Symlink-Escape** | Auflösung darf den Vault-Root nicht verlassen. | +| **Normalisierter Vault-Path** | `\`→`/` normalisieren, doppelte Slashes, `.`-Segmente, dann Präfix-Prüfung. | +| **Delete nur auf existierendem Object** | DELETE gegen nicht-existierenden Pfad → **DENIED** (keine stille No-Op-Mutation, keine Fremddatei). | +| **Rename Source+Destination im Vault** | Beide Pfade müssen im Vault-Root liegen. | +| **Kein Überschreiben außerhalb erlaubter Semantik** | Kein Überschreiben von Pfaden, die nicht Teil des Vault-Roots sind. | + +> **Hinweis:** Diese Regeln sind **Defense-in-Depth** zusätzlich zur Auth. Sie werden in AUTH.2+ +> als Server-seitige Validierung implementiert. Auth allein ersetzt sie nicht; die Kombination +> aus Auth + Path-Safety ist die vollständige Write-Sicherheit. + +--- + +## 9. AUTH-API-CONTRACT (angestrebtes Verhalten, AUTH.2+) + +``` +READ POST /api/vault/{list,content,entry,search} → unverändert (öffentlich intern) +SAVE POST /api/vault/save ohne/falsch/Scope-falsch Credential → 401/403, KEINE Mutation +SAVE POST /api/vault/save mit gültigem SAVE-Credential → bestehende Semantik +RENAME POST /api/vault/rename mit gültigem SAVE-Credential → erlaubt +RENAME-FILENAME POST /api/vault/rename-filename mit SAVE-Credential → erlaubt +DELETE POST /api/vault/delete ohne/falsch/SAVE-Credential → 401/403, KEINE Mutation +DELETE POST /api/vault/delete mit gültigem DELETE-Credential → technisch erlaubt (Human Gate bleibt vorgelagert) +``` + +- **Header:** `Authorization: Bearer ` +- **Token-Vergleich:** constant-time (`secrets.compare_digest` / gleichwertig), kein Timing-Leak. +- **Nur Header-Auth:** Token im Body/Query wird nicht akzeptiert. +- **Duplicate/Malformed Header:** → DENIED (400). + +--- + +## 10. OUT-OF-SCOPE (AUTH.1) + +AUTH.1 implementiert **nichts** davon produktiv: +- ✅ nur: dieses Contract-Dokument + isolierte Test-Suite (Test-Harness, synthetische Werte) +- ❌ keine Server-Enforcement (AUTH.2) +- ❌ kein Caller-Integration (AUTH.3/AUTH.4) +- ❌ keine Tokens erzeugen/lesen/ausgeben/kopieren +- ❌ keine ENV-/Deployment-/Netzwerk-/Container-Änderung +- ❌ keine produktiven Endpoint-Probes / Mutationen + +--- + +## 11. Quellen & Verifikationsbasis + +- `C5_SYNC_ARCHITECTURE_DESIGN.md` §5 (Z.81-91): "`POST /api/vault/save` ist unauthentifiziert" — SECURITY DEBT. +- §31 (Z.380-384): PRE_HERMES_SECURITY_GATE — Entscheidung A/B/C nötig vor Hermes-Autonomie. +- `tolaria-c2a-canonicalization-canary.md`: Bridge-Quelle `/src/mock-tauri/vault-api.ts` — Endpoint-Surface: `save`→`/save`, `rename`→`/rename`, `rename-filename`→`/rename-filename`, `delete`→`/delete`; `list/content/entry/all-content/search` = Reads. +- `tolaria-verified-endpoints.md`: verifizierte Endpoints (list/content/save; `/app/vault/`-Präfix Pflicht). +- Discovery-Incident (2026-08-27): unauthentifizierter `save` (0-Byte-Datei) + unauthentifizierter `delete` (beliebiger Pfad) — **bestätigt CRITICAL**. diff --git a/tolaria/tolaria-write-auth/test_tolaria_write_auth.py b/tolaria/tolaria-write-auth/test_tolaria_write_auth.py new file mode 100644 index 0000000..f9677f1 --- /dev/null +++ b/tolaria/tolaria-write-auth/test_tolaria_write_auth.py @@ -0,0 +1,811 @@ +#!/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())