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.
This commit is contained in:
parent
9402a92cc3
commit
40bbc40a49
2 changed files with 1005 additions and 0 deletions
|
|
@ -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 <token>`) | **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 <token>`, `-----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>`
|
||||
- **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**.
|
||||
811
tolaria/tolaria-write-auth/test_tolaria_write_auth.py
Normal file
811
tolaria/tolaria-write-auth/test_tolaria_write_auth.py
Normal file
|
|
@ -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())
|
||||
Loading…
Reference in a new issue