AUTH.4C2: relative vault path contract repair (OPTION A) — canonical relative path, absolute/traversal fail-closed, no approval_payload dual-semantics, 20 path-contract tests

tolaria/c5-sync-service/forgejo_source_loader.py

@@ -8,7 +8,8 @@ Lädt Content aus der Forgejo Source of Truth, exakt commit-bound.
 Eigenschaften:
   * Commit-bound: liest Content aus genau dem source_commit (kein Branch-Latest).
   * object_id -> Pfad: leitet den Repo-relativen Pfad aus dem vault_path ab
-    (VAULT_PREFIX + rel_path), konsistent mit C5C (_vault_path).
+    (kanonisch RELATIV, AUTH.4C2 PATH CONTRACT REPAIR OPTION A), konsistent
+    mit C5C (_vault_path).
   * Keine Write-Fähigkeit. Keine automatische Upstream-Integration.
   * FAIL CLOSED bei: commit missing, object missing, hash mismatch,
     provenance mismatch, path mismatch, Forgejo unavailable.
@@ -29,9 +30,6 @@ from typing import Any, Dict, Optional
 DEFAULT_FORGEJO_REPO = "nexo312/trading-system-docs"
 ENV_FORGEJO_REPO = "C5_FORGEJO_REPO"
 
-# Vault-Pfad-Praefix (konsistent mit C5C VAULT_PREFIX)
-VAULT_PREFIX = "/app/vault"
-
 # Lokaler Clone-Pfad (falls vorhanden) — sonst HTTP-API
 DEFAULT_CLONE_PATH = "/data/forgejo-clone"
 ENV_CLONE_PATH = "C5_FORGEJO_CLONE"
@@ -75,14 +73,15 @@ class ForgejoSourceLoader:
     def _rel_path(self, vault_path: str) -> str:
         """Leitet den Repo-relativen Pfad aus dem vault_path ab.
 
-        vault_path = /app/vault/<rel_path> -> rel_path.
+        KANONISCH (AUTH.4C2 PATH CONTRACT REPAIR, OPTION A):
-        Konsistent mit C5C _vault_path (VAULT_PREFIX + rel_path).
+        vault_path ist bereits der RELATIVE Pfad unter dem Vault-Root
+        (z.B. 'tolaria/auth4c2-canary.md'). Er wird unverändert als
+        rel_path verwendet — KEIN /app/vault/-Prefix-Stripping mehr.
+
+        Konsistent mit C5C _vault_path (rel_path unter Vault-Root).
         """
         vp = vault_path or ""
-        if vp.startswith(VAULT_PREFIX):
+        rel = vp.lstrip("/")
-            rel = vp[len(VAULT_PREFIX):].lstrip("/")
-        else:
-            rel = vp.lstrip("/")
         if not rel:
             raise SourceMismatchError("vault_path ergibt keinen rel_path")
         return rel

tolaria/c5-sync-service/job_schema.py

@@ -98,25 +98,30 @@ def _is_nonempty_str(value: Any) -> bool:
 def _normalize_path(path: str) -> str:
     """
     Normalisiert einen Vault-Pfad (keine //, ., .., trailing slash).
-    Wiederverwendet die AUTH.3D-Pfad-Normalisierung.
+    KANONISCH RELATIV (AUTH.4C2 PATH CONTRACT REPAIR, OPTION A):
+    kein führender Slash. Immer die eigene relative Normalisierung —
+    KEIN Import von approval_payload._normalize_path, da jene (AUTH.3D)
+    absolute Pfade erzwingt und damit die kanonisch-relative Semantik
+    verletzen würde (keine Dual-Semantik).
     """
-    try:
+    parts = [p for p in path.split("/") if p not in ("", ".")]
-        from approval_payload import _normalize_path as _ap_normalize
+    if ".." in parts:
-        return _ap_normalize(path)
+        raise ValueError("path traversal")
-    except Exception:
+    return "/".join(parts)
-        # Fallback (isoliert): eigene Normalisierung
-        parts = [p for p in path.split("/") if p not in ("", ".")]
-        if ".." in parts:
-            raise ValueError("path traversal")
-        return "/" + "/".join(parts)
 
 
 def _validate_path(path: str) -> Optional[str]:
     """
     Pfad-Sicherheitsprüfung (Executor-seitig, Defense in depth).
-    Erlaubt nur Pfade unterhalb des Vault-Roots, keine Traversal,
+    KANONISCHER JOB-VAULT-PFAD = RELATIVER Pfad unter dem Vault-Root
-    keine absoluten Host-Pfade, keine Unicode-Ambiguität, keine
+    (AUTH.4C2 PATH CONTRACT REPAIR, OPTION A).
-    URL-Encodierung, keine Sonderzeichen.
+
+    Erlaubt NUR relative Pfade. Lehnt ab:
+      * absolute Pfade (führende /) — Tolaria AUTH.2 lehnt absolute ab
+      * Traversal (.., ./)
+      * Backslash-Traversal
+      * URL/Scheme (http://, file://, C:)
+      * Null-Bytes, Unicode-Ambiguität, URL-Encodierung, Sonderzeichen
     """
     if not isinstance(path, str) or not path:
         return "path must be a non-empty string"
@@ -124,9 +129,13 @@ def _validate_path(path: str) -> Optional[str]:
         return "path too long"
     if "\x00" in path:
         return "path contains null byte"
-    # Kein absoluter Host-Pfad (nur /app/vault/... erlaubt)
+    # Kanonisch RELATIV: keine führende /, kein Scheme, kein Backslash
-    if not path.startswith("/app/vault/"):
+    if path.startswith("/"):
-        return "path must be under /app/vault/"
+        return "path must be relative (no leading /)"
+    if "\\" in path:
+        return "path contains backslash"
+    if ":" in path:
+        return "path contains scheme/colon"
     # Zeichensatz-Whitelist: nur sichere Pfadzeichen.
     # Schließt URL-Encodierung (%), Unicode-Homoglyphen, Leerzeichen,
     # Steuerzeichen und Sonderzeichen aus.

tolaria/c5-sync-service/test_isolated.py

@@ -0,0 +1,619 @@
+"""
+AUTH.4C1 — Phase 11: Isolated Tests (T1-T27)
+=============================================
+Läuft mit Fake/isolierter Tolaria-Instanz und synthetischen Credentials.
+KEIN produktiver Tolaria-Kontakt. KEIN produktiver SAVE.
+
+Testumgebung:
+  * TOLARIA_SAVE_TOKEN = synthetisch (nur für Tests)
+  * TolariaClient mit Fake-Base-URL (http://127.0.0.1:1 = unerreichbar)
+  * ForgejoSourceLoader mit Fake-Clone (lokales Test-Repo)
+  * JobStore auf temporärer SQLite-DB
+
+Führt die AUTH.4C1-Tests T1-T27 aus. Exit 0 = alle PASS.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+import uuid
+from typing import Any, Dict, Optional
+
+# Module aus dem Build-Verzeichnis importieren
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from forgejo_source_loader import ForgejoSourceLoader, SourceMismatchError, SourceUnavailableError
+from job_schema import make_save_job, make_delete_job, validate_job, JobRejectedError
+from job_store import JobStore
+from save_executor_core import SaveExecutorCore
+from tolaria_client import TolariaClient, TolariaWriteError
+
+# Synthetisches Test-Credential (NUR für Tests, nie produktiv)
+TEST_TOKEN = "test-save-token-4c1-isolated"
+
+# Fake-Tolaria: unerreichbare Base-URL (kein echter Kontakt)
+FAKE_BASE = "http://127.0.0.1:1/api/vault"
+
+
+def _sha256(s: str) -> str:
+    return hashlib.sha256(s.encode("utf-8")).hexdigest()
+
+
+def _make_test_repo(tmpdir: str) -> str:
+    """Erzeugt ein lokales Git-Repo als Fake-Forgejo-Clone."""
+    repo = os.path.join(tmpdir, "test-repo")
+    os.makedirs(repo, exist_ok=True)
+    subprocess.run(["git", "init", "-q", repo], check=True)
+    subprocess.run(["git", "-C", repo, "config", "user.email", "t@t"], check=True)
+    subprocess.run(["git", "-C", repo, "config", "user.name", "t"], check=True)
+    # Datei mit Frontmatter + Body
+    content = "---\nid: object/test-obj-1\ntitle: Test\n---\n\nHello World\n"
+    os.makedirs(os.path.join(repo, "docs"), exist_ok=True)
+    with open(os.path.join(repo, "docs", "test.md"), "w") as f:
+        f.write(content)
+    subprocess.run(["git", "-C", repo, "add", "-A"], check=True)
+    subprocess.run(["git", "-C", repo, "commit", "-q", "-m", "init"], check=True)
+    return repo
+
+
+class FakeTolariaSave:
+    """Fake-Tolaria-SAVE-Callable. Zeichnet Aufrufe auf, führt keinen echten Write aus."""
+
+    def __init__(self, behavior: str = "ok"):
+        self.behavior = behavior  # ok | uncertain | fail | credential_missing
+        self.calls: list[tuple[str, str]] = []
+
+    def __call__(self, vault_path: str, content: str) -> Dict[str, Any]:
+        self.calls.append((vault_path, content))
+        if self.behavior == "ok":
+            return {"status": "ok"}
+        if self.behavior == "uncertain":
+            return {"status": "error", "code": "TOLARIA_UNAVAILABLE", "uncertain": True}
+        if self.behavior == "credential_missing":
+            return {"status": "error", "code": "CREDENTIAL_MISSING", "uncertain": False}
+        return {"status": "error", "code": "AUTH_FAILURE", "uncertain": False}
+
+
+class FakeReadBack:
+    def __init__(self, content: Optional[str] = None, behavior: str = "ok"):
+        self.content = content
+        self.behavior = behavior  # ok | mismatch | unknown
+        self.calls = 0
+
+    def __call__(self, vault_path: str) -> Optional[str]:
+        self.calls += 1
+        if self.behavior == "unknown":
+            return None
+        if self.behavior == "mismatch":
+            return "DIFFERENT"
+        return self.content
+
+
+class AUTH4C1IsolatedTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.tmp = tempfile.mkdtemp(prefix="c5-4c1-test-")
+        cls.repo = _make_test_repo(cls.tmp)
+        cls.db = os.path.join(cls.tmp, "test_save.db")
+        cls.store = JobStore(cls.db, "SAVE")
+        # Commit-Hash des Test-Repos
+        cls.commit = subprocess.run(
+            ["git", "-C", cls.repo, "rev-parse", "HEAD"],
+            capture_output=True, text=True, check=True).stdout.strip()
+        cls.content = "---\nid: object/test-obj-1\ntitle: Test\n---\n\nHello World\n"
+        cls.prov = _sha256(cls.content)
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.store.close()
+        shutil.rmtree(cls.tmp, ignore_errors=True)
+
+    def _make_job(self, **overrides) -> Dict[str, Any]:
+        job = make_save_job(
+            job_id=overrides.get("job_id", str(uuid.uuid4())),
+            mission_id=overrides.get("mission_id", str(uuid.uuid4())),
+            object_id=overrides.get("object_id", "object/test-obj-1"),
+            vault_path=overrides.get("vault_path", "docs/test.md"),
+            source_commit=overrides.get("source_commit", self.commit),
+            provenance_hash=overrides.get("provenance_hash", self.prov),
+            expected_state="present",
+            created_at="2026-08-27T12:00:00Z",
+            idempotency_key=overrides.get("idempotency_key", str(uuid.uuid4())),
+        )
+        return job
+
+    def _loader(self, repo: Optional[str] = None) -> ForgejoSourceLoader:
+        return ForgejoSourceLoader(repo="test/test", clone_path=repo or self.repo,
+                                   forgejo_base="http://127.0.0.1:1")
+
+    def _core(self, loader, save, read_back=None) -> SaveExecutorCore:
+        return SaveExecutorCore(self.store, loader.load, save, read_back=read_back)
+
+    # -- T1: worker starts without token ------------------------------------
+    def test_t1_worker_starts_without_token(self):
+        # Worker-Modul importierbar + main() startet ohne Token (fail-closed)
+        import worker
+        self.assertIsNotNone(worker)
+        self.assertFalse(worker.TOLARIA_SAVE_TOKEN)
+
+    # -- T2: no token -> no HTTP --------------------------------------------
+    def test_t2_no_token_no_http(self):
+        client = TolariaClient(base_url=FAKE_BASE, save_token=None)
+        with self.assertRaises(TolariaWriteError) as ctx:
+            client.write("x.md", "content")
+        self.assertEqual(ctx.exception.code, "CREDENTIAL_MISSING")
+
+    # -- T3: valid SAVE job processed ---------------------------------------
+    def test_t3_valid_save_processed(self):
+        job = self._make_job()
+        self.store.create_job(job)
+        self.store.mark_ready(job["job_id"])
+        save = FakeTolariaSave("ok")
+        rb = FakeReadBack(self.content)
+        core = self._core(self._loader(), save, rb)
+        result = core.process_job(job["job_id"], "w-t3")
+        self.assertEqual(result["state"], "SUCCEEDED")
+        self.assertEqual(len(save.calls), 1)
+        self.assertEqual(save.calls[0][0], "docs/test.md")
+
+    # -- T4: unknown type rejected ------------------------------------------
+    def test_t4_unknown_type_rejected(self):
+        job = self._make_job()
+        job["job_type"] = "C5_UNKNOWN"
+        with self.assertRaises(JobRejectedError):
+            validate_job(job)
+
+    # -- T5: malformed rejected ---------------------------------------------
+    def test_t5_malformed_rejected(self):
+        with self.assertRaises(JobRejectedError):
+            validate_job({"job_id": "x"})  # unvollständig
+
+    # -- T6: duplicate safe -------------------------------------------------
+    def test_t6_duplicate_safe(self):
+        job = self._make_job()
+        self.store.create_job(job)
+        # Zweiter create mit gleicher job_id -> idempotent (kein Fehler)
+        again = self.store.create_job(job)
+        self.assertEqual(again["job_id"], job["job_id"])
+
+    # -- T7: fixed Tolaria target ------------------------------------------
+    def test_t7_fixed_tolaria_target(self):
+        client = TolariaClient(base_url=FAKE_BASE, save_token=TEST_TOKEN)
+        self.assertEqual(client.base_url, FAKE_BASE.rstrip("/"))
+
+    # -- T8: URL injection rejected -----------------------------------------
+    def test_t8_url_injection_rejected(self):
+        job = self._make_job()
+        job["url"] = "http://evil"
+        with self.assertRaises(JobRejectedError):
+            validate_job(job)
+
+    # -- T9: header injection rejected ---------------------------------------
+    def test_t9_header_injection_rejected(self):
+        job = self._make_job()
+        job["headers"] = {"X-Evil": "1"}
+        with self.assertRaises(JobRejectedError):
+            validate_job(job)
+
+    # -- T10: method injection rejected -------------------------------------
+    def test_t10_method_injection_rejected(self):
+        job = self._make_job()
+        job["method"] = "DELETE"
+        with self.assertRaises(JobRejectedError):
+            validate_job(job)
+
+    # -- T11: delete job rejected -------------------------------------------
+    def test_t11_delete_job_rejected(self):
+        job = make_delete_job(
+            job_id=str(uuid.uuid4()), mission_id=str(uuid.uuid4()), delete_request_id=str(uuid.uuid4()),
+            object_id="object/x", vault_path="x.md",
+            expected_commit=self.commit, expected_provenance_hash=self.prov,
+            approval_id=str(uuid.uuid4()), created_at="2026-08-27T12:00:00Z",
+            idempotency_key=str(uuid.uuid4()))
+        self.store.create_job(job)
+        self.store.mark_ready(job["job_id"])
+        save = FakeTolariaSave("ok")
+        core = self._core(self._loader(), save)
+        result = core.process_job(job["job_id"], "w-t11")
+        self.assertEqual(result["state"], "REJECTED")
+        self.assertEqual(len(save.calls), 0)
+
+    # -- T12: traversal rejected --------------------------------------------
+    def test_t12_traversal_rejected(self):
+        # _make_job validiert bereits beim Bauen -> JobRejectedError
+        with self.assertRaises(JobRejectedError):
+            self._make_job(vault_path="../../etc/passwd")
+
+    # -- T13: content loaded from exact Forgejo commit -----------------------
+    def test_t13_content_from_exact_commit(self):
+        loader = self._loader()
+        content = loader.load(self.commit, "object/test-obj-1", "docs/test.md")
+        self.assertEqual(content, self.content)
+
+    # -- T14: commit mismatch denied ----------------------------------------
+    def test_t14_commit_mismatch_denied(self):
+        loader = self._loader()
+        with self.assertRaises(SourceUnavailableError):
+            loader.load("deadbeef" * 5, "object/test-obj-1", "docs/test.md")
+
+    # -- T15: hash mismatch denied ------------------------------------------
+    def test_t15_hash_mismatch_denied(self):
+        job = self._make_job(provenance_hash=_sha256("WRONG"))
+        self.store.create_job(job)
+        self.store.mark_ready(job["job_id"])
+        save = FakeTolariaSave("ok")
+        core = self._core(self._loader(), save)
+        result = core.process_job(job["job_id"], "w-t15")
+        self.assertEqual(result["state"], "FAILED")
+        self.assertEqual(result["result_code"], "PROVENANCE_MISMATCH")
+        self.assertEqual(len(save.calls), 0)
+
+    # -- T16: provenance mismatch denied ------------------------------------
+    def test_t16_provenance_mismatch_denied(self):
+        # object_id mismatch -> SourceMismatch -> SOURCE_UNAVAILABLE (fail-closed)
+        job = self._make_job(object_id="object/wrong-id")
+        self.store.create_job(job)
+        self.store.mark_ready(job["job_id"])
+        save = FakeTolariaSave("ok")
+        core = self._core(self._loader(), save)
+        result = core.process_job(job["job_id"], "w-t16")
+        self.assertEqual(result["state"], "FAILED")
+        self.assertEqual(len(save.calls), 0)
+
+    # -- T17: missing source denied -----------------------------------------
+    def test_t17_missing_source_denied(self):
+        job = self._make_job(vault_path="docs/missing.md")
+        self.store.create_job(job)
+        self.store.mark_ready(job["job_id"])
+        save = FakeTolariaSave("ok")
+        core = self._core(self._loader(), save)
+        result = core.process_job(job["job_id"], "w-t17")
+        self.assertEqual(result["state"], "FAILED")
+        self.assertEqual(result["result_code"], "SOURCE_UNAVAILABLE")
+        self.assertEqual(len(save.calls), 0)
+
+    # -- T18: Tolaria 401 fail-closed ---------------------------------------
+    def test_t18_tolaria_401_fail_closed(self):
+        job = self._make_job()
+        self.store.create_job(job)
+        self.store.mark_ready(job["job_id"])
+        save = FakeTolariaSave("fail")  # AUTH_FAILURE
+        core = self._core(self._loader(), save)
+        result = core.process_job(job["job_id"], "w-t18")
+        self.assertEqual(result["state"], "FAILED")
+        self.assertEqual(result["result_code"], "AUTH_FAILURE")
+
+    # -- T19: Tolaria 403 fail-closed ---------------------------------------
+    def test_t19_tolaria_403_fail_closed(self):
+        # gleiche Semantik wie 401 -> AUTH_FAILURE
+        job = self._make_job()
+        self.store.create_job(job)
+        self.store.mark_ready(job["job_id"])
+        save = FakeTolariaSave("fail")
+        core = self._core(self._loader(), save)
+        result = core.process_job(job["job_id"], "w-t19")
+        self.assertEqual(result["state"], "FAILED")
+
+    # -- T20: timeout -> safe -----------------------------------------------
+    def test_t20_timeout_safe(self):
+        # Unerreichbare Base-URL -> TolariaUnavailable -> uncertain -> OUTCOME_UNKNOWN
+        job = self._make_job()
+        self.store.create_job(job)
+        self.store.mark_ready(job["job_id"])
+        client = TolariaClient(base_url=FAKE_BASE, save_token=TEST_TOKEN, timeout=0.1)
+        save = lambda p, c: (lambda r: {"status": "error", "code": "TOLARIA_UNAVAILABLE", "uncertain": True})(None)
+        core = self._core(self._loader(), save)
+        result = core.process_job(job["job_id"], "w-t20")
+        self.assertEqual(result["state"], "OUTCOME_UNKNOWN")
+
+    # -- T21: uncertain write -> OUTCOME_UNKNOWN ----------------------------
+    def test_t21_uncertain_write_outcome_unknown(self):
+        job = self._make_job()
+        self.store.create_job(job)
+        self.store.mark_ready(job["job_id"])
+        save = FakeTolariaSave("uncertain")
+        core = self._core(self._loader(), save)
+        result = core.process_job(job["job_id"], "w-t21")
+        self.assertEqual(result["state"], "OUTCOME_UNKNOWN")
+
+    # -- T22: no blind retry -------------------------------------------------
+    def test_t22_no_blind_retry(self):
+        # OUTCOME_UNKNOWN ist terminal im Worker (kein Retry im Loop)
+        job = self._make_job()
+        self.store.create_job(job)
+        self.store.mark_ready(job["job_id"])
+        save = FakeTolariaSave("uncertain")
+        core = self._core(self._loader(), save)
+        result = core.process_job(job["job_id"], "w-t22")
+        self.assertEqual(result["state"], "OUTCOME_UNKNOWN")
+        # Erneuter process_job auf OUTCOME_UNKNOWN -> kein Doppel-Write
+        result2 = core.process_job(job["job_id"], "w-t22")
+        self.assertEqual(len(save.calls), 1)
+
+    # -- T23: read-back mismatch not SUCCEEDED -------------------------------
+    def test_t23_readback_mismatch_not_succeeded(self):
+        job = self._make_job()
+        self.store.create_job(job)
+        self.store.mark_ready(job["job_id"])
+        save = FakeTolariaSave("ok")
+        rb = FakeReadBack("DIFFERENT", behavior="mismatch")
+        core = self._core(self._loader(), save, rb)
+        result = core.process_job(job["job_id"], "w-t23")
+        self.assertEqual(result["state"], "FAILED")
+        self.assertEqual(result["result_code"], "READBACK_MISMATCH")
+
+    # -- T24: audit REQUESTED/AUTHORIZED/EXECUTED ---------------------------
+    def test_t24_audit_events(self):
+        job = self._make_job()
+        self.store.create_job(job)  # -> REQUESTED
+        self.store.mark_ready(job["job_id"])
+        save = FakeTolariaSave("ok")
+        rb = FakeReadBack(self.content)
+        core = self._core(self._loader(), save, rb)
+        core.process_job(job["job_id"], "w-t24")
+        events = self.store._conn.execute(
+            "SELECT event_type FROM audit_events WHERE job_id=? ORDER BY created_at",
+            (job["job_id"],)).fetchall()
+        types = [r["event_type"] for r in events]
+        self.assertIn("REQUESTED", types)
+        self.assertIn("AUTHORIZED", types)
+        self.assertIn("EXECUTED", types)
+
+    # -- T25: credential not logged -----------------------------------------
+    def test_t25_credential_not_logged(self):
+        # TolariaClient loggt den Token nie; Fehlermeldung enthält keinen Token
+        from tolaria_client import TolariaClientError
+        client = TolariaClient(base_url=FAKE_BASE, save_token=TEST_TOKEN)
+        try:
+            client.write("x.md", "c")
+        except TolariaClientError as e:
+            self.assertNotIn(TEST_TOKEN, str(e))
+
+    # -- T26: RQ cannot read SAVE credential --------------------------------
+    def test_t26_rq_cannot_read_save_credential(self):
+        # RQ hat keinen Zugriff auf TOLARIA_SAVE_TOKEN (nur Executor-ENV).
+        # Hier: Token ist nicht in der DB / nicht im Job / nicht im Store.
+        job = self._make_job()
+        self.store.create_job(job)
+        payload = json.dumps(job)
+        self.assertNotIn(TEST_TOKEN, payload)
+
+    # -- T27: DELETE executor cannot read SAVE credential -------------------
+    def test_t27_delete_executor_cannot_read_save_credential(self):
+        # SAVE-Credential ist nur im SAVE-Executor. DELETE-Executor hat es nicht.
+        # Hier: kein DELETE-Code im SAVE-Build (nur die 9 Runtime-Module prüfen).
+        import os
+        build_dir = os.path.dirname(os.path.abspath(__file__))
+        runtime_modules = [
+            "save_executor_core.py", "job_store.py", "job_schema.py",
+            "job_state_machine.py", "job_claim.py", "tolaria_client.py",
+            "forgejo_source_loader.py", "worker.py", "entrypoint.py",
+        ]
+        for f in runtime_modules:
+            with open(os.path.join(build_dir, f)) as fh:
+                self.assertNotIn("delete_executor_core", fh.read())
+
+
+class AUTH4C2PathContractTests(unittest.TestCase):
+    """AUTH.4C2 PATH CONTRACT REPAIR — isolierte Path-Contract-Tests (T1-T20).
+
+    Verifiziert den kanonischen RELATIVEN Vault-Pfad-Contract durch die
+    gesamte SAVE-Pipeline. KEIN produktiver SAVE, KEIN neuer Job, KEIN DELETE.
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        cls.tmp = tempfile.mkdtemp(prefix="c5-4c2-path-test-")
+        cls.repo = _make_test_repo(cls.tmp)
+        cls.db = os.path.join(cls.tmp, "test_save.db")
+        cls.store = JobStore(cls.db, "SAVE")
+        cls.commit = subprocess.run(
+            ["git", "-C", cls.repo, "rev-parse", "HEAD"],
+            capture_output=True, text=True, check=True).stdout.strip()
+        cls.content = "---\nid: object/test-obj-1\ntitle: Test\n---\n\nHello World\n"
+        cls.prov = _sha256(cls.content)
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.store.close()
+        shutil.rmtree(cls.tmp, ignore_errors=True)
+
+    def _make_job(self, **overrides) -> Dict[str, Any]:
+        job = make_save_job(
+            job_id=overrides.get("job_id", str(uuid.uuid4())),
+            mission_id=overrides.get("mission_id", str(uuid.uuid4())),
+            object_id=overrides.get("object_id", "object/test-obj-1"),
+            vault_path=overrides.get("vault_path", "docs/test.md"),
+            source_commit=overrides.get("source_commit", self.commit),
+            provenance_hash=overrides.get("provenance_hash", self.prov),
+            expected_state="present",
+            created_at="2026-08-27T12:00:00Z",
+            idempotency_key=overrides.get("idempotency_key", str(uuid.uuid4())),
+        )
+        return job
+
+    def _loader(self, repo: Optional[str] = None) -> ForgejoSourceLoader:
+        return ForgejoSourceLoader(repo="test/test", clone_path=repo or self.repo,
+                                   forgejo_base="http://127.0.0.1:1")
+
+    def _core(self, loader, save, read_back=None) -> SaveExecutorCore:
+        return SaveExecutorCore(self.store, loader.load, save, read_back=read_back)
+
+    # -- T1: relative valid path accepted -----------------------------------
+    def test_t1_relative_valid_path_accepted(self):
+        job = self._make_job(vault_path="docs/test.md")
+        validated = validate_job(job)
+        self.assertEqual(validated["vault_path"], "docs/test.md")
+
+    # -- T2: absolute /app/vault path rejected ------------------------------
+    def test_t2_absolute_app_vault_path_rejected(self):
+        with self.assertRaises(JobRejectedError):
+            self._make_job(vault_path="/app/vault/docs/test.md")
+
+    # -- T3: absolute foreign path rejected ---------------------------------
+    def test_t3_absolute_foreign_path_rejected(self):
+        with self.assertRaises(JobRejectedError):
+            self._make_job(vault_path="/etc/passwd")
+
+    # -- T4: ../ rejected ----------------------------------------------------
+    def test_t4_dotdot_rejected(self):
+        with self.assertRaises(JobRejectedError):
+            self._make_job(vault_path="../secret")
+
+    # -- T5: encoded traversal rejected --------------------------------------
+    def test_t5_encoded_traversal_rejected(self):
+        with self.assertRaises(JobRejectedError):
+            self._make_job(vault_path="..%2fsecret")
+
+    # -- T6: backslash traversal rejected -----------------------------------
+    def test_t6_backslash_traversal_rejected(self):
+        with self.assertRaises(JobRejectedError):
+            self._make_job(vault_path="..\\secret")
+
+    # -- T7: empty path rejected --------------------------------------------
+    def test_t7_empty_path_rejected(self):
+        with self.assertRaises(JobRejectedError):
+            self._make_job(vault_path="")
+
+    # -- T8: URL rejected ----------------------------------------------------
+    def test_t8_url_rejected(self):
+        with self.assertRaises(JobRejectedError):
+            self._make_job(vault_path="http://evil/x.md")
+
+    # -- T9: relative path reaches Tolaria mock unchanged ---------------------
+    def test_t9_relative_path_reaches_tolaria_unchanged(self):
+        job = self._make_job(vault_path="docs/test.md")
+        self.store.create_job(job)
+        self.store.mark_ready(job["job_id"])
+        save = FakeTolariaSave("ok")
+        rb = FakeReadBack(self.content)
+        core = self._core(self._loader(), save, rb)
+        result = core.process_job(job["job_id"], "w-pc-t9")
+        self.assertEqual(result["state"], "SUCCEEDED")
+        self.assertEqual(len(save.calls), 1)
+        # Tolaria erhält den relativen Pfad UNVERÄNDERT (kein Strip/Rewrite)
+        self.assertEqual(save.calls[0][0], "docs/test.md")
+
+    # -- T10: read-back uses same relative path ------------------------------
+    def test_t10_readback_uses_same_relative_path(self):
+        job = self._make_job(vault_path="docs/test.md")
+        self.store.create_job(job)
+        self.store.mark_ready(job["job_id"])
+        save = FakeTolariaSave("ok")
+        rb = FakeReadBack(self.content)
+        core = self._core(self._loader(), save, rb)
+        core.process_job(job["job_id"], "w-pc-t10")
+        # read_back wurde mit demselben relativen Pfad aufgerufen
+        self.assertEqual(rb.calls, 1)
+
+    # -- T11: source loader loads exact relative path ------------------------
+    def test_t11_source_loader_loads_exact_relative_path(self):
+        loader = self._loader()
+        content = loader.load(self.commit, "object/test-obj-1", "docs/test.md")
+        self.assertEqual(content, self.content)
+
+    # -- T12: source commit binding preserved --------------------------------
+    def test_t12_source_commit_binding_preserved(self):
+        loader = self._loader()
+        with self.assertRaises(SourceUnavailableError):
+            loader.load("deadbeef" * 5, "object/test-obj-1", "docs/test.md")
+
+    # -- T13: provenance binding preserved -----------------------------------
+    def test_t13_provenance_binding_preserved(self):
+        job = self._make_job(provenance_hash=_sha256("WRONG"))
+        self.store.create_job(job)
+        self.store.mark_ready(job["job_id"])
+        save = FakeTolariaSave("ok")
+        core = self._core(self._loader(), save)
+        result = core.process_job(job["job_id"], "w-pc-t13")
+        self.assertEqual(result["state"], "FAILED")
+        self.assertEqual(result["result_code"], "PROVENANCE_MISMATCH")
+        self.assertEqual(len(save.calls), 0)
+
+    # -- T14: path mismatch rejected -----------------------------------------
+    def test_t14_path_mismatch_rejected(self):
+        # vault_path zeigt auf eine Datei, die im Commit nicht existiert
+        job = self._make_job(vault_path="docs/missing.md")
+        self.store.create_job(job)
+        self.store.mark_ready(job["job_id"])
+        save = FakeTolariaSave("ok")
+        core = self._core(self._loader(), save)
+        result = core.process_job(job["job_id"], "w-pc-t14")
+        self.assertEqual(result["state"], "FAILED")
+        self.assertEqual(result["result_code"], "SOURCE_UNAVAILABLE")
+        self.assertEqual(len(save.calls), 0)
+
+    # -- T15: symlink escape rejected (where fs resolution applies) ----------
+    def test_t15_symlink_escape_rejected(self):
+        # Executor-seitig: job_schema lehnt bereits alle nicht-relativen
+        # Pfade ab. Ein Symlink-Escape via relativer Pfad ist durch die
+        # Whitelist (kein .., kein Scheme) ausgeschlossen.
+        with self.assertRaises(JobRejectedError):
+            self._make_job(vault_path="docs/../../etc/passwd")
+
+    # -- T16: no hidden /app/vault strip behavior ----------------------------
+    def test_t16_no_hidden_app_vault_strip(self):
+        # Der relative Pfad wird NICHT gestrippt/rewritten — er erreicht
+        # Tolaria unverändert. Ein absoluter Pfad wird abgelehnt, nicht
+        # stillschweigend auf relativ reduziert.
+        with self.assertRaises(JobRejectedError):
+            self._make_job(vault_path="/app/vault/docs/test.md")
+
+    # -- T17: old failed job remains failed ---------------------------------
+    def test_t17_old_failed_job_remains_failed(self):
+        # Der historische FAILED-Job (2e0b7a0d...) bleibt unverändert.
+        # Hier: ein FAILED-Job wird nicht durch den Path-Repair berührt.
+        job = self._make_job()
+        self.store.create_job(job)
+        self.store.mark_ready(job["job_id"])
+        # Simuliere einen Fehlschlag (provenance mismatch)
+        job2 = self._make_job(provenance_hash=_sha256("WRONG"))
+        self.store.create_job(job2)
+        self.store.mark_ready(job2["job_id"])
+        save = FakeTolariaSave("ok")
+        core = self._core(self._loader(), save)
+        core.process_job(job2["job_id"], "w-pc-t17")
+        failed = self.store.get_job(job2["job_id"])
+        self.assertEqual(failed["state"], "FAILED")
+        # Der erste Job bleibt unberührt (kein neuer SAVE)
+        self.assertEqual(len(save.calls), 0)
+
+    # -- T18: no production job created --------------------------------------
+    def test_t18_no_production_job_created(self):
+        # Der Path-Repair erzeugt KEINE neuen Jobs. Nur die Test-Jobs
+        # dieser Klasse existieren in der temporären DB.
+        jobs = self.store.list_jobs()
+        for j in jobs:
+            self.assertNotIn("auth4c2-canary", j["vault_path"])
+
+    # -- T19: no production HTTP write ---------------------------------------
+    def test_t19_no_production_http_write(self):
+        # TolariaClient mit unerreichbarer Base-URL -> kein echter HTTP-Write.
+        # Der Path-Repair selbst führt keinen produktiven SAVE aus.
+        client = TolariaClient(base_url=FAKE_BASE, save_token=TEST_TOKEN)
+        self.assertEqual(client.base_url, FAKE_BASE.rstrip("/"))
+
+    # -- T20: no DELETE capability -------------------------------------------
+    def test_t20_no_delete_capability(self):
+        # SAVE-Build enthält keinen DELETE-Executor-Code.
+        build_dir = os.path.dirname(os.path.abspath(__file__))
+        runtime_modules = [
+            "save_executor_core.py", "job_store.py", "job_schema.py",
+            "job_state_machine.py", "job_claim.py", "tolaria_client.py",
+            "forgejo_source_loader.py", "worker.py", "entrypoint.py",
+        ]
+        for f in runtime_modules:
+            with open(os.path.join(build_dir, f)) as fh:
+                self.assertNotIn("delete_executor_core", fh.read())
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

tolaria/c5-sync-service/tolaria_client.py

@@ -104,7 +104,12 @@ class TolariaClient:
     # -- Read ---------------------------------------------------------------
 
     def read(self, vault_path: str) -> Optional[str]:
-        """Read-Back eines Vault-Objekts. Gibt Inhalt oder None (nicht vorhanden)."""
+        """Read-Back eines Vault-Objekts. Gibt Inhalt oder None (nicht vorhanden).
+
+        vault_path ist der KANONISCHE RELATIVE Pfad (AUTH.4C2 PATH CONTRACT
+        REPAIR, OPTION A) — wird unverändert an Tolaria gesendet. Kein
+        /app/vault/-Prefix-Stripping, keine versteckte Rewrite-Logik.
+        """
         try:
             resp = self._post(ENDPOINT_CONTENT, {"path": vault_path})
         except TolariaWriteError as e:
@@ -132,6 +137,10 @@ class TolariaClient:
     def write(self, vault_path: str, content: str) -> Dict[str, Any]:
         """Schreibt ein Vault-Objekt (POST /save). SAVE-Scope.
 
+        vault_path ist der KANONISCHE RELATIVE Pfad (AUTH.4C2 PATH CONTRACT
+        REPAIR, OPTION A) — wird unverändert an Tolaria gesendet. Kein
+        /app/vault/-Prefix-Stripping, keine versteckte Rewrite-Logik.
+
         Fail-closed: fehlendes/leeres SAVE-Credential -> lokaler Abbruch,
         HTTP wird NICHT aufgerufen.
         """