nexo312/trading-system-docs
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

AUTH.3A: C5 Caller Auth Integration (SAVE/DELETE Credential, fail-closed, Tests, Sensitivity, Gap-Doku)

Browse source 
- 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.
This commit is contained in:
Red Queen 2026-08-27 08:42:26 +00:00
parent 40bbc40a49
commit a11c1bbe53
6 changed files with 731 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

133
tolaria/c5-sync-service/auth3a-sensitivity.sh Normal file
View file

@ -0,0 +1,133 @@
#!/usr/bin/env bash
# AUTH.3A PHASE 11 — Test-Sensitivity: Beweise ROT bei gebrochener Auth-Semantik.
#
# Mutiert rq_c5c.py temporaer (A-H) und prueft, dass die AUTH.3A-Testsuite
# ROT wird. Keine hardcodierten PASS-Werte. Keine Produktion.
#
# A. fehlendes Token sendet Request trotzdem
# B. Authorization Header entfernt
# C. SAVE Caller laedt DELETE Token
# D. DeleteExecutor laedt SAVE Token
# E. Delete ohne Human Approval erlaubt
# F. Token wird geloggt
# G. Retry verliert Authorization
# H. Replay faellt auf unauthenticated zurueck
set -u
cd "$(dirname "$0")"
TARGET="rq_c5c.py"
TARGET2="rq_c5_delete.py"
TEST="test_c5_auth3a.py"
PY="python3 -m unittest ${TEST%.py}"
fail=0
run_mutation() {
local name="$1"; shift
local expect_red="$1"; shift
# Mutation anwenden
"$@" || { echo "MUTATION-APPLY-FAIL: $name"; fail=1; return; }
# Tests ausfuehren
local out
out=$($PY 2>&1)
local rc=$?
# Mutation zurueckrollen (aus .bak, NICHT git checkout — AUTH.3A-Aenderungen sind uncommitted)
cp "$TARGET.bak" "$TARGET"
cp "$TARGET2.bak" "$TARGET2"
if [ "$expect_red" = "RED" ]; then
if [ $rc -ne 0 ]; then
echo "OK [$name] -> ROT (rc=$rc)"
else
echo "FAIL[$name] -> GRUEN (sollte ROT sein)"
fail=1
fi
else
if [ $rc -eq 0 ]; then
echo "OK [$name] -> GRUEN"
else
echo "FAIL[$name] -> ROT (sollte GRUEN sein)"
fail=1
fi
fi
}
cp "$TARGET" "$TARGET.bak"
cp "$TARGET2" "$TARGET2.bak"
# A. fehlendes Token sendet Request trotzdem (entferne _require_token in write)
run_mutation "A-fehlendes-token-sendet-trotzdem" RED \
python3 - <<'EOF'
import re
p="rq_c5c.py"
s=open(p).read()
# write(): _require_token entfernen -> sendet ohne Token
s=s.replace("token = self._require_token(self.save_token, \"SAVE\")","token = self.save_token")
open(p,"w").write(s)
EOF
# B. Authorization Header entfernt (kein auth_token an _post)
run_mutation "B-auth-header-entfernt" RED \
python3 - <<'EOF'
p="rq_c5c.py"
s=open(p).read()
s=s.replace("if auth_token is not None:\n headers[\"Authorization\"] = f\"Bearer {auth_token}\"","pass")
open(p,"w").write(s)
EOF
# C. SAVE Caller laedt DELETE Token (write nutzt delete_token)
run_mutation "C-save-laedt-delete-token" RED \
python3 - <<'EOF'
p="rq_c5c.py"
s=open(p).read()
s=s.replace("token = self._require_token(self.save_token, \"SAVE\")","token = self._require_token(self.delete_token, \"SAVE\")")
open(p,"w").write(s)
EOF
# D. DeleteExecutor laedt SAVE Token (delete nutzt save_token)
run_mutation "D-delete-laedt-save-token" RED \
python3 - <<'EOF'
p="rq_c5c.py"
s=open(p).read()
s=s.replace("token = self._require_token(self.delete_token, \"DELETE\")","token = self._require_token(self.save_token, \"DELETE\")")
open(p,"w").write(s)
EOF
# E. Delete ohne Human Approval erlaubt (entferne _validate_approval in execute)
run_mutation "E-delete-ohne-approval-erlaubt" RED \
python3 - <<'EOF'
p="rq_c5_delete.py"
s=open(p).read()
s=s.replace("approval = self._validate_approval(commit_sha, change)","approval = {'approval_id': 'x', 'workflow_commit': commit_sha, 'object_id': change.get('object_id'), 'path': change.get('path_before') or change.get('path_after')}")
open(p,"w").write(s)
EOF
# F. Token wird geloggt (repr enthaelt Token)
run_mutation "F-token-geloggt" RED \
python3 - <<'EOF'
p="rq_c5c.py"
s=open(p).read()
# Fuege eine __repr__ ein, die den save_token exponiert (nach self.delete_token = delete_token)
s=s.replace(" self.save_token = save_token\n self.delete_token = delete_token",
" self.save_token = save_token\n self.delete_token = delete_token\n\n def __repr__(self):\n return f\"TolariaClient(save_token={self.save_token})\"")
open(p,"w").write(s)
EOF
# G. Retry verliert Authorization (write sendet kein Token beim 2. Aufruf)
run_mutation "G-retry-verliert-auth" RED \
python3 - <<'EOF'
p="rq_c5c.py"
s=open(p).read()
s=s.replace("token = self._require_token(self.save_token, \"SAVE\")","token = self._require_token(self.save_token, \"SAVE\") if not hasattr(self,'_retried') else None\n self._retried = True")
open(p,"w").write(s)
EOF
# H. Replay faellt auf unauthenticated zurueck (delete ohne Token)
run_mutation "H-replay-unauthenticated" RED \
python3 - <<'EOF'
p="rq_c5c.py"
s=open(p).read()
s=s.replace("token = self._require_token(self.delete_token, \"DELETE\")","token = None")
open(p,"w").write(s)
EOF
rm -f "$TARGET.bak" "$TARGET2.bak"
echo "=== SENSITIVITY RESULT: $([ $fail -eq 0 ] && echo ALL-RED || echo FAILURES) ==="
exit $fail

13
tolaria/c5-sync-service/rq_c5_cli.py
View file

@ -51,6 +51,8 @@ from rq_c5c import (
assert_no_master_write, assert_no_master_write,
DEFAULT_TOLARIA_BASE, DEFAULT_TOLARIA_BASE,
ENV_TOLARIA_BASE, ENV_TOLARIA_BASE,
ENV_TOLARIA_SAVE_TOKEN,
ENV_TOLARIA_DELETE_TOKEN,
) )
from rq_c5d import ( from rq_c5d import (
C5DEngine, C5DEngine,
@ -341,7 +343,12 @@ def cmd_c5e_guarantees(args: argparse.Namespace) -> int:
def _delete_executor(args: argparse.Namespace) -> DeleteExecutor: def _delete_executor(args: argparse.Namespace) -> DeleteExecutor:
store = _store(args.db) store = _store(args.db)
client = TolariaClient(base_url=args.tolaria_base) # AUTH.3A: DELETE-Caller liest NUR das DELETE-Credential aus der
# Runtime-Konfiguration (Least Privilege, AUTH.1 §3). Kein SAVE-Token.
client = TolariaClient(
base_url=args.tolaria_base,
delete_token=os.environ.get(ENV_TOLARIA_DELETE_TOKEN),
)
return DeleteExecutor(store, client) return DeleteExecutor(store, client)
@ -417,7 +424,7 @@ def cmd_c5_delete_approve(args: argparse.Namespace) -> int:
def cmd_c5_delete_execute(args: argparse.Namespace) -> int: def cmd_c5_delete_execute(args: argparse.Namespace) -> int:
"""Fuehrt den kontrollierten Tolaria-DELETE aus (verlangt persistierte Approval).""" """Fuehrt den kontrollierten Tolaria-DELETE aus (verlangt persistierte Approval)."""
store = _store(args.db) store = _store(args.db)
executor = DeleteExecutor(store, TolariaClient(base_url=args.tolaria_base)) executor = _delete_executor(args)
try: try:
result = executor.execute(args.commit) result = executor.execute(args.commit)
print(json.dumps(result, ensure_ascii=False, indent=2)) print(json.dumps(result, ensure_ascii=False, indent=2))
@ -451,7 +458,7 @@ def cmd_c5_delete_status(args: argparse.Namespace) -> int:
def cmd_c5_delete_replay(args: argparse.Namespace) -> int: def cmd_c5_delete_replay(args: argparse.Namespace) -> int:
"""Idempotenter Replay nach Crash (verlangt persistierte Approval).""" """Idempotenter Replay nach Crash (verlangt persistierte Approval)."""
store = _store(args.db) store = _store(args.db)
executor = DeleteExecutor(store, TolariaClient(base_url=args.tolaria_base)) executor = _delete_executor(args)
try: try:
result = executor.replay(args.commit) result = executor.replay(args.commit)
print(json.dumps(result, ensure_ascii=False, indent=2)) print(json.dumps(result, ensure_ascii=False, indent=2))

72
tolaria/c5-sync-service/rq_c5c.py
View file

@ -91,6 +91,15 @@ from rq_c5b import (
DEFAULT_TOLARIA_BASE = "http://187.124.31.123:5173/api/vault" DEFAULT_TOLARIA_BASE = "http://187.124.31.123:5173/api/vault"
ENV_TOLARIA_BASE = "C5_TOLARIA_BASE" ENV_TOLARIA_BASE = "C5_TOLARIA_BASE"
# AUTH.3A ENV-Contract (PHASE 3): Eindeutige Namen, festgelegt in AUTH.2
# (vite.config.ts: process.env.TOLARIA_SAVE_TOKEN / TOLARIA_DELETE_TOKEN).
# AUTH.3A implementiert NUR die code-seitige Faehigkeit, diese aus der
# Runtime-Konfiguration zu lesen. Es werden KEINE echten ENV-Werte gesetzt.
# Least Privilege: SAVE-Caller liest NUR TOLARIA_SAVE_TOKEN, DELETE-Caller
# liest NUR TOLARIA_DELETE_TOKEN. Kein Prozess laedt beide automatisch.
ENV_TOLARIA_SAVE_TOKEN = "TOLARIA_SAVE_TOKEN"
ENV_TOLARIA_DELETE_TOKEN = "TOLARIA_DELETE_TOKEN"
# Vault-Pfad-Praefix (verbindlich — ohne Praefix liest /content das Tolaria-eigene # Vault-Pfad-Praefix (verbindlich — ohne Praefix liest /content das Tolaria-eigene
# README statt des Vault-Objekts -> false drift) # README statt des Vault-Objekts -> false drift)
VAULT_PREFIX = "/app/vault" VAULT_PREFIX = "/app/vault"
@ -207,18 +216,42 @@ class TolariaClient:
Keine generische Agent-Write-Funktion. Kein Search-Rebuild-Aufruf. Keine generische Agent-Write-Funktion. Kein Search-Rebuild-Aufruf.
""" """
def __init__(self, base_url: Optional[str] = None, timeout: float = 15.0): def __init__(
self,
base_url: Optional[str] = None,
timeout: float = 15.0,
save_token: Optional[str] = None,
delete_token: Optional[str] = None,
):
self.base_url = (base_url or os.environ.get(ENV_TOLARIA_BASE) self.base_url = (base_url or os.environ.get(ENV_TOLARIA_BASE)
or DEFAULT_TOLARIA_BASE).rstrip("/") or DEFAULT_TOLARIA_BASE).rstrip("/")
self.timeout = timeout self.timeout = timeout
# AUTH.3A: Explizite Credential-Injection (kein verstecktes globales
# Credential, kein Default-Token). SAVE- und DELETE-Credential sind
# strikt getrennt (AUTH.1 §3). Fehlende/leere Werte -> fail-closed
# beim jeweiligen mutierenden Aufruf (AUTH.1 §4).
self.save_token = save_token
self.delete_token = delete_token
# -- HTTP-Helfer -------------------------------------------------------- # -- HTTP-Helfer --------------------------------------------------------
def _post(self, endpoint: str, payload: Dict[str, Any]) -> Dict[str, Any]: def _post(
self,
endpoint: str,
payload: Dict[str, Any],
auth_token: Optional[str] = None,
) -> Dict[str, Any]:
url = f"{self.base_url}/{endpoint.lstrip('/')}" url = f"{self.base_url}/{endpoint.lstrip('/')}"
data = json.dumps(payload).encode("utf-8") data = json.dumps(payload).encode("utf-8")
headers: Dict[str, str] = {"Content-Type": "application/json"}
# AUTH.3A: Authorization-Header NUR wenn ein Token explizit uebergeben
# wird (mutierende SAVE/DELETE-Aufrufe). READ-Aufrufe senden KEIN
# Credential (Least Privilege, AUTH.1 §2). Der Token-Wert wird nie
# geloggt (AUTH.1 §6).
if auth_token is not None:
headers["Authorization"] = f"Bearer {auth_token}"
req = urllib.request.Request( req = urllib.request.Request(
url, data=data, headers={"Content-Type": "application/json"}, url, data=data, headers=headers,
method="POST", method="POST",
) )
try: try:
@ -264,9 +297,31 @@ class TolariaClient:
# -- Write (NUR C5C-Writer) -------------------------------------------- # -- Write (NUR C5C-Writer) --------------------------------------------
def _require_token(self, token: Optional[str], scope: str) -> str:
"""Fail-closed: fehlendes/leeres/malformed Credential -> kein HTTP.
AUTH.1 §4: Ein mutierender Request darf NIE ohne gültiges Credential
gesendet werden. Fehlendes oder leeres Token -> lokaler Abbruch
(TolariaWriteError, RC_AUTH_FAILURE), HTTP wird NICHT aufgerufen.
"""
if not token or not isinstance(token, str) or not token.strip():
raise TolariaWriteError(
f"Tolaria {scope}-Credential fehlt oder ist leer "
f"(fail-closed, kein Request gesendet)",
RC_AUTH_FAILURE,
)
return token
def write(self, vault_path: str, content: str) -> Dict[str, Any]: def write(self, vault_path: str, content: str) -> Dict[str, Any]:
"""Schreibt ein Vault-Objekt (POST /save). Isolierte Write-Komponente.""" """Schreibt ein Vault-Objekt (POST /save). Isolierte Write-Komponente.
resp = self._post("save", {"path": vault_path, "content": content})
AUTH.3A: SAVE-Scope. Sendet Authorization: Bearer <save_token>.
Fail-closed: fehlendes/leeres SAVE-Credential -> lokaler Abbruch,
HTTP wird NICHT aufgerufen (AUTH.1 §4).
"""
token = self._require_token(self.save_token, "SAVE")
resp = self._post("save", {"path": vault_path, "content": content},
auth_token=token)
# HTTP 2xx + JSON null / leerer Payload -> erfolgreicher Transport. # HTTP 2xx + JSON null / leerer Payload -> erfolgreicher Transport.
# Nur hier (write /save) wird dieser Fall als Erfolg gewertet; der # Nur hier (write /save) wird dieser Fall als Erfolg gewertet; der
# nachgelagerte Read-Back (verify) bleibt zwingend für einen C5C-Step. # nachgelagerte Read-Back (verify) bleibt zwingend für einen C5C-Step.
@ -289,8 +344,13 @@ class TolariaClient:
Endpoint-/Payload-Contract (POST /delete mit {"path": ...}) ist aus dem Endpoint-/Payload-Contract (POST /delete mit {"path": ...}) ist aus dem
vorhandenen Tolaria-Code/Router und der dokumentierten Incident-Evidence vorhandenen Tolaria-Code/Router und der dokumentierten Incident-Evidence
bestimmt. KEINE produktive Endpoint-Probe (Incident-Regel). bestimmt. KEINE produktive Endpoint-Probe (Incident-Regel).
AUTH.3A: DELETE-Scope. Sendet Authorization: Bearer <delete_token>.
Fail-closed: fehlendes/leeres DELETE-Credential -> lokaler Abbruch,
HTTP wird NICHT aufgerufen (AUTH.1 §4).
""" """
resp = self._post("delete", {"path": vault_path}) token = self._require_token(self.delete_token, "DELETE")
resp = self._post("delete", {"path": vault_path}, auth_token=token)
if resp is None: if resp is None:
resp = {} resp = {}
if "error" in resp: if "error" in resp:

444
tolaria/c5-sync-service/test_c5_auth3a.py Normal file
View file

@ -0,0 +1,444 @@
#!/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()

27
tolaria/c5-sync-service/test_c5c.py
View file

@ -962,7 +962,7 @@ class TestTolariaReadNotFoundSemantic(unittest.TestCase):
def _post(self, status: int, body: str, reason_code=None): def _post(self, status: int, body: str, reason_code=None):
"""Liefert einen Callable, der ein TolariaWriteError/JSON wirft/liefert.""" """Liefert einen Callable, der ein TolariaWriteError/JSON wirft/liefert."""
def fake_post(endpoint, payload): def fake_post(endpoint, payload, auth_token=None):
if status == 200: if status == 200:
return json.loads(body) return json.loads(body)
raise TolariaWriteError( raise TolariaWriteError(
@ -1002,7 +1002,7 @@ class TestTolariaReadNotFoundSemantic(unittest.TestCase):
# E) HTTP 500 -> weiterhin technischer Fehler (retrybar, Unavailable) # E) HTTP 500 -> weiterhin technischer Fehler (retrybar, Unavailable)
def test_e_500_is_unavailable(self): def test_e_500_is_unavailable(self):
def fake_500(endpoint, payload): def fake_500(endpoint, payload, auth_token=None):
raise TolariaUnavailableError("Tolaria HTTP 500", "TOLARIA_UNAVAILABLE") raise TolariaUnavailableError("Tolaria HTTP 500", "TOLARIA_UNAVAILABLE")
with mock.patch.object(self.client, "_post", side_effect=fake_500): with mock.patch.object(self.client, "_post", side_effect=fake_500):
with self.assertRaises(TolariaUnavailableError): with self.assertRaises(TolariaUnavailableError):
@ -1035,11 +1035,16 @@ class TestTolariaWriteNullResponseSemantic(unittest.TestCase):
Read-Back bleibt zwingend; 4xx/5xx weiterhin fail-closed.""" Read-Back bleibt zwingend; 4xx/5xx weiterhin fail-closed."""
def setUp(self): def setUp(self):
self.client = TolariaClient(base_url="http://fake") # AUTH.3A: write() ist fail-closed ohne SAVE-Credential. Fuer die
# /save-Semantik-Tests wird ein synthetisches SAVE-Token injiziert.
self.client = TolariaClient(
base_url="http://fake",
save_token="auth3a-save-synthetic-token-0001",
)
def _post(self, status: int, body: str, reason_code=None): def _post(self, status: int, body: str, reason_code=None):
"""Liefert einen Callable, der ein TolariaWriteError/JSON wirft/liefert.""" """Liefert einen Callable, der ein TolariaWriteError/JSON wirft/liefert."""
def fake_post(endpoint, payload): def fake_post(endpoint, payload, auth_token=None):
if status == 200: if status == 200:
return json.loads(body) return json.loads(body)
raise TolariaWriteError( raise TolariaWriteError(
@ -1054,7 +1059,7 @@ class TestTolariaWriteNullResponseSemantic(unittest.TestCase):
# A) /save HTTP 200 + JSON null -> write() kein Crash, Save-Step erfolgreich # A) /save HTTP 200 + JSON null -> write() kein Crash, Save-Step erfolgreich
def test_a_save_200_null_no_crash(self): def test_a_save_200_null_no_crash(self):
def fake_post(endpoint, payload): def fake_post(endpoint, payload, auth_token=None):
self.assertEqual(endpoint, "save") self.assertEqual(endpoint, "save")
return None # HTTP 200 + JSON null return None # HTTP 200 + JSON null
with mock.patch.object(self.client, "_post", side_effect=fake_post): with mock.patch.object(self.client, "_post", side_effect=fake_post):
@ -1064,13 +1069,13 @@ class TestTolariaWriteNullResponseSemantic(unittest.TestCase):
# B) /save HTTP 200 + {} -> weiterhin erfolgreich # B) /save HTTP 200 + {} -> weiterhin erfolgreich
def test_b_save_200_empty_ok(self): def test_b_save_200_empty_ok(self):
with mock.patch.object(self.client, "_post", side_effect=lambda ep, pl: {}): with mock.patch.object(self.client, "_post", side_effect=lambda ep, pl, auth_token=None: {}):
resp = self.client.write("/app/vault/x.md", self._content()) resp = self.client.write("/app/vault/x.md", self._content())
self.assertIsInstance(resp, dict) self.assertIsInstance(resp, dict)
# C) /save HTTP 200 + normaler Success-Payload -> erfolgreich # C) /save HTTP 200 + normaler Success-Payload -> erfolgreich
def test_c_save_200_payload_ok(self): def test_c_save_200_payload_ok(self):
with mock.patch.object(self.client, "_post", side_effect=lambda ep, pl: {"ok": True}): with mock.patch.object(self.client, "_post", side_effect=lambda ep, pl, auth_token=None: {"ok": True}):
resp = self.client.write("/app/vault/x.md", self._content()) resp = self.client.write("/app/vault/x.md", self._content())
self.assertEqual(resp, {"ok": True}) self.assertEqual(resp, {"ok": True})
@ -1089,7 +1094,7 @@ class TestTolariaWriteNullResponseSemantic(unittest.TestCase):
# F) /save 500 -> technischer Fehler / Retry Contract (Unavailable) # F) /save 500 -> technischer Fehler / Retry Contract (Unavailable)
def test_f_save_500_unavailable(self): def test_f_save_500_unavailable(self):
def fake_500(endpoint, payload): def fake_500(endpoint, payload, auth_token=None):
raise TolariaUnavailableError("Tolaria HTTP 500", "TOLARIA_UNAVAILABLE") raise TolariaUnavailableError("Tolaria HTTP 500", "TOLARIA_UNAVAILABLE")
with mock.patch.object(self.client, "_post", side_effect=fake_500): with mock.patch.object(self.client, "_post", side_effect=fake_500):
with self.assertRaises(TolariaUnavailableError): with self.assertRaises(TolariaUnavailableError):
@ -1100,7 +1105,7 @@ class TestTolariaWriteNullResponseSemantic(unittest.TestCase):
content = self._content() content = self._content()
calls = {"n": 0} calls = {"n": 0}
def fake_post(endpoint, payload): def fake_post(endpoint, payload, auth_token=None):
calls["n"] += 1 calls["n"] += 1
if endpoint == "save": if endpoint == "save":
return None # HTTP 200 + JSON null return None # HTTP 200 + JSON null
@ -1119,7 +1124,7 @@ class TestTolariaWriteNullResponseSemantic(unittest.TestCase):
content = self._content("TARGET") content = self._content("TARGET")
drifted = self._content("DRIFTED") drifted = self._content("DRIFTED")
def fake_post(endpoint, payload): def fake_post(endpoint, payload, auth_token=None):
if endpoint == "save": if endpoint == "save":
return None return None
if endpoint == "content": if endpoint == "content":
@ -1137,7 +1142,7 @@ class TestTolariaWriteNullResponseSemantic(unittest.TestCase):
content = self._content() content = self._content()
n = {"n": 0} n = {"n": 0}
def fake_post(endpoint, payload): def fake_post(endpoint, payload, auth_token=None):
n["n"] += 1 n["n"] += 1
if endpoint == "save": if endpoint == "save":
return None return None

62
tolaria/tolaria-write-auth/AUTH3A_HUMAN_APPROVAL_AUTHENTICITY_GAP.md Normal file
View file

@ -0,0 +1,62 @@
# AUTH.3A — HUMAN APPROVAL AUTHENTICITY GAP (Dokumentation)
**Status:** OPEN (nicht in AUTH.3A repariert — bewusste Scope-Begrenzung)
**Mission:** AUTH.3A (C5 Caller Auth Integration)
**Datum:** 2026-08-27
## Gap-Beschreibung
Der bekannte PRE_HERMES-Gap:
- `approved_by="human:christian"` ist ein **frei waehlbares CLI-Argument**
(`rq_c5_cli.py`, `--approved-by` mit Default `"human:christian"`).
- Es gibt **keine kryptographische Bindung** der Human-Approval an Christian
(keine Signatur, kein Nonce-Beweis, kein Out-of-Band-Verifikationskanal).
Das bedeutet: Ein Angreifer mit Schreibzugriff auf die C5-DB (oder auf den
CLI-Aufruf) koennte eine Approval mit `approved_by="human:christian"` faelschen,
ohne dass dies kryptographisch nachweisbar waere.
## Scope-Begrenzung (AUTH.3A)
AUTH.3A repariert diesen Gap **NICHT**. Gruende:
1. AUTH.3A ist strikt auf **C5-Caller-Auth-Integration** begrenzt
(SAVE-/DELETE-Credential fuer die legitimen Writer).
2. Die Approval-Authenticity ist ein **eigenstaendiges Security-Thema**
(eigene Mission, eigene Freigabe durch Christian).
3. Keine Scope-Ausweitung ueber die AUTH.3A-Freigabe hinaus.
## Was AUTH.3A sicherstellt (in Bezug auf DELETE)
AUTH.3A fuegt dem DELETE-Pfad ein **zweites, unabhaengiges Gate** hinzu:
- **Gate 1 (bestehend):** gueltige persistierte Human-Approval
(`DeleteExecutor._validate_approval` — Approval existiert, APPROVED,
bindet exakt Commit/Objekt/Pfad).
- **Gate 2 (AUTH.3A, neu):** gueltiges DELETE-Credential
(`TolariaClient._require_token` — fail-closed, kein Request ohne Token).
Beide Gates sind **zwingend und unabhaengig**:
- DELETE-Credential ersetzt NIE die Human-Approval.
- Human-Approval ersetzt NIE das DELETE-Credential.
- Fehlt eines von beiden → FAIL CLOSED (kein Request gesendet).
## Verbleibende Luecke (nicht in AUTH.3A)
Die **Authentizitaet** der Human-Approval selbst (dass sie wirklich von
Christian stammt) bleibt ungeloest. Das DELETE-Credential schuetzt vor
unautorisierten *Ausfuehrern*, aber nicht vor einer *gefaelschten Approval*.
## Kennung
```
HUMAN_APPROVAL_AUTHENTICITY_GAP = OPEN
```
## Naechste Schritte (separate Freigabe erforderlich)
- Eigene Security-Mission zur kryptographischen Bindung der Human-Approval
an Christian (z. B. Signatur, Nonce-Beweis, Out-of-Band-Verifikation).
- Kein Teil von AUTH.3A / AUTH.3B / AUTH.4.