nexo312/trading-system-docs
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
trading-system-docs/tolaria/tolaria-write-auth/test_tolaria_write_auth.py
Red Queen 40bbc40a49 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.
2026-08-27 05:57:22 +00:00

811 lines
39 KiB
Python
Raw Blame History

#!/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())
Reference in a new issue View git blame Copy permalink