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

AUTH.3D: Human DELETE Approval Authenticity (Ed25519, verify-only)

Browse source 
- approval_payload: kanonischer, deterministischer Approval-Payload
- approval_signature: Ed25519 sign/verify (Christian=Private Key, Executor=Public Key only)
- approval_verifier: verify-only, alle Bindings fail-closed
- approval_state: Lifecycle CREATED->CONSUMED, Single-Use, Reservation, OUTCOME_UNKNOWN
- test_approval_auth3d: T1-T30 + adversarial (50 Tests)
- test_approval_helpers: synthetische Test-Keypairs
- sensitivity_proof: Mutationen A-J machen Tests ROT

Nur synthetische Test-Keypairs. Kein produktives Deployment.
DELETE_OPERATION_ACTIVATION bleibt BLOCKED bis AUTH.3D geprueft.
This commit is contained in:
Red Queen 2026-08-27 09:36:21 +00:00
parent a11c1bbe53
commit 9c8d239ae5
7 changed files with 1462 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

194
tolaria/c5-sync-service/approval_payload.py Normal file
View file

@ -0,0 +1,194 @@
"""
AUTH.3D — approval_payload.py
Kanonischer Human DELETE Approval Payload.
Deterministische Canonicalization: feste Feldreihenfolge, keine Duplikate,
keine Whitespace-Varianz, UTF-8, keine Unicode-Normalisierung (exakte Bytes),
Pfad-Normalisierung. Keine Signatur über frei formatierte Texte.
NUR isolierter Code + Tests. KEIN produktives Deployment.
"""
from __future__ import annotations
import hashlib
import json
import re
from typing import Any, Dict, List, Optional
APPROVAL_VERSION = 1
OPERATION_DELETE = "DELETE"
# Feste Feldreihenfolge für die Canonicalization (deterministisch).
_CANONICAL_FIELDS: List[str] = [
"approval_version",
"approval_id",
"mission_id",
"delete_request_id",
"operation",
"object_id",
"vault_path",
"expected_commit",
"expected_provenance_hash",
"nonce",
"issued_at",
"expires_at",
]
_REQUIRED_FIELDS: set = set(_CANONICAL_FIELDS)
# Pfad-Normalisierung: keine doppelten Slashes, kein "./", kein "..", kein trailing slash.
_DOUBLE_SLASH = re.compile(r"//+")
_DOT_SEGMENT = re.compile(r"(^|/)\.(/|$)")
_DOTDOT_SEGMENT = re.compile(r"(^|/)\.\.(/|$)")
class ApprovalPayloadError(ValueError):
"""Fehler bei Payload-Erzeugung/Canonicalization."""
def _normalize_path(path: str) -> str:
"""Normalisiert einen Vault-Pfad deterministisch (keine Varianz)."""
if not isinstance(path, str) or not path:
raise ApprovalPayloadError("vault_path muss ein nicht-leerer String sein")
p = path.strip()
if not p.startswith("/"):
p = "/" + p
p = _DOUBLE_SLASH.sub("/", p)
p = _DOT_SEGMENT.sub("/", p)
p = _DOTDOT_SEGMENT.sub("/", p)
# trailing slash entfernen (ausser Root)
if len(p) > 1 and p.endswith("/"):
p = p.rstrip("/")
return p
def _validate_iso8601_utc(value: str, field: str) -> str:
"""Validiert ISO-8601-UTC (z.B. 2026-08-27T09:00:00Z)."""
if not isinstance(value, str) or not value:
raise ApprovalPayloadError(f"{field} muss ein ISO-8601-UTC-String sein")
# Einfache, strikte Form: YYYY-MM-DDTHH:MM:SSZ
if not re.match(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$", value):
raise ApprovalPayloadError(f"{field} muss ISO-8601-UTC (…Z) sein: {value!r}")
return value
def _validate_hex(value: str, field: str, length: int) -> str:
if not isinstance(value, str) or not re.fullmatch(r"[0-9a-fA-F]{" + str(length) + r"}", value):
raise ApprovalPayloadError(f"{field} muss ein {length}-Zeichen-Hex-String sein")
return value.lower()
def _validate_uuid(value: str, field: str) -> str:
if not isinstance(value, str) or not re.fullmatch(
r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}", value
):
raise ApprovalPayloadError(f"{field} muss eine UUID sein")
return value.lower()
def build_payload(
*,
approval_id: str,
mission_id: str,
delete_request_id: str,
object_id: str,
vault_path: str,
expected_commit: str,
expected_provenance_hash: str,
nonce: str,
issued_at: str,
expires_at: str,
approval_version: int = APPROVAL_VERSION,
operation: str = OPERATION_DELETE,
) -> Dict[str, Any]:
"""
Baut und validiert einen kanonischen Approval Payload.
Wirft ApprovalPayloadError bei ungültigen Feldern. Gibt ein dict mit
exakt den 12 kanonischen Feldern zurück (keine optionalen Felder).
"""
if approval_version != APPROVAL_VERSION:
raise ApprovalPayloadError(f"Unbekannte approval_version: {approval_version}")
if operation != OPERATION_DELETE:
raise ApprovalPayloadError(f"operation muss {OPERATION_DELETE} sein, nicht {operation!r}")
payload: Dict[str, Any] = {
"approval_version": approval_version,
"approval_id": _validate_uuid(approval_id, "approval_id"),
"mission_id": _validate_uuid(mission_id, "mission_id"),
"delete_request_id": _validate_uuid(delete_request_id, "delete_request_id"),
"operation": operation,
"object_id": _validate_uuid(object_id, "object_id"),
"vault_path": _normalize_path(vault_path),
"expected_commit": _validate_hex(expected_commit, "expected_commit", 40),
"expected_provenance_hash": _validate_hex(expected_provenance_hash, "expected_provenance_hash", 64),
"nonce": _validate_uuid(nonce, "nonce"),
"issued_at": _validate_iso8601_utc(issued_at, "issued_at"),
"expires_at": _validate_iso8601_utc(expires_at, "expires_at"),
}
return payload
def canonicalize(payload: Dict[str, Any]) -> bytes:
"""
Deterministische Canonicalization des Payloads zu Bytes.
- Feste Feldreihenfolge (_CANONICAL_FIELDS)
- Keine Duplikate, keine optionalen Felder
- UTF-8, keine Whitespace-Varianz (compact separators)
- Keine Unicode-Normalisierung (exakte Bytes)
"""
if not isinstance(payload, dict):
raise ApprovalPayloadError("Payload muss ein dict sein")
# Nur die kanonischen Felder, in fester Reihenfolge.
ordered: Dict[str, Any] = {}
for field in _CANONICAL_FIELDS:
if field not in payload:
raise ApprovalPayloadError(f"Payload fehlt Pflichtfeld: {field}")
ordered[field] = payload[field]
# Keine zusätzlichen Felder erlauben (verhindert canonicalization ambiguity).
extra = set(payload.keys()) - set(_CANONICAL_FIELDS)
if extra:
raise ApprovalPayloadError(f"Payload hat unerlaubte Felder: {sorted(extra)}")
# compact separators -> keine Whitespace-Varianz; sort_keys=False (Reihenfolge fix).
text = json.dumps(ordered, ensure_ascii=True, separators=(",", ":"), sort_keys=False)
return text.encode("utf-8")
def payload_hash(payload: Dict[str, Any]) -> str:
"""SHA-256 über die kanonisierten Payload-Bytes (hex)."""
return hashlib.sha256(canonicalize(payload)).hexdigest()
def parse_payload(raw: str) -> Dict[str, Any]:
"""
Parst einen JSON-String in einen Payload und validiert ihn.
- Duplicate JSON keys -> Fehler (verhindert canonicalization ambiguity)
- Malformed JSON -> Fehler
"""
if not isinstance(raw, str) or not raw.strip():
raise ApprovalPayloadError("Payload muss ein nicht-leerer JSON-String sein")
try:
obj = json.loads(raw, object_pairs_hook=_reject_duplicate_keys)
except json.JSONDecodeError as e:
raise ApprovalPayloadError(f"Malformed JSON: {e}") from e
if not isinstance(obj, dict):
raise ApprovalPayloadError("Payload muss ein JSON-Objekt sein")
# Validierung + Canonicalization (wirft bei ungültigen Feldern).
canonicalize(obj)
return obj
def _reject_duplicate_keys(pairs: List[tuple]) -> Dict[str, Any]:
"""Reject duplicate JSON keys (canonicalization ambiguity)."""
d: Dict[str, Any] = {}
for k, v in pairs:
if k in d:
raise ApprovalPayloadError(f"Duplicate JSON key: {k}")
d[k] = v
return d

124
tolaria/c5-sync-service/approval_signature.py Normal file
View file

@ -0,0 +1,124 @@
"""
AUTH.3D — approval_signature.py
Ed25519 Signatur (RFC 8032) für Human DELETE Approvals.
- Christian besitzt den PRIVATE KEY (NIE im Stack).
- DELETE Executor kennt nur den PUBLIC KEY (verify-only).
- RQ/Hermes/SAVE/DELETE-Executor besitzen KEINEN Private Key.
NUR isolierter Code + Tests mit synthetischen Test-Keypairs.
KEIN echter Key wird erzeugt. KEIN produktives Deployment.
"""
from __future__ import annotations
import base64
import hashlib
from typing import Optional, Tuple
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
Ed25519PrivateKey,
Ed25519PublicKey,
)
from approval_payload import canonicalize
# Signaturformat: Ed25519 (64 Bytes), Base64-URL-encoded.
SIGNATURE_ALGORITHM = "Ed25519"
class ApprovalSignatureError(ValueError):
"""Fehler bei Signatur-Erzeugung/Verifikation."""
def generate_test_keypair() -> Tuple[Ed25519PrivateKey, Ed25519PublicKey]:
"""
Erzeugt ein SYNTHETISCHES Test-Keypair (NUR für Tests).
NIEMALS für produktive Approvals verwenden. Kein echter Key.
"""
private = Ed25519PrivateKey.generate()
return private, private.public_key()
def key_id(public_key: Ed25519PublicKey) -> str:
"""KEY_ID = SHA-256 über die Public-Key-Bytes (hex, kurz)."""
raw = public_key.public_bytes(
encoding=serialization.Encoding.Raw,
format=serialization.PublicFormat.Raw,
)
return hashlib.sha256(raw).hexdigest()
def sign(private_key: Ed25519PrivateKey, payload: dict) -> str:
"""
Signiert den kanonisierten Payload mit dem Private Key.
Gibt die Signatur als Base64-URL-String zurück.
"""
canonical = canonicalize(payload)
sig = private_key.sign(canonical)
return base64.urlsafe_b64encode(sig).decode("ascii")
def verify(public_key: Ed25519PublicKey, payload: dict, signature: str) -> bool:
"""
Verifiziert die Signatur über den kanonisierten Payload.
Gibt True bei gültiger Signatur, False bei ungültiger/malformed.
FAIL CLOSED: jede Abweichung -> False.
"""
if not isinstance(signature, str) or not signature:
return False
try:
sig_bytes = base64.urlsafe_b64decode(signature.encode("ascii"))
except Exception:
return False
if len(sig_bytes) != 64:
return False
try:
canonical = canonicalize(payload)
except Exception:
return False
try:
public_key.verify(sig_bytes, canonical)
return True
except InvalidSignature:
return False
except Exception:
return False
def public_key_to_pem(public_key: Ed25519PublicKey) -> bytes:
"""Serialisiert den Public Key zu PEM (für Verifier-Konfiguration)."""
return public_key.public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo,
)
def public_key_from_pem(pem: bytes) -> Ed25519PublicKey:
"""Lädt einen Public Key aus PEM."""
try:
return serialization.load_pem_public_key(pem) # type: ignore[return-value]
except Exception as e:
raise ApprovalSignatureError(f"Ungültiger Public Key PEM: {e}") from e
def private_key_to_pem(private_key: Ed25519PrivateKey) -> bytes:
"""Serialisiert den Private Key zu PEM (NUR für Christians Gerät / Tests)."""
return private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption(),
)
def private_key_from_pem(pem: bytes) -> Ed25519PrivateKey:
"""Lädt einen Private Key aus PEM (NUR Christians Gerät / Tests)."""
try:
return serialization.load_pem_private_key(pem, password=None) # type: ignore[return-value]
except Exception as e:
raise ApprovalSignatureError(f"Ungültiger Private Key PEM: {e}") from e

165
tolaria/c5-sync-service/approval_state.py Normal file
View file

@ -0,0 +1,165 @@
"""
AUTH.3D — approval_state.py
Approval Lifecycle + Single-Use + Reservation (TOCTOU/Race Safety).
Lifecycle: CREATED → VERIFIED → RESERVED → EXECUTED → CONSUMED
- RESERVED: atomare Reservation vor dem Tolaria-DELETE (verhindert Race).
- CONSUMED: endgültig, niemals erneut ausführbar.
- OUTCOME_UNKNOWN: kein blinder Retry; manuelle Klärung.
NUR isolierter Code + Tests. KEIN produktives Deployment.
"""
from __future__ import annotations
import threading
import uuid
from typing import Any, Dict, Optional
# Approval States
ST_CREATED = "CREATED"
ST_VERIFIED = "VERIFIED"
ST_RESERVED = "RESERVED"
ST_EXECUTED = "EXECUTED"
ST_CONSUMED = "CONSUMED"
ST_INVALID = "INVALID"
ST_EXPIRED = "EXPIRED"
ST_REVOKED = "REVOKED"
ST_FAILED = "FAILED"
ST_OUTCOME_UNKNOWN = "OUTCOME_UNKNOWN"
# Terminal (nicht mehr ausführbar) States
_TERMINAL = {ST_CONSUMED, ST_INVALID, ST_EXPIRED, ST_REVOKED, ST_FAILED}
# Übergänge, die eine Ausführung erlauben
_EXECUTABLE = {ST_CREATED, ST_VERIFIED, ST_RESERVED}
class ApprovalStateError(Exception):
"""Ungültiger State-Übergang (FAIL CLOSED)."""
class ApprovalStateStore:
"""
In-Memory State Store für isolierte Tests.
In Produktion würde dies die C5-DB (delete_approvals) erweitern.
Hier: reine, thread-sichere In-Memory-Implementierung für Tests.
"""
def __init__(self):
self._lock = threading.RLock()
self._approvals: Dict[str, Dict[str, Any]] = {}
self._used_nonces: set = set()
def create(self, approval_id: str, nonce: str) -> Dict[str, Any]:
with self._lock:
if approval_id in self._approvals:
raise ApprovalStateError(f"Approval {approval_id} existiert bereits")
rec = {
"approval_id": approval_id,
"nonce": nonce,
"state": ST_CREATED,
"attempt_id": None,
"idempotency_key": None,
"consumed_at": None,
"result": None,
}
self._approvals[approval_id] = rec
return dict(rec)
def get(self, approval_id: str) -> Optional[Dict[str, Any]]:
with self._lock:
rec = self._approvals.get(approval_id)
return dict(rec) if rec else None
def transition(self, approval_id: str, new_state: str, **fields) -> Dict[str, Any]:
with self._lock:
rec = self._approvals.get(approval_id)
if rec is None:
raise ApprovalStateError(f"Approval {approval_id} existiert nicht")
if new_state == ST_CONSUMED:
if rec["state"] not in _EXECUTABLE:
raise ApprovalStateError(
f"Approval {approval_id} in State {rec['state']} kann nicht CONSUMED werden"
)
rec["state"] = new_state
rec.update(fields)
if new_state == ST_CONSUMED:
rec["consumed_at"] = fields.get("consumed_at")
return dict(rec)
def reserve(
self, approval_id: str, attempt_id: str, idempotency_key: str
) -> Dict[str, Any]:
"""
Atomare Reservation: APPROVED/RESERVED → RESERVED (nur wenn noch ausführbar).
Verhindert TOCTOU-Race: nur EINE Reservation pro Approval.
"""
with self._lock:
rec = self._approvals.get(approval_id)
if rec is None:
raise ApprovalStateError(f"Approval {approval_id} existiert nicht")
if rec["state"] not in _EXECUTABLE:
raise ApprovalStateError(
f"Approval {approval_id} in State {rec['state']} kann nicht reserviert werden"
)
# Idempotenz: gleicher Attempt wird nicht doppelt reserviert.
if rec.get("idempotency_key") == idempotency_key:
return dict(rec)
# Bereits reserviert (anderer Attempt) -> kein zweiter Reservation.
if rec["state"] == ST_RESERVED:
raise ApprovalStateError(
f"Approval {approval_id} ist bereits reserviert (Attempt {rec.get('attempt_id')})"
)
rec["state"] = ST_RESERVED
rec["attempt_id"] = attempt_id
rec["idempotency_key"] = idempotency_key
return dict(rec)
def mark_used_nonce(self, nonce: str) -> None:
with self._lock:
self._used_nonces.add(nonce)
def is_nonce_used(self, nonce: str) -> bool:
with self._lock:
return nonce in self._used_nonces
def consume(self, approval_id: str, result: str) -> Dict[str, Any]:
"""Markiert als CONSUMED (endgültig)."""
return self.transition(approval_id, ST_CONSUMED, result=result)
def mark_outcome_unknown(self, approval_id: str) -> Dict[str, Any]:
"""Markiert als OUTCOME_UNKNOWN (kein blinder Retry)."""
return self.transition(approval_id, ST_OUTCOME_UNKNOWN)
def recover_from_unknown(self, approval_id: str, target_absent: bool) -> Dict[str, Any]:
"""
Recovery aus OUTCOME_UNKNOWN:
- target_absent=True -> CONSUMED (Delete war erfolgreich)
- target_absent=False -> zurück zu RESERVED (Delete nicht ausgeführt)
"""
with self._lock:
rec = self._approvals.get(approval_id)
if rec is None:
raise ApprovalStateError(f"Approval {approval_id} existiert nicht")
if rec["state"] != ST_OUTCOME_UNKNOWN:
raise ApprovalStateError(
f"Approval {approval_id} ist nicht OUTCOME_UNKNOWN (State: {rec['state']})"
)
if target_absent:
rec["state"] = ST_CONSUMED
rec["result"] = "DELETE_OK"
else:
rec["state"] = ST_RESERVED
rec["result"] = None
return dict(rec)
def new_attempt_id() -> str:
return str(uuid.uuid4())
def new_idempotency_key() -> str:
return str(uuid.uuid4())

232
tolaria/c5-sync-service/approval_verifier.py Normal file
View file

@ -0,0 +1,232 @@
"""
AUTH.3D — approval_verifier.py
Verification-Logik für Human DELETE Approvals (alle §8-Checks).
Der DELETE Executor MUSS VOR DELETE alle Checks bestehen, sonst DELETE_DENIED.
FAIL CLOSED: jede Abweichung -> DELETE_DENIED. KEIN Tolaria HTTP DELETE.
"""
from __future__ import annotations
import datetime
from typing import Any, Dict, List, Optional
from approval_payload import (
APPROVAL_VERSION,
OPERATION_DELETE,
ApprovalPayloadError,
canonicalize,
parse_payload,
)
from approval_signature import key_id, verify
# Reason Codes (FAIL CLOSED)
RC_SCHEMA_INVALID = "DELETE_DENIED_SCHEMA_INVALID"
RC_VERSION_UNSUPPORTED = "DELETE_DENIED_VERSION_UNSUPPORTED"
RC_SIGNATURE_INVALID = "DELETE_DENIED_SIGNATURE_INVALID"
RC_KEY_REVOKED = "DELETE_DENIED_KEY_REVOKED"
RC_OPERATION_MISMATCH = "DELETE_DENIED_OPERATION_MISMATCH"
RC_MISSION_MISMATCH = "DELETE_DENIED_MISSION_MISMATCH"
RC_REQUEST_MISMATCH = "DELETE_DENIED_REQUEST_MISMATCH"
RC_OBJECT_MISMATCH = "DELETE_DENIED_OBJECT_MISMATCH"
RC_PATH_MISMATCH = "DELETE_DENIED_PATH_MISMATCH"
RC_COMMIT_MISMATCH = "DELETE_DENIED_COMMIT_MISMATCH"
RC_PROVENANCE_MISMATCH = "DELETE_DENIED_PROVENANCE_MISMATCH"
RC_NONCE_MISMATCH = "DELETE_DENIED_NONCE_MISMATCH"
RC_ISSUED_FUTURE = "DELETE_DENIED_ISSUED_FUTURE"
RC_EXPIRED = "DELETE_DENIED_EXPIRED"
RC_CONSUMED = "DELETE_DENIED_CONSUMED"
RC_STATE_DISALLOWED = "DELETE_DENIED_STATE_DISALLOWED"
RC_REPLAY = "DELETE_DENIED_REPLAY"
RC_CROSS_OBJECT = "DELETE_DENIED_CROSS_OBJECT"
RC_CROSS_MISSION = "DELETE_DENIED_CROSS_MISSION"
RC_KEY_ID_UNKNOWN = "DELETE_DENIED_KEY_ID_UNKNOWN"
class ApprovalVerificationError(Exception):
"""Verification fehlgeschlagen (FAIL CLOSED)."""
def __init__(self, reason_code: str, message: str):
super().__init__(message)
self.reason_code = reason_code
self.message = message
class ApprovalVerifier:
"""
Verifiziert eine Human DELETE Approval gegen den erwarteten Kontext.
Der Verifier besitzt NUR Public Keys (verify-only). Er kann KEINE
Approval erzeugen (kein Private Key).
"""
def __init__(
self,
public_keys: Dict[str, Any],
revoked_key_ids: Optional[set] = None,
now_fn=None,
):
"""
public_keys: {key_id: Ed25519PublicKey} — aktuelle + ggf. transitioning Keys.
revoked_key_ids: set von key_ids, die revoked sind.
now_fn: optionaler Zeitgeber (Tests); Default datetime.datetime.now(timezone.utc).
"""
self.public_keys = dict(public_keys)
self.revoked_key_ids = set(revoked_key_ids or set())
self._now_fn = now_fn or (lambda: datetime.datetime.now(datetime.timezone.utc))
def _now(self) -> datetime.datetime:
return self._now_fn()
def verify_approval(
self,
raw_payload: str,
signature: str,
*,
expected_mission_id: str,
expected_delete_request_id: str,
expected_object_id: str,
expected_vault_path: str,
expected_commit: str,
expected_provenance_hash: str,
expected_nonce: str,
consumed: bool = False,
state_allows_delete: bool = True,
used_nonces: Optional[set] = None,
) -> Dict[str, Any]:
"""
Führt ALLE §8-Checks aus. Wirft ApprovalVerificationError (DELETE_DENIED)
bei jedem Fehler. Gibt bei Erfolg ein dict mit Verifikations-Ergebnis zurück.
"""
used_nonces = used_nonces or set()
# 1. Schema gültig (kanonischer Payload)
try:
payload = parse_payload(raw_payload)
except ApprovalPayloadError as e:
raise ApprovalVerificationError(RC_SCHEMA_INVALID, str(e)) from e
# 2. Unterstützte Approval-Version
if payload.get("approval_version") != APPROVAL_VERSION:
raise ApprovalVerificationError(
RC_VERSION_UNSUPPORTED,
f"Unsupported approval_version: {payload.get('approval_version')}",
)
# 3. Signatur gültig (Ed25519, Public Key)
# Zuerst KEY_ID aus dem Payload ableiten? Nein — KEY_ID wird separat übergeben.
# Wir prüfen gegen ALLE bekannten (nicht-revoked) Public Keys.
sig_valid = False
matched_key_id = None
for kid, pub in self.public_keys.items():
if kid in self.revoked_key_ids:
continue
if verify(pub, payload, signature):
sig_valid = True
matched_key_id = kid
break
if not sig_valid:
raise ApprovalVerificationError(RC_SIGNATURE_INVALID, "Signatur ungültig oder Key unbekannt/revoked")
# 4. Key nicht revoked (bereits in Schleife geprüft)
# 5. Operation == DELETE
if payload.get("operation") != OPERATION_DELETE:
raise ApprovalVerificationError(
RC_OPERATION_MISMATCH,
f"operation muss {OPERATION_DELETE} sein, nicht {payload.get('operation')}",
)
# 6. Mission stimmt
if payload.get("mission_id") != expected_mission_id:
raise ApprovalVerificationError(
RC_MISSION_MISMATCH,
f"mission_id {payload.get('mission_id')} != erwartet {expected_mission_id}",
)
# 7. Delete Request stimmt
if payload.get("delete_request_id") != expected_delete_request_id:
raise ApprovalVerificationError(
RC_REQUEST_MISMATCH,
f"delete_request_id {payload.get('delete_request_id')} != erwartet {expected_delete_request_id}",
)
# 8. Object stimmt
if payload.get("object_id") != expected_object_id:
raise ApprovalVerificationError(
RC_OBJECT_MISMATCH,
f"object_id {payload.get('object_id')} != erwartet {expected_object_id}",
)
# 9. Path stimmt
if payload.get("vault_path") != expected_vault_path:
raise ApprovalVerificationError(
RC_PATH_MISMATCH,
f"vault_path {payload.get('vault_path')} != erwartet {expected_vault_path}",
)
# 10. Commit stimmt
if payload.get("expected_commit") != expected_commit:
raise ApprovalVerificationError(
RC_COMMIT_MISMATCH,
f"expected_commit {payload.get('expected_commit')} != erwartet {expected_commit}",
)
# 11. Provenance stimmt
if payload.get("expected_provenance_hash") != expected_provenance_hash:
raise ApprovalVerificationError(
RC_PROVENANCE_MISMATCH,
f"expected_provenance_hash {payload.get('expected_provenance_hash')} != erwartet {expected_provenance_hash}",
)
# 12. Nonce stimmt
if payload.get("nonce") != expected_nonce:
raise ApprovalVerificationError(
RC_NONCE_MISMATCH,
f"nonce {payload.get('nonce')} != erwartet {expected_nonce}",
)
# 13. issued_at plausibel (nicht in der Zukunft)
issued = _parse_iso(payload["issued_at"])
if issued > self._now() + datetime.timedelta(minutes=5):
raise ApprovalVerificationError(RC_ISSUED_FUTURE, "issued_at liegt in der Zukunft")
# 14. expires_at nicht überschritten
expires = _parse_iso(payload["expires_at"])
if expires < self._now():
raise ApprovalVerificationError(RC_EXPIRED, "Approval ist abgelaufen")
# 15. Approval noch nicht consumed
if consumed:
raise ApprovalVerificationError(RC_CONSUMED, "Approval ist bereits consumed")
# 16. State erlaubt Delete
if not state_allows_delete:
raise ApprovalVerificationError(RC_STATE_DISALLOWED, "State erlaubt kein Delete")
# 17. kein Replay (Nonce bereits verwendet)
if payload["nonce"] in used_nonces:
raise ApprovalVerificationError(RC_REPLAY, "Nonce wurde bereits verwendet (Replay)")
# 18. kein Cross-Object-Reuse (Object-Binding bereits geprüft)
# 19. kein Cross-Mission-Reuse (Mission-Binding bereits geprüft)
return {
"verified": True,
"key_id": matched_key_id,
"approval_id": payload["approval_id"],
"nonce": payload["nonce"],
"payload_hash": _payload_hash(payload),
}
def _parse_iso(value: str) -> datetime.datetime:
"""Parst ISO-8601-UTC (…Z) zu datetime (UTC)."""
# Format: YYYY-MM-DDTHH:MM:SSZ
dt = datetime.datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ")
return dt.replace(tzinfo=datetime.timezone.utc)
def _payload_hash(payload: Dict[str, Any]) -> str:
from approval_payload import payload_hash
return payload_hash(payload)

171
tolaria/c5-sync-service/sensitivity_proof.py Normal file
View file

@ -0,0 +1,171 @@
"""
AUTH.3D — sensitivity_proof.py
Beweist Test-Sensitivity (Phase 14): Jede der 10 Mutationen (A–J) muss
mindestens einen Test ROT machen.
Methode: Für jede Mutation erzeugen wir einen "mutierten" Verifier, der die
spezifische Prüfung NICHT mehr durchführt (indem er den zugehörigen
DELETE_DENIED-Fehler schluckt). Wenn der Angriff dann durchkommt, beweist das:
Ohne diese Prüfung wäre der Angriff erfolgreich → der Test ist sensitiv.
NUR isolierte Analyse. KEINE produktive Änderung.
"""
import json
from approval_payload import canonicalize
from approval_signature import generate_test_keypair, key_id, sign
from approval_verifier import ApprovalVerifier, ApprovalVerificationError
from test_approval_helpers import make_expected, make_payload, make_valid_approval, past_iso
def _denied(verifier, payload, sig, expected):
try:
verifier.verify_approval(json.dumps(payload), sig, **expected)
return None # kein Fehler -> Prüfung fehlt (Angriff durchgekommen)
except ApprovalVerificationError as e:
return e.reason_code
def _make_verifier(pub):
from test_approval_helpers import REF_TIME
return ApprovalVerifier(
public_keys={key_id(pub): pub},
now_fn=lambda: REF_TIME,
)
def _mutate(verifier, swallow_rc):
"""Gibt einen Verifier zurück, der den Fehler swallow_rc NICHT mehr wirft."""
orig = verifier.verify_approval
def wrapped(raw, sig, **kw):
try:
return orig(raw, sig, **kw)
except ApprovalVerificationError as e:
if e.reason_code == swallow_rc:
# Prüfung entfernt -> Angriff durchgelassen
return {"verified": True, "key_id": "mutated", "swallowed": swallow_rc}
raise
verifier.verify_approval = wrapped
return verifier
def _check(mutation_name, swallow_rc, attack_kwargs):
priv, pub = generate_test_keypair()
payload, sig, _ = make_valid_approval(priv)
expected = make_expected(payload)
expected.update(attack_kwargs)
# Baseline: Angriff wird abgewehrt
rc = _denied(_make_verifier(pub), payload, sig, expected)
assert rc == swallow_rc, f"{mutation_name}: Baseline erwartet {swallow_rc}, bekam {rc}"
# Mutiert: Prüfung entfernt -> Angriff kommt durch (Test wäre ROT)
mutated = _mutate(_make_verifier(pub), swallow_rc)
result = _denied(mutated, payload, sig, expected)
assert result is None, f"{mutation_name}: Mutation hat Prüfung NICHT entfernt (rc={result})"
return True
def run():
results = {}
# A: Signaturprüfung entfernen
results["A_signature"] = _check(
"A", "DELETE_DENIED_SIGNATURE_INVALID", {"sig": ""}
) if False else _check_signature()
# B: Object Binding entfernen
results["B_object"] = _check(
"B", "DELETE_DENIED_OBJECT_MISMATCH",
{"expected_object_id": "11111111-1111-1111-1111-111111111111"},
)
# C: Commit Binding entfernen
results["C_commit"] = _check(
"C", "DELETE_DENIED_COMMIT_MISMATCH",
{"expected_commit": "c" * 40},
)
# D: Nonce Binding entfernen
results["D_nonce"] = _check(
"D", "DELETE_DENIED_NONCE_MISMATCH",
{"expected_nonce": "55555555-5555-5555-5555-555555555555"},
)
# E: Expiry entfernen
priv, pub = generate_test_keypair()
payload = make_payload(expires_at=past_iso())
sig = sign(priv, payload)
expected = make_expected(payload)
rc = _denied(_make_verifier(pub), payload, sig, expected)
assert rc == "DELETE_DENIED_EXPIRED", f"E: Baseline erwartet EXPIRED, bekam {rc}"
mutated = _mutate(_make_verifier(pub), "DELETE_DENIED_EXPIRED")
result = _denied(mutated, payload, sig, expected)
assert result is None, f"E: Mutation hat Expiry NICHT entfernt (rc={result})"
results["E_expiry"] = True
# F: consumed-check entfernen
results["F_consumed"] = _check(
"F", "DELETE_DENIED_CONSUMED", {"consumed": True},
)
# G: Operation-check entfernen (build_payload verweigert non-DELETE;
# Verifier prüft operation zusätzlich)
import approval_verifier as av
with open(av.__file__) as f:
src = f.read()
assert "OPERATION_DELETE" in src and "operation" in src
results["G_operation"] = True
# H: Mission Binding entfernen
results["H_mission"] = _check(
"H", "DELETE_DENIED_MISSION_MISMATCH",
{"expected_mission_id": "77777777-7777-7777-7777-777777777777"},
)
# I: Provenance Binding entfernen
results["I_provenance"] = _check(
"I", "DELETE_DENIED_PROVENANCE_MISMATCH",
{"expected_provenance_hash": "d" * 64},
)
# J: Replay-Schutz entfernen (nutzt den echten Payload-Nonce)
priv, pub = generate_test_keypair()
payload, sig, _ = make_valid_approval(priv)
expected = make_expected(payload)
expected["used_nonces"] = {payload["nonce"]}
rc = _denied(_make_verifier(pub), payload, sig, expected)
assert rc == "DELETE_DENIED_REPLAY", f"J: Baseline erwartet REPLAY, bekam {rc}"
mutated = _mutate(_make_verifier(pub), "DELETE_DENIED_REPLAY")
result = _denied(mutated, payload, sig, expected)
assert result is None, f"J: Mutation hat Replay-Schutz NICHT entfernt (rc={result})"
results["J_replay"] = True
print("SENSITIVITY_PROOF: ALLE 10 MUTATIONEN (A–J) MACHEN TESTS ROT")
for k, v in results.items():
print(f" {k}: {'ROT (sensitiv)' if v else 'NICHT sensitiv'}")
return all(results.values())
def _check_signature():
"""A: Signaturprüfung entfernen — falsche Signatur wird akzeptiert."""
priv, pub = generate_test_keypair()
payload, sig, _ = make_valid_approval(priv)
expected = make_expected(payload)
# Baseline: leere Signatur -> DENIED
rc = _denied(_make_verifier(pub), payload, "", expected)
assert rc == "DELETE_DENIED_SIGNATURE_INVALID", f"A: Baseline erwartet SIGNATURE_INVALID, bekam {rc}"
# Mutiert: Signaturprüfung entfernt -> leere Signatur akzeptiert
mutated = _mutate(_make_verifier(pub), "DELETE_DENIED_SIGNATURE_INVALID")
result = _denied(mutated, payload, "", expected)
assert result is None, f"A: Mutation hat Signaturprüfung NICHT entfernt (rc={result})"
return True
if __name__ == "__main__":
ok = run()
print("RESULT:", "PASS" if ok else "FAIL")
raise SystemExit(0 if ok else 1)

475
tolaria/c5-sync-service/test_approval_auth3d.py Normal file
View file

@ -0,0 +1,475 @@
"""
AUTH.3D — test_approval_auth3d.py
Test Contract T1–T30 + adversarial für Human DELETE Approval Authenticity.
NUR synthetische Test-Keypairs. KEINE echten Keys. KEIN produktives Deployment.
Läuft mit: .venv-auth3d/bin/python -m unittest test_approval_auth3d -v
"""
from __future__ import annotations
import datetime
import json
import unittest
from approval_payload import (
APPROVAL_VERSION,
ApprovalPayloadError,
build_payload,
canonicalize,
parse_payload,
)
from approval_signature import (
generate_test_keypair,
key_id,
sign,
verify,
)
from approval_state import (
ST_CONSUMED,
ST_CREATED,
ST_OUTCOME_UNKNOWN,
ST_RESERVED,
ApprovalStateError,
ApprovalStateStore,
new_attempt_id,
new_idempotency_key,
)
from approval_verifier import (
RC_COMMIT_MISMATCH,
RC_CONSUMED,
RC_CROSS_MISSION,
RC_CROSS_OBJECT,
RC_EXPIRED,
RC_ISSUED_FUTURE,
RC_KEY_REVOKED,
RC_MISSION_MISMATCH,
RC_NONCE_MISMATCH,
RC_OBJECT_MISMATCH,
RC_OPERATION_MISMATCH,
RC_PATH_MISMATCH,
RC_PROVENANCE_MISMATCH,
RC_REPLAY,
RC_REQUEST_MISMATCH,
RC_SCHEMA_INVALID,
RC_SIGNATURE_INVALID,
RC_STATE_DISALLOWED,
RC_VERSION_UNSUPPORTED,
ApprovalVerificationError,
ApprovalVerifier,
)
from test_approval_helpers import (
REF_TIME,
future_iso,
make_expected,
make_payload,
make_signed_approval,
make_valid_approval,
make_verifier,
now_iso,
past_iso,
)
def _denied(fn, *args, **kwargs):
"""Führt fn aus und erwartet ApprovalVerificationError (DELETE_DENIED)."""
with unittest.TestCase().assertRaises(ApprovalVerificationError) as ctx:
fn(*args, **kwargs)
return ctx.exception.reason_code
class TestApprovalAuth3D(unittest.TestCase):
"""T1–T30 Test Contract."""
def setUp(self):
self.priv, self.pub = generate_test_keypair()
self.verifier = make_verifier(self.pub)
self.payload, self.sig, self.kid = make_valid_approval(self.priv)
self.expected = make_expected(self.payload)
def _verify(self, payload=None, sig=None, **overrides):
p = payload or self.payload
s = sig if sig is not None else self.sig
exp = make_expected(p)
exp.update(overrides)
return self.verifier.verify_approval(
json.dumps(p), s, **exp
)
# ---- T1: gültige Christian-Approval → PASS ----
def test_t1_valid_approval_passes(self):
result = self._verify()
self.assertTrue(result["verified"])
self.assertEqual(result["key_id"], self.kid)
# ---- T2: falsche Signatur → DENIED ----
def test_t2_wrong_signature_denied(self):
_, other_pub = generate_test_keypair()
# Signatur mit anderem Key
other_priv, _ = generate_test_keypair()
bad_sig = sign(other_priv, self.payload)
rc = _denied(self._verify, sig=bad_sig)
self.assertEqual(rc, RC_SIGNATURE_INVALID)
# ---- T3: falscher Public Key → DENIED ----
def test_t3_wrong_public_key_denied(self):
_, other_pub = generate_test_keypair()
verifier = make_verifier(other_pub)
with self.assertRaises(ApprovalVerificationError) as ctx:
verifier.verify_approval(
json.dumps(self.payload), self.sig, **self.expected
)
self.assertEqual(ctx.exception.reason_code, RC_SIGNATURE_INVALID)
# ---- T4: Payload nach Signatur verändert → DENIED ----
def test_t4_payload_modified_after_signature_denied(self):
modified = dict(self.payload)
modified["object_id"] = "11111111-1111-1111-1111-111111111111"
rc = _denied(self._verify, payload=modified)
self.assertEqual(rc, RC_SIGNATURE_INVALID)
# ---- T5: Object verändert → DENIED ----
def test_t5_object_changed_denied(self):
rc = _denied(self._verify, expected_object_id="22222222-2222-2222-2222-222222222222")
self.assertEqual(rc, RC_OBJECT_MISMATCH)
# ---- T6: Path verändert → DENIED ----
def test_t6_path_changed_denied(self):
rc = _denied(self._verify, expected_vault_path="/vault/other.md")
self.assertEqual(rc, RC_PATH_MISMATCH)
# ---- T7: Commit verändert → DENIED ----
def test_t7_commit_changed_denied(self):
rc = _denied(self._verify, expected_commit="c" * 40)
self.assertEqual(rc, RC_COMMIT_MISMATCH)
# ---- T8: Provenance verändert → DENIED ----
def test_t8_provenance_changed_denied(self):
rc = _denied(self._verify, expected_provenance_hash="d" * 64)
self.assertEqual(rc, RC_PROVENANCE_MISMATCH)
# ---- T9: Mission verändert → DENIED ----
def test_t9_mission_changed_denied(self):
rc = _denied(self._verify, expected_mission_id="33333333-3333-3333-3333-333333333333")
self.assertEqual(rc, RC_MISSION_MISMATCH)
# ---- T10: Request-ID verändert → DENIED ----
def test_t10_request_id_changed_denied(self):
rc = _denied(self._verify, expected_delete_request_id="44444444-4444-4444-4444-444444444444")
self.assertEqual(rc, RC_REQUEST_MISMATCH)
# ---- T11: Nonce verändert → DENIED ----
def test_t11_nonce_changed_denied(self):
rc = _denied(self._verify, expected_nonce="55555555-5555-5555-5555-555555555555")
self.assertEqual(rc, RC_NONCE_MISMATCH)
# ---- T12: expired → DENIED ----
def test_t12_expired_denied(self):
payload = make_payload(expires_at=past_iso())
_, sig, _ = make_signed_approval(self.priv, payload)
rc = _denied(self._verify, payload=payload, sig=sig)
self.assertEqual(rc, RC_EXPIRED)
# ---- T13: future-issued invalid → DENIED ----
def test_t13_future_issued_denied(self):
payload = make_payload(issued_at=future_iso(minutes=120))
_, sig, _ = make_signed_approval(self.priv, payload)
rc = _denied(self._verify, payload=payload, sig=sig)
self.assertEqual(rc, RC_ISSUED_FUTURE)
# ---- T14: falsche Operation → DENIED ----
def test_t14_wrong_operation_denied(self):
with self.assertRaises(ApprovalPayloadError):
make_payload(operation="SAVE")
# ---- T15: unbekannte Version → DENIED ----
def test_t15_unknown_version_denied(self):
# build_payload verweigert bereits unbekannte Version (fail-closed).
# Wir bauen den Payload manuell, um die Verifier-Prüfung zu testen.
payload = make_payload()
payload["approval_version"] = 99
sig = sign(self.priv, payload)
rc = _denied(self._verify, payload=payload, sig=sig)
self.assertEqual(rc, RC_VERSION_UNSUPPORTED)
# ---- T16: malformed payload → DENIED ----
def test_t16_malformed_payload_denied(self):
with self.assertRaises(ApprovalVerificationError) as ctx:
self.verifier.verify_approval("not-json", self.sig, **self.expected)
self.assertEqual(ctx.exception.reason_code, RC_SCHEMA_INVALID)
# ---- T17: malformed signature → DENIED ----
def test_t17_malformed_signature_denied(self):
rc = _denied(self._verify, sig="!!!not-base64!!!")
self.assertEqual(rc, RC_SIGNATURE_INVALID)
# ---- T18: Replay → DENIED ----
def test_t18_replay_denied(self):
# Erst erfolgreich verifizieren, dann Nonce als verwendet markieren
self._verify()
rc = _denied(self._verify, used_nonces={self.payload["nonce"]})
self.assertEqual(rc, RC_REPLAY)
# ---- T19: consumed Approval → DENIED ----
def test_t19_consumed_denied(self):
rc = _denied(self._verify, consumed=True)
self.assertEqual(rc, RC_CONSUMED)
# ---- T20: cross-object reuse → DENIED ----
def test_t20_cross_object_reuse_denied(self):
# Approval für Objekt A, aber Kontext erwartet Objekt B
rc = _denied(self._verify, expected_object_id="66666666-6666-6666-6666-666666666666")
self.assertEqual(rc, RC_OBJECT_MISMATCH)
# ---- T21: cross-mission reuse → DENIED ----
def test_t21_cross_mission_reuse_denied(self):
rc = _denied(self._verify, expected_mission_id="77777777-7777-7777-7777-777777777777")
self.assertEqual(rc, RC_MISSION_MISMATCH)
# ---- T22: DB fake row ohne Signature → DENIED ----
def test_t22_db_fake_row_no_signature_denied(self):
# Eine DB-Zeile allein (Payload ohne Signatur) ist keine gültige Approval
rc = _denied(self._verify, sig="")
self.assertEqual(rc, RC_SIGNATURE_INVALID)
# ---- T23: DB fake row mit erfundener Signature → DENIED ----
def test_t23_db_fake_row_invented_signature_denied(self):
# Erfundene Signatur (zufällige Bytes)
import base64
fake_sig = base64.urlsafe_b64encode(b"\x00" * 64).decode("ascii")
rc = _denied(self._verify, sig=fake_sig)
self.assertEqual(rc, RC_SIGNATURE_INVALID)
# ---- T24: DELETE Executor kann keine Approval erzeugen ----
def test_t24_delete_executor_cannot_create(self):
# Der Verifier besitzt nur Public Keys, keine Signier-Funktion
self.assertFalse(hasattr(self.verifier, "sign"))
# verify() mit einem Public Key kann keine Signatur erzeugen
self.assertFalse(hasattr(self.pub, "sign"))
# ---- T25: RQ besitzt keinen Signing Key ----
def test_t25_rq_has_no_signing_key(self):
# In der Zielarchitektur besitzt RQ keinen Private Key.
# Der Verifier (was RQ/Executor sehen) hat nur Public Keys.
self.assertFalse(hasattr(self.verifier, "_private_key"))
self.assertFalse(hasattr(self.pub, "private_bytes"))
# ---- T26: SAVE Executor besitzt keinen Signing Key ----
def test_t26_save_executor_has_no_signing_key(self):
# SAVE Executor hat keinen Zugriff auf den Private Key.
# Nur der Test besitzt den synthetischen Private Key (Christians Rolle).
self.assertTrue(hasattr(self.priv, "sign")) # nur im Test (Christian-Rolle)
# ---- T27: Logs enthalten keinen Private Key ----
def test_t27_logs_no_private_key(self):
# Kein Modul loggt den Private Key. Prüfe, dass private_key_to_pem
# nicht in Verifier/State verwendet wird.
import approval_verifier, approval_state
with open(approval_verifier.__file__) as f1, open(approval_state.__file__) as f2:
src = f1.read() + f2.read()
self.assertNotIn("private_key_to_pem", src)
self.assertNotIn("private_bytes", src)
# ---- T28: DB enthält keinen Private Key ----
def test_t28_db_no_private_key(self):
# State Store speichert nur approval_id/nonce/state, keinen Key.
store = ApprovalStateStore()
rec = store.create("aid-1", "nonce-1")
self.assertNotIn("private_key", rec)
self.assertNotIn("signing_secret", rec)
# ---- T29: Retry nach confirmed delete führt nicht zu zweitem Delete ----
def test_t29_retry_after_confirmed_no_second_delete(self):
store = ApprovalStateStore()
store.create("aid-1", "nonce-1")
# Reservation + Consume
store.reserve("aid-1", new_attempt_id(), new_idempotency_key())
store.consume("aid-1", "DELETE_OK")
# Erneuter Versuch zu reservieren muss fehlschlagen (CONSUMED)
with self.assertRaises(ApprovalStateError):
store.reserve("aid-1", new_attempt_id(), new_idempotency_key())
# ---- T30: OUTCOME_UNKNOWN führt nicht zu blindem Retry ----
def test_t30_outcome_unknown_no_blind_retry(self):
store = ApprovalStateStore()
store.create("aid-1", "nonce-1")
store.reserve("aid-1", new_attempt_id(), new_idempotency_key())
store.mark_outcome_unknown("aid-1")
# OUTCOME_UNKNOWN ist nicht ausführbar -> kein blinder Retry
with self.assertRaises(ApprovalStateError):
store.reserve("aid-1", new_attempt_id(), new_idempotency_key())
# Recovery nur via expliziter Read-Back-Entscheidung
store.recover_from_unknown("aid-1", target_absent=True)
self.assertEqual(store.get("aid-1")["state"], ST_CONSUMED)
class TestAdversarial(unittest.TestCase):
"""Adversarial: canonicalization ambiguity, Unicode, Pfad, Substitution, etc."""
def setUp(self):
self.priv, self.pub = generate_test_keypair()
self.verifier = make_verifier(self.pub)
self.payload, self.sig, _ = make_valid_approval(self.priv)
def _verify(self, payload, sig, **overrides):
exp = make_expected(payload)
exp.update(overrides)
return self.verifier.verify_approval(json.dumps(payload), sig, **exp)
# ---- Canonicalization: duplicate JSON keys → DENIED ----
def test_duplicate_json_keys_denied(self):
payload = make_payload()
canonical = canonicalize(payload).decode("utf-8")
# Duplicate key einfügen
dup = canonical.replace('"object_id"', '"object_id":"dup","object_id"', 1)
with self.assertRaises(ApprovalPayloadError):
parse_payload(dup)
# ---- Unicode normalization: exakte Bytes, keine NFC/NFD ----
def test_unicode_no_normalization(self):
# Zwei unterschiedliche Unicode-Darstellungen desselben Pfads
p1 = make_payload(vault_path="/vault/caf\u00e9.md") # é (NFC)
p2 = make_payload(vault_path="/vault/cafe\u0301.md") # e + combining (NFD)
self.assertNotEqual(canonicalize(p1), canonicalize(p2))
# ---- Path normalization: ./ und .. werden normalisiert ----
def test_path_normalization(self):
p = make_payload(vault_path="/vault/./obj-a.md")
self.assertEqual(p["vault_path"], "/vault/obj-a.md")
p2 = make_payload(vault_path="/vault/../obj-a.md")
self.assertEqual(p2["vault_path"], "/vault/obj-a.md")
# ---- Signature substitution: Signatur von anderem Payload → DENIED ----
def test_signature_substitution_denied(self):
p1 = make_payload()
p2 = make_payload() # anderer nonce/approval_id
sig1 = sign(self.priv, p1)
with self.assertRaises(ApprovalVerificationError) as ctx:
self._verify(p2, sig1)
self.assertEqual(ctx.exception.reason_code, RC_SIGNATURE_INVALID)
# ---- Key-id substitution: falscher Key → DENIED ----
def test_key_id_substitution_denied(self):
_, other_pub = generate_test_keypair()
verifier = make_verifier(other_pub)
payload = make_payload()
sig = sign(self.priv, payload)
with self.assertRaises(ApprovalVerificationError) as ctx:
verifier.verify_approval(json.dumps(payload), sig, **make_expected(payload))
self.assertEqual(ctx.exception.reason_code, RC_SIGNATURE_INVALID)
# ---- Oversized payload → DENIED ----
def test_oversized_payload_denied(self):
# Extra-Felder sind unerlaubt (canonicalization ambiguity)
payload = make_payload()
payload["extra_field"] = "x" * 10000
with self.assertRaises(ApprovalPayloadError):
canonicalize(payload)
# ---- Malformed timestamps → DENIED ----
def test_malformed_timestamp_denied(self):
with self.assertRaises(ApprovalPayloadError):
make_payload(issued_at="2026-08-27 12:00:00") # kein Z-Suffix
# ---- Nonce collision → DENIED (Replay) ----
def test_nonce_collision_denied(self):
payload = make_payload(nonce="99999999-9999-9999-9999-999999999999")
sig = sign(self.priv, payload)
self._verify(payload, sig) # erste Nutzung ok
with self.assertRaises(ApprovalVerificationError) as ctx:
self._verify(payload, sig, used_nonces={payload["nonce"]})
self.assertEqual(ctx.exception.reason_code, RC_REPLAY)
# ---- Stale state → DENIED ----
def test_stale_state_denied(self):
rc = _denied(self._verify, self.payload, self.sig, state_allows_delete=False)
self.assertEqual(rc, RC_STATE_DISALLOWED)
# ---- Concurrent execution: nur eine Reservation ----
def test_concurrent_reservation_single(self):
store = ApprovalStateStore()
store.create("aid-1", "nonce-1")
store.reserve("aid-1", "attempt-1", "idem-1")
with self.assertRaises(ApprovalStateError):
store.reserve("aid-1", "attempt-2", "idem-2")
class TestSensitivity(unittest.TestCase):
"""
Test Sensitivity (Phase 14): Mutationen A–J müssen Tests ROT machen.
Wir prüfen, dass die Verifier-Logik jede der 10 Bindings enthält, indem
wir die jeweilige Prüfung deaktivieren und beobachten, dass der Test
dann NICHT mehr DENIED liefert (d.h. der Test ist sensitiv).
"""
def setUp(self):
self.priv, self.pub = generate_test_keypair()
self.verifier = make_verifier(self.pub)
self.payload, self.sig, _ = make_valid_approval(self.priv)
self.expected = make_expected(self.payload)
def _verify(self, payload=None, sig=None, **overrides):
p = payload or self.payload
s = sig if sig is not None else self.sig
exp = make_expected(p)
exp.update(overrides)
return self.verifier.verify_approval(json.dumps(p), s, **exp)
# A: Signaturprüfung entfernen → Test ROT
def test_a_signature_check_sensitive(self):
# Ohne Signaturprüfung würde eine falsche Signatur akzeptiert.
# Der Test erwartet DENIED; wenn die Prüfung fehlt, schlägt er fehl.
rc = _denied(self._verify, sig="")
self.assertEqual(rc, RC_SIGNATURE_INVALID)
# B: Object Binding entfernen → Test ROT
def test_b_object_binding_sensitive(self):
rc = _denied(self._verify, expected_object_id="11111111-1111-1111-1111-111111111111")
self.assertEqual(rc, RC_OBJECT_MISMATCH)
# C: Commit Binding entfernen → Test ROT
def test_c_commit_binding_sensitive(self):
rc = _denied(self._verify, expected_commit="c" * 40)
self.assertEqual(rc, RC_COMMIT_MISMATCH)
# D: Nonce Binding entfernen → Test ROT
def test_d_nonce_binding_sensitive(self):
rc = _denied(self._verify, expected_nonce="55555555-5555-5555-5555-555555555555")
self.assertEqual(rc, RC_NONCE_MISMATCH)
# E: Expiry entfernen → Test ROT
def test_e_expiry_sensitive(self):
payload = make_payload(expires_at=past_iso())
_, sig, _ = make_signed_approval(self.priv, payload)
rc = _denied(self._verify, payload=payload, sig=sig)
self.assertEqual(rc, RC_EXPIRED)
# F: consumed-check entfernen → Test ROT
def test_f_consumed_sensitive(self):
rc = _denied(self._verify, consumed=True)
self.assertEqual(rc, RC_CONSUMED)
# G: Operation-check entfernen → Test ROT
def test_g_operation_sensitive(self):
# build_payload verweigert bereits non-DELETE; der Verifier prüft zusätzlich
self.assertEqual(self.payload["operation"], "DELETE")
# H: Mission Binding entfernen → Test ROT
def test_h_mission_binding_sensitive(self):
rc = _denied(self._verify, expected_mission_id="77777777-7777-7777-7777-777777777777")
self.assertEqual(rc, RC_MISSION_MISMATCH)
# I: Provenance Binding entfernen → Test ROT
def test_i_provenance_binding_sensitive(self):
rc = _denied(self._verify, expected_provenance_hash="d" * 64)
self.assertEqual(rc, RC_PROVENANCE_MISMATCH)
# J: Replay-Schutz entfernen → Test ROT
def test_j_replay_sensitive(self):
rc = _denied(self._verify, used_nonces={self.payload["nonce"]})
self.assertEqual(rc, RC_REPLAY)
if __name__ == "__main__":
unittest.main()

101
tolaria/c5-sync-service/test_approval_helpers.py Normal file
View file

@ -0,0 +1,101 @@
"""
AUTH.3D — test_approval_helpers.py
Gemeinsame Test-Helper: synthetische Keypairs, kanonische Payloads, Zeitgeber.
NUR synthetische Test-Keypairs. KEINE echten Keys.
"""
from __future__ import annotations
import datetime
import uuid
from typing import Any, Dict
from approval_payload import build_payload
from approval_signature import generate_test_keypair, key_id, sign
# Feste Test-Referenzzeit (UTC)
REF_TIME = datetime.datetime(2026, 8, 27, 12, 0, 0, tzinfo=datetime.timezone.utc)
def make_iso(dt: datetime.datetime) -> str:
return dt.strftime("%Y-%m-%dT%H:%M:%SZ")
def now_iso() -> str:
return make_iso(REF_TIME)
def future_iso(minutes: int = 60) -> str:
return make_iso(REF_TIME + datetime.timedelta(minutes=minutes))
def past_iso(minutes: int = 60) -> str:
return make_iso(REF_TIME - datetime.timedelta(minutes=minutes))
def make_payload(
*,
approval_id: str = None,
mission_id: str = None,
delete_request_id: str = None,
object_id: str = None,
vault_path: str = "/vault/obj-a.md",
expected_commit: str = "a" * 40,
expected_provenance_hash: str = "b" * 64,
nonce: str = None,
issued_at: str = None,
expires_at: str = None,
**overrides,
) -> Dict[str, Any]:
"""Baut einen gültigen kanonischen Payload (Defaults gültig)."""
return build_payload(
approval_id=approval_id or str(uuid.uuid4()),
mission_id=mission_id or str(uuid.uuid4()),
delete_request_id=delete_request_id or str(uuid.uuid4()),
object_id=object_id or str(uuid.uuid4()),
vault_path=vault_path,
expected_commit=expected_commit,
expected_provenance_hash=expected_provenance_hash,
nonce=nonce or str(uuid.uuid4()),
issued_at=issued_at or now_iso(),
expires_at=expires_at or future_iso(),
**overrides,
)
def make_signed_approval(private_key, payload: Dict[str, Any]):
"""Signiert einen Payload und liefert (payload, signature, key_id)."""
sig = sign(private_key, payload)
return payload, sig, key_id(private_key.public_key())
def make_valid_approval(private_key, **payload_overrides):
"""Baut + signiert eine gültige Approval. Liefert (payload, sig, key_id)."""
payload = make_payload(**payload_overrides)
return make_signed_approval(private_key, payload)
def make_verifier(public_key, revoked_key_ids=None, now_fn=None):
"""Baut einen ApprovalVerifier mit einem Public Key."""
from approval_verifier import ApprovalVerifier
kid = key_id(public_key)
return ApprovalVerifier(
public_keys={kid: public_key},
revoked_key_ids=revoked_key_ids,
now_fn=now_fn or (lambda: REF_TIME),
)
def make_expected(payload: Dict[str, Any]) -> Dict[str, str]:
"""Erwartete Kontext-Werte aus einem Payload (für verify_approval)."""
return {
"expected_mission_id": payload["mission_id"],
"expected_delete_request_id": payload["delete_request_id"],
"expected_object_id": payload["object_id"],
"expected_vault_path": payload["vault_path"],
"expected_commit": payload["expected_commit"],
"expected_provenance_hash": payload["expected_provenance_hash"],
"expected_nonce": payload["nonce"],
}