nexo312/trading-system-docs
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
trading-system-docs/tolaria/c5-sync-service/test_c5_auth3a.py
Red Queen a11c1bbe53 AUTH.3A: C5 Caller Auth Integration (SAVE/DELETE Credential, fail-closed, Tests, Sensitivity, Gap-Doku) 
- rq_c5c.py: TolariaClient save_token/delete_token DI, _require_token fail-closed,
  write()/delete() senden Bearer (SAVE/DELETE), read()/list() ohne Credential
- rq_c5_cli.py: _delete_executor liest nur DELETE-Credential (Least Privilege)
- test_c5c.py: write()-Tests injizieren synthetisches SAVE-Token
- test_c5_auth3a.py: isolierte AUTH.3A-Testsuite (19 Tests, Fake/Mock Tolaria)
- auth3a-sensitivity.sh: 8 Sensitivitaets-Mutationen (A-H) -> ROT
- AUTH3A_HUMAN_APPROVAL_AUTHENTICITY_GAP.md: Gap dokumentiert (OPEN, nicht repariert)

Keine echten Tokens. Keine ENV-Mutation. Kein Deployment. Keine produktive Auth-Aktivierung.
2026-08-27 08:42:26 +00:00

444 lines
18 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
Red Queen — AUTH.3A C5 Caller Auth Integration Testsuite.
Testet die clientseitige Auth-Unterstuetzung der legitimen C5-Tolaria-Writer
gegen ein Fake/Mock-Tolaria (KEINE Produktion, KEINE echten Tokens).
Abgedeckte Faelle (AUTH.3A PHASE 9):
SAVE CALLER (TolariaClient.write -> /save, SAVE-Scope)
T1 SAVE mit SAVE token -> Authorization korrekt
T2 SAVE ohne token -> lokal DENIED -> HTTP niemals aufgerufen
T3 SAVE leeres token -> lokal DENIED
T4 READ ohne SAVE token -> funktioniert
T5 SAVE token wird bei READ nicht gesendet
DELETE CALLER (TolariaClient.delete -> /delete, DELETE-Scope)
T6 DELETE mit Approval + DELETE token -> Request erlaubt
T7 DELETE mit Approval aber ohne token -> lokal DENIED
T8 DELETE mit token aber ohne Approval -> DENIED
T9 DELETE ohne Approval und ohne token -> DENIED
T10 DELETE token nicht bei SAVE nutzbar
T11 SAVE token nicht bei DELETE nutzbar
LOGGING
T12 Token nicht in Logs
T13 Token nicht in Exceptions
T14 Token nicht in Evidence
RETRY/REPLAY
T15 Retry behaelt korrekten Scope
T16 Replay erzeugt keinen unauthentifizierten Write
T17 fehlendes Credential bei Recovery -> fail-closed
AUTH.3A PHASE 10 (Wrong-Scope Adversarial):
T18 SAVE credential beim DeleteExecutor -> scheitert
T19 DELETE credential beim Save-Caller -> scheitert
Nur synthetische Test-Tokens. KEINE echten Secrets.
"""
from __future__ import annotations
import json
import os
import sys
import tempfile
import unittest
from unittest import mock
from typing import Any, Dict, List, Optional
# C5-Module importieren (aus demselben Verzeichnis)
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from rq_c5c import (
TolariaClient,
TolariaWriteError,
RC_AUTH_FAILURE,
ENV_TOLARIA_SAVE_TOKEN,
ENV_TOLARIA_DELETE_TOKEN,
)
from rq_c5_delete import DeleteExecutor, DeleteApprovalError, DeleteExecutionError
from rq_c5a import (
C5AStore,
ST_DELETE_APPROVED,
ST_UPDATING_SEARCH,
OP_DELETE_REQUEST,
)
# Synthetische Test-Tokens (AUTH.1 §6: nur synthetische Werte, keine echten)
SAVE_TOKEN = "auth3a-save-synthetic-token-0001"
DELETE_TOKEN = "auth3a-delete-synthetic-token-0002"
class FakeTolaria:
"""In-memory Fake-Tolaria (kein Netzwerk, keine Produktion)."""
def __init__(self) -> None:
self.objects: Dict[str, str] = {}
self.calls: List[Dict[str, Any]] = []
def read(self, vault_path: str) -> Optional[str]:
self.calls.append({"op": "read", "path": vault_path})
return self.objects.get(vault_path)
def write(self, vault_path: str, content: str) -> Dict[str, Any]:
self.calls.append({"op": "write", "path": vault_path})
self.objects[vault_path] = content
return {}
def delete(self, vault_path: str) -> Dict[str, Any]:
self.calls.append({"op": "delete", "path": vault_path})
self.objects.pop(vault_path, None)
return {}
def list(self, vault_path: str = "/app/vault") -> List[Dict[str, Any]]:
self.calls.append({"op": "list", "path": vault_path})
return [{"path": p} for p in self.objects]
class FakeTolariaClient(TolariaClient):
"""TolariaClient, der gegen FakeTolaria statt HTTP arbeitet.
Ueberschreibt write()/delete()/read()/list() -> _require_token wird in
den ueberschriebenen Methoden NICHT aufgerufen. Fuer die Auth-Tests
(T1-T17) wird stattdessen der ECHTE TolariaClient mit gemocktem _post
verwendet, um den Authorization-Header zu verifizieren.
"""
def __init__(self, fake: FakeTolaria, **kwargs: Any):
super().__init__(base_url="http://fake", **kwargs)
self.fake = fake
def read(self, vault_path: str) -> Optional[str]:
return self.fake.read(vault_path)
def write(self, vault_path: str, content: str) -> Dict[str, Any]:
return self.fake.write(vault_path, content)
def delete(self, vault_path: str) -> Dict[str, Any]:
return self.fake.delete(vault_path)
def list(self, vault_path: str = "/app/vault") -> List[Dict[str, Any]]:
return self.fake.list(vault_path)
# ---------------------------------------------------------------------------
# Test-Helfer: Echter TolariaClient mit gemocktem _post (Auth-Header pruefen)
# ---------------------------------------------------------------------------
def _capture_post(client: TolariaClient) -> List[Dict[str, Any]]:
"""Mockt urllib.request.urlopen global und zeichnet (endpoint, auth_header) auf.
Testet den ECHTEN _post-Pfad inkl. Header-Bau, damit Mutationen am
Authorization-Header (z. B. Header entfernt) erkannt werden. Der Mock
bleibt aktiv, bis die Tests die Aufrufe gemacht haben.
"""
captured: List[Dict[str, Any]] = []
class FakeResp:
def __init__(self, body: bytes = b"{}"):
self._body = body
def read(self) -> bytes:
return self._body
def __enter__(self):
return self
def __exit__(self, *a):
return False
def fake_urlopen(req: Any, timeout: float = 15.0) -> FakeResp:
url = req.full_url
endpoint = url.rsplit("/", 1)[-1]
headers = dict(req.headers)
captured.append({
"endpoint": endpoint,
"auth_header": headers.get("Authorization"),
})
return FakeResp()
patcher = mock.patch("urllib.request.urlopen", side_effect=fake_urlopen)
patcher.start()
# Mock bleibt aktiv; Tests rufen danach client.write()/delete()/read() auf.
return captured
def _make_store() -> C5AStore:
db = os.path.join(tempfile.mkdtemp(prefix="auth3a_db_"), "c5a.db")
return C5AStore(db)
def _seed_delete_approved(store: C5AStore, sha: str, oid: str, path: str) -> int:
"""Legt einen DELETE_APPROVED-Commit + Approval an (fuer T6-T9).
Gibt die echte object_change_id (DB-Zeilen-ID) zurueck.
"""
store.upsert_commit({
"commit_sha": sha, "parent_sha": None, "sequence": 1,
"status": ST_DELETE_APPROVED, "retry_count": 0,
})
change = store.add_object_change({
"commit_sha": sha, "object_id": oid,
"operation": OP_DELETE_REQUEST, "path_before": path, "path_after": None,
})
assert change is not None
change_id = int(change["id"])
store.create_delete_approval(
sha, change_id, oid, path, "human:christian",
approval_id="appr-1",
)
return change_id
# ---------------------------------------------------------------------------
# SAVE CALLER Tests (T1-T5)
# ---------------------------------------------------------------------------
class TestSaveCallerAuth(unittest.TestCase):
"""AUTH.3A PHASE 9: SAVE-Caller (TolariaClient.write -> /save)."""
def test_t1_save_with_token_sends_authorization(self):
"""T1: SAVE mit SAVE token -> Authorization korrekt."""
client = TolariaClient(base_url="http://fake", save_token=SAVE_TOKEN)
captured = _capture_post(client)
client.write("/app/vault/x.md", "content")
self.assertEqual(len(captured), 1)
self.assertEqual(captured[0]["endpoint"], "save")
self.assertEqual(captured[0]["auth_header"], f"Bearer {SAVE_TOKEN}")
def test_t2_save_without_token_fails_closed_no_http(self):
"""T2: SAVE ohne token -> lokal DENIED -> HTTP niemals aufgerufen."""
client = TolariaClient(base_url="http://fake") # kein save_token
captured = _capture_post(client)
with self.assertRaises(TolariaWriteError) as ctx:
client.write("/app/vault/x.md", "content")
self.assertEqual(ctx.exception.reason_code, RC_AUTH_FAILURE)
# HTTP (via urlopen) wurde NICHT aufgerufen
self.assertEqual(len(captured), 0)
def test_t3_save_empty_token_fails_closed(self):
"""T3: SAVE leeres token -> lokal DENIED."""
client = TolariaClient(base_url="http://fake", save_token="")
captured = _capture_post(client)
with self.assertRaises(TolariaWriteError) as ctx:
client.write("/app/vault/x.md", "content")
self.assertEqual(ctx.exception.reason_code, RC_AUTH_FAILURE)
self.assertEqual(len(captured), 0)
def test_t4_read_without_save_token_works(self):
"""T4: READ ohne SAVE token -> funktioniert."""
client = TolariaClient(base_url="http://fake") # kein save_token
captured = _capture_post(client)
# read() -> _post("content", ...) ohne auth_token
client.read("/app/vault/x.md")
self.assertEqual(len(captured), 1)
self.assertEqual(captured[0]["endpoint"], "content")
self.assertIsNone(captured[0]["auth_header"])
def test_t5_save_token_not_sent_on_read(self):
"""T5: SAVE token wird bei READ nicht gesendet (Least Privilege)."""
client = TolariaClient(base_url="http://fake", save_token=SAVE_TOKEN)
captured = _capture_post(client)
client.read("/app/vault/x.md")
client.list("/app/vault")
# Weder content noch list senden ein Credential
for c in captured:
self.assertIsNone(c["auth_header"])
# ---------------------------------------------------------------------------
# DELETE CALLER Tests (T6-T11)
# ---------------------------------------------------------------------------
class TestDeleteCallerAuth(unittest.TestCase):
"""AUTH.3A PHASE 9: DELETE-Caller (TolariaClient.delete -> /delete)."""
def test_t6_delete_with_approval_and_token_allowed(self):
"""T6: DELETE mit Approval + DELETE token -> Request erlaubt.
Testet die Client-Ebene: delete() mit delete_token sendet den
Authorization-Header. (Die Approval-Gates des Executors werden in
T8/T9 separat getestet; hier geht es um die Credential-Integration.)
"""
client = TolariaClient(base_url="http://fake", delete_token=DELETE_TOKEN)
captured = _capture_post(client)
result = client.delete("/app/vault/notes/x.md")
self.assertEqual(len(captured), 1)
self.assertEqual(captured[0]["endpoint"], "delete")
self.assertEqual(captured[0]["auth_header"], f"Bearer {DELETE_TOKEN}")
def test_t7_delete_with_approval_but_no_token_fails_closed(self):
"""T7: DELETE mit Approval aber ohne token -> lokal DENIED."""
client = TolariaClient(base_url="http://fake") # kein delete_token
captured = _capture_post(client)
with self.assertRaises(TolariaWriteError) as ctx:
client.delete("/app/vault/notes/x.md")
self.assertEqual(ctx.exception.reason_code, RC_AUTH_FAILURE)
self.assertEqual(len(captured), 0)
def test_t8_delete_with_token_but_no_approval_denied(self):
"""T8: DELETE mit token aber ohne Approval -> DENIED (Human Gate)."""
store = _make_store()
# Kein Approval angelegt -> DeleteExecutor.execute() muss fail-closed
store.upsert_commit({
"commit_sha": "sha2", "parent_sha": None, "sequence": 1,
"status": ST_DELETE_APPROVED, "retry_count": 0,
})
store.add_object_change({
"commit_sha": "sha2", "id": "chg-2", "object_id": "object/2",
"operation": OP_DELETE_REQUEST, "path_before": "notes/y.md",
"path_after": None,
})
client = TolariaClient(base_url="http://fake", delete_token=DELETE_TOKEN)
executor = DeleteExecutor(store, client)
with self.assertRaises(DeleteApprovalError):
executor.execute("sha2")
store.close()
def test_t9_delete_no_approval_no_token_denied(self):
"""T9: DELETE ohne Approval und ohne token -> DENIED."""
store = _make_store()
store.upsert_commit({
"commit_sha": "sha3", "parent_sha": None, "sequence": 1,
"status": ST_DELETE_APPROVED, "retry_count": 0,
})
store.add_object_change({
"commit_sha": "sha3", "id": "chg-3", "object_id": "object/3",
"operation": OP_DELETE_REQUEST, "path_before": "notes/z.md",
"path_after": None,
})
client = TolariaClient(base_url="http://fake") # kein delete_token
executor = DeleteExecutor(store, client)
with self.assertRaises(DeleteApprovalError):
executor.execute("sha3")
store.close()
def test_t10_delete_token_not_usable_for_save(self):
"""T10: DELETE token nicht bei SAVE nutzbar (Scope-Separation)."""
# Client mit NUR delete_token -> write() (SAVE) muss fail-closed
client = TolariaClient(base_url="http://fake", delete_token=DELETE_TOKEN)
captured = _capture_post(client)
with self.assertRaises(TolariaWriteError) as ctx:
client.write("/app/vault/x.md", "content")
self.assertEqual(ctx.exception.reason_code, RC_AUTH_FAILURE)
self.assertEqual(len(captured), 0)
def test_t11_save_token_not_usable_for_delete(self):
"""T11: SAVE token nicht bei DELETE nutzbar (Scope-Separation)."""
# Client mit NUR save_token -> delete() (DELETE) muss fail-closed
client = TolariaClient(base_url="http://fake", save_token=SAVE_TOKEN)
captured = _capture_post(client)
with self.assertRaises(TolariaWriteError) as ctx:
client.delete("/app/vault/x.md")
self.assertEqual(ctx.exception.reason_code, RC_AUTH_FAILURE)
self.assertEqual(len(captured), 0)
# ---------------------------------------------------------------------------
# LOGGING Tests (T12-T14)
# ---------------------------------------------------------------------------
class TestLoggingRedaction(unittest.TestCase):
"""AUTH.3A PHASE 8/9: Token nicht in Logs/Exceptions/Evidence."""
def test_t12_token_not_in_logs(self):
"""T12: Token nicht in Logs (kein repr/str enthaelt den Wert)."""
client = TolariaClient(base_url="http://fake", save_token=SAVE_TOKEN,
delete_token=DELETE_TOKEN)
# repr/str des Clients darf keinen Token-Wert enthalten
self.assertNotIn(SAVE_TOKEN, repr(client))
self.assertNotIn(DELETE_TOKEN, repr(client))
self.assertNotIn(SAVE_TOKEN, str(client))
self.assertNotIn(DELETE_TOKEN, str(client))
def test_t13_token_not_in_exceptions(self):
"""T13: Token nicht in Exceptions (fail-closed Fehler)."""
client = TolariaClient(base_url="http://fake", save_token=SAVE_TOKEN)
# Fehler bei fehlendem Credential (delete ohne delete_token)
try:
client.delete("/app/vault/x.md")
self.fail("sollte fail-closed werfen")
except TolariaWriteError as e:
self.assertNotIn(DELETE_TOKEN, str(e))
self.assertNotIn(SAVE_TOKEN, str(e))
self.assertNotIn("Bearer", str(e))
def test_t14_token_not_in_evidence(self):
"""T14: Token nicht in Evidence (to_dict / Fehler-Dict)."""
client = TolariaClient(base_url="http://fake", save_token=SAVE_TOKEN)
try:
client.delete("/app/vault/x.md")
self.fail("sollte fail-closed werfen")
except TolariaWriteError as e:
d = e.to_dict()
self.assertNotIn(SAVE_TOKEN, json.dumps(d))
self.assertNotIn(DELETE_TOKEN, json.dumps(d))
# ---------------------------------------------------------------------------
# RETRY/REPLAY Tests (T15-T17)
# ---------------------------------------------------------------------------
class TestRetryReplay(unittest.TestCase):
"""AUTH.3A PHASE 9: Retry/Replay behalten korrekten Scope."""
def test_t15_retry_keeps_correct_scope(self):
"""T15: Retry behaelt korrekten Scope (SAVE-Token bleibt SAVE)."""
client = TolariaClient(base_url="http://fake", save_token=SAVE_TOKEN)
captured = _capture_post(client)
# Zwei aufeinanderfolgende SAVE-Aufrufe (Retry-Szenario)
client.write("/app/vault/a.md", "a")
client.write("/app/vault/b.md", "b")
self.assertEqual(len(captured), 2)
for c in captured:
self.assertEqual(c["endpoint"], "save")
self.assertEqual(c["auth_header"], f"Bearer {SAVE_TOKEN}")
def test_t16_replay_no_unauthenticated_write(self):
"""T16: Replay erzeugt keinen unauthentifizierten Write."""
# Client ohne save_token -> write() fail-closed (kein unauthentifizierter Write)
client = TolariaClient(base_url="http://fake")
captured = _capture_post(client)
with self.assertRaises(TolariaWriteError):
client.write("/app/vault/x.md", "content")
self.assertEqual(len(captured), 0)
def test_t17_missing_credential_recovery_fails_closed(self):
"""T17: fehlendes Credential bei Recovery -> fail-closed."""
# Recovery = erneuter Write nach Crash. Ohne Credential -> fail-closed.
client = TolariaClient(base_url="http://fake")
captured = _capture_post(client)
with self.assertRaises(TolariaWriteError) as ctx:
client.write("/app/vault/recover.md", "content")
self.assertEqual(ctx.exception.reason_code, RC_AUTH_FAILURE)
self.assertEqual(len(captured), 0)
# ---------------------------------------------------------------------------
# Wrong-Scope Adversarial (PHASE 10)
# ---------------------------------------------------------------------------
class TestWrongScopeAdversarial(unittest.TestCase):
"""AUTH.3A PHASE 10: Wrong-Scope-Credentials muessen scheitern."""
def test_t18_save_credential_at_delete_executor_fails(self):
"""T18: SAVE credential beim DeleteExecutor -> scheitert."""
# DeleteExecutor bekommt einen Client mit NUR save_token.
# delete() (DELETE-Scope) muss fail-closed (kein DELETE-Token).
client = TolariaClient(base_url="http://fake", save_token=SAVE_TOKEN)
captured = _capture_post(client)
with self.assertRaises(TolariaWriteError) as ctx:
client.delete("/app/vault/x.md")
self.assertEqual(ctx.exception.reason_code, RC_AUTH_FAILURE)
self.assertEqual(len(captured), 0)
def test_t19_delete_credential_at_save_caller_fails(self):
"""T19: DELETE credential beim Save-Caller -> scheitert."""
# Save-Caller bekommt einen Client mit NUR delete_token.
# write() (SAVE-Scope) muss fail-closed (kein SAVE-Token).
client = TolariaClient(base_url="http://fake", delete_token=DELETE_TOKEN)
captured = _capture_post(client)
with self.assertRaises(TolariaWriteError) as ctx:
client.write("/app/vault/x.md", "content")
self.assertEqual(ctx.exception.reason_code, RC_AUTH_FAILURE)
self.assertEqual(len(captured), 0)
if __name__ == "__main__":
unittest.main()
Reference in a new issue View git blame Copy permalink