trading-system-docs/tolaria/c5-sync-service/test_job_channel_adversarial.py

"""
AUTH.3E — test_job_channel_adversarial.py
==========================================
Adversarial Tests (Mission §22) + Test Sensitivity (Mission §23).

Adversarial:
  * JSON duplicate keys
  * Unicode path ambiguity
  * encoded traversal
  * job_type casing tricks
  * enum confusion
  * integer/string confusion
  * oversized payload
  * stale mission
  * stale provenance
  * race claim
  * replay after restart
  * DB row manipulation
  * cross-executor DB injection
  * approval substitution
  * object substitution after claim
  * path substitution after claim
  * state mutation after approval
  * crash before HTTP
  * crash after uncertain HTTP

Sensitivity (Mutationen A–L): Jede Mutation muss relevante Tests ROT machen.
"""

from __future__ import annotations

import hashlib
import json
import os
import tempfile
import unittest
import uuid

from job_schema import (
    JOB_TYPE_DELETE,
    JOB_TYPE_SAVE,
    JobRejectedError,
    make_delete_job,
    make_save_job,
    validate_job,
)
from job_state_machine import (
    ST_CLAIMED,
    ST_EXECUTING,
    ST_FAILED,
    ST_OUTCOME_UNKNOWN,
    ST_READY,
    ST_REJECTED,
    ST_SUCCEEDED,
    InvalidTransitionError,
    transition,
)
from job_store import JobStore
from save_executor_core import SaveExecutorCore
from delete_executor_core import DeleteExecutorCore


def _uuid() -> str:
    return str(uuid.uuid4())


def _sha256(content: str) -> str:
    return hashlib.sha256(content.encode("utf-8")).hexdigest()


def _now_iso() -> str:
    return "2026-08-27T00:00:00Z"


def _make_save_job(**overrides) -> dict:
    job = make_save_job(
        job_id=_uuid(), mission_id=_uuid(), object_id="obj-001",
        vault_path="/app/vault/notes/obj-001.md", source_commit=_sha256("commit"),
        provenance_hash=_sha256("content"), expected_state="ready",
        created_at=_now_iso(), idempotency_key=_uuid(),
    )
    job.update(overrides)
    return job


def _make_delete_job(**overrides) -> dict:
    job = make_delete_job(
        job_id=_uuid(), mission_id=_uuid(), delete_request_id=_uuid(),
        object_id="obj-001", vault_path="/app/vault/notes/obj-001.md",
        expected_commit=_sha256("commit"), expected_provenance_hash=_sha256("content"),
        approval_id=_uuid(), created_at=_now_iso(), idempotency_key=_uuid(),
    )
    job.update(overrides)
    return job


def _valid_approval(job: dict) -> dict:
    return {
        "approval_id": job["approval_id"], "mission_id": job["mission_id"],
        "object_id": job["object_id"], "vault_path": job["vault_path"],
        "expected_commit": job["expected_commit"],
        "expected_provenance_hash": job["expected_provenance_hash"],
        "nonce": _uuid(), "expired": False, "consumed": False,
        "signature_valid": True,
    }


class AdversarialSchemaTests(unittest.TestCase):
    """Adversarial: Schema-Angriffe."""

    def test_json_duplicate_keys(self):
        # JSON mit doppelten Keys -> parse_payload (AUTH.3D) lehnt ab.
        # Hier: defensiv — validate_job auf einem dict mit doppeltem Key.
        job = _make_save_job()
        # Simuliere Duplicate-Key durch dict mit zwei gleichen Keys (letzter gewinnt)
        # -> kein Fehler, aber wir prüfen, dass kein privilegiertes Feld eingeschleust wird
        job["command"] = "evil"
        with self.assertRaises(JobRejectedError):
            validate_job(job)

    def test_unicode_path_ambiguity(self):
        # Unicode-Homoglyph / nicht-NFC-normalisierter Pfad
        job = _make_save_job(vault_path="/app/vault/notes/obj\u0301.md")  # e + combining accent
        with self.assertRaises(JobRejectedError):
            validate_job(job)

    def test_encoded_traversal(self):
        # URL-encoded traversal
        job = _make_save_job(vault_path="/app/vault/%2e%2e/etc/passwd")
        with self.assertRaises(JobRejectedError):
            validate_job(job)

    def test_job_type_casing_tricks(self):
        job = _make_save_job(job_type="c5_save_object")
        with self.assertRaises(JobRejectedError):
            validate_job(job)

    def test_enum_confusion(self):
        # job_type als Nicht-String (z.B. int)
        job = _make_save_job(job_type=123)
        with self.assertRaises(JobRejectedError):
            validate_job(job)

    def test_integer_string_confusion(self):
        # job_version als String statt int
        job = _make_save_job(job_version="1")
        with self.assertRaises(JobRejectedError):
            validate_job(job)

    def test_oversized_payload(self):
        # Riesiger object_id
        job = _make_save_job(object_id="x" * 100000)
        with self.assertRaises(JobRejectedError):
            validate_job(job)

    def test_stale_mission(self):
        # mission_id ungültig (kein UUID)
        job = _make_save_job(mission_id="stale-mission")
        with self.assertRaises(JobRejectedError):
            validate_job(job)

    def test_stale_provenance(self):
        # provenance_hash ungültig (kein sha256)
        job = _make_save_job(provenance_hash="not-a-hash")
        with self.assertRaises(JobRejectedError):
            validate_job(job)


class AdversarialRuntimeTests(unittest.TestCase):
    """Adversarial: Runtime-Angriffe."""

    def setUp(self):
        self.tmpdir = tempfile.mkdtemp()
        self.save_db = os.path.join(self.tmpdir, "c5a_save.db")
        self.delete_db = os.path.join(self.tmpdir, "c5a_delete.db")
        self.http_calls = []
        self.approvals = {}
        self.content = "hello world"
        self.provenance = _sha256(self.content)

    def _source_loader(self, commit, object_id):
        return self.content

    def _tolaria_save(self, vault_path, content):
        self.http_calls.append(("SAVE", vault_path))
        return {"status": "ok"}

    def _approval_loader(self, approval_id):
        return self.approvals.get(approval_id)

    def _approval_verify(self, approval):
        return {"valid": approval.get("signature_valid", False), "reason": "ok"}

    def _tolaria_delete(self, vault_path):
        self.http_calls.append(("DELETE", vault_path))
        return {"status": "ok"}

    def test_race_claim(self):
        # Zwei Worker claimen denselben Job -> nur einer gewinnt
        store = JobStore(self.save_db, "SAVE")
        job = _make_save_job(provenance_hash=self.provenance)
        store.create_job(job)
        store.mark_ready(job["job_id"])
        from job_claim import JobAlreadyClaimedError
        store.claim_job(job["job_id"], "worker-1")
        with self.assertRaises(JobAlreadyClaimedError):
            store.claim_job(job["job_id"], "worker-2")
        store.close()

    def test_replay_after_restart(self):
        # Nach Restart (neuer Store auf gleicher DB) kein Doppel-Delete
        job = _make_delete_job()
        self.approvals[job["approval_id"]] = _valid_approval(job)
        store1 = JobStore(self.delete_db, "DELETE")
        store1.create_job(job)
        store1.mark_ready(job["job_id"])
        exec1 = DeleteExecutorCore(
            store1, self._approval_loader, self._approval_verify, self._tolaria_delete
        )
        exec1.process_job(job["job_id"], "delete-worker")
        store1.close()
        # Restart
        store2 = JobStore(self.delete_db, "DELETE")
        exec2 = DeleteExecutorCore(
            store2, self._approval_loader, self._approval_verify, self._tolaria_delete
        )
        exec2.process_job(job["job_id"], "delete-worker")
        self.assertEqual(len(self.http_calls), 1)  # kein 2. Delete
        store2.close()

    def test_db_row_manipulation(self):
        # Manipulierter Payload (job_type geändert) -> Executor lehnt ab
        store = JobStore(self.save_db, "SAVE")
        job = _make_save_job(provenance_hash=self.provenance)
        store.create_job(job)
        store.mark_ready(job["job_id"])
        # Manipulation: Payload job_type auf DELETE ändern (Angriffsfläche)
        row = store._conn.execute(
            "SELECT payload FROM jobs WHERE job_id = ?", (job["job_id"],)
        ).fetchone()
        payload = json.loads(row[0])
        payload["job_type"] = JOB_TYPE_DELETE
        store._conn.execute(
            "UPDATE jobs SET payload = ? WHERE job_id = ?",
            (json.dumps(payload), job["job_id"]),
        )
        store._conn.commit()
        exec_core = SaveExecutorCore(store, self._source_loader, self._tolaria_save)
        result = exec_core.process_job(job["job_id"], "save-worker")
        self.assertEqual(result["state"], ST_REJECTED)
        self.assertEqual(self.http_calls, [])
        store.close()

    def test_cross_executor_db_injection(self):
        # DELETE-Job in SAVE-DB injiziert -> SAVE-Executor lehnt ab
        store = JobStore(self.save_db, "SAVE")
        job = _make_delete_job()
        store.create_job(job)
        store.mark_ready(job["job_id"])
        exec_core = SaveExecutorCore(store, self._source_loader, self._tolaria_save)
        result = exec_core.process_job(job["job_id"], "save-worker")
        self.assertEqual(result["state"], ST_REJECTED)
        self.assertEqual(self.http_calls, [])
        store.close()

    def test_approval_substitution(self):
        # Approval für anderes Objekt substituiert -> denied
        job = _make_delete_job()
        approval = _valid_approval(job)
        approval["object_id"] = "obj-OTHER"
        self.approvals[job["approval_id"]] = approval
        store = JobStore(self.delete_db, "DELETE")
        store.create_job(job)
        store.mark_ready(job["job_id"])
        exec_core = DeleteExecutorCore(
            store, self._approval_loader, self._approval_verify, self._tolaria_delete
        )
        result = exec_core.process_job(job["job_id"], "delete-worker")
        self.assertEqual(result["result_code"], "OBJECT_MISMATCH")
        self.assertEqual(self.http_calls, [])
        store.close()

    def test_object_substitution_after_claim(self):
        # object_id im Payload manipuliert -> Executor validiert gegen Approval
        # (TOCTOU-Defense: Job wird nach dem Claim neu gelesen)
        job = _make_delete_job()
        self.approvals[job["approval_id"]] = _valid_approval(job)
        store = JobStore(self.delete_db, "DELETE")
        store.create_job(job)
        store.mark_ready(job["job_id"])
        # Payload manipulieren (object_id ändern)
        row = store._conn.execute(
            "SELECT payload FROM jobs WHERE job_id = ?", (job["job_id"],)
        ).fetchone()
        payload = json.loads(row[0])
        payload["object_id"] = "obj-OTHER"
        store._conn.execute(
            "UPDATE jobs SET payload = ? WHERE job_id = ?",
            (json.dumps(payload), job["job_id"]),
        )
        store._conn.commit()
        exec_core = DeleteExecutorCore(
            store, self._approval_loader, self._approval_verify, self._tolaria_delete
        )
        result = exec_core.process_job(job["job_id"], "delete-worker")
        self.assertEqual(result["result_code"], "OBJECT_MISMATCH")
        self.assertEqual(self.http_calls, [])
        store.close()

    def test_path_substitution_after_claim(self):
        job = _make_delete_job()
        self.approvals[job["approval_id"]] = _valid_approval(job)
        store = JobStore(self.delete_db, "DELETE")
        store.create_job(job)
        store.mark_ready(job["job_id"])
        row = store._conn.execute(
            "SELECT payload FROM jobs WHERE job_id = ?", (job["job_id"],)
        ).fetchone()
        payload = json.loads(row[0])
        payload["vault_path"] = "/app/vault/notes/OTHER.md"
        store._conn.execute(
            "UPDATE jobs SET payload = ? WHERE job_id = ?",
            (json.dumps(payload), job["job_id"]),
        )
        store._conn.commit()
        exec_core = DeleteExecutorCore(
            store, self._approval_loader, self._approval_verify, self._tolaria_delete
        )
        result = exec_core.process_job(job["job_id"], "delete-worker")
        self.assertEqual(result["result_code"], "PATH_MISMATCH")
        self.assertEqual(self.http_calls, [])
        store.close()

    def test_state_mutation_after_approval(self):
        # expected_commit im Payload mutiert -> Executor validiert gegen Approval
        job = _make_delete_job()
        self.approvals[job["approval_id"]] = _valid_approval(job)
        store = JobStore(self.delete_db, "DELETE")
        store.create_job(job)
        store.mark_ready(job["job_id"])
        row = store._conn.execute(
            "SELECT payload FROM jobs WHERE job_id = ?", (job["job_id"],)
        ).fetchone()
        payload = json.loads(row[0])
        payload["expected_commit"] = _sha256("other-commit")
        store._conn.execute(
            "UPDATE jobs SET payload = ? WHERE job_id = ?",
            (json.dumps(payload), job["job_id"]),
        )
        store._conn.commit()
        exec_core = DeleteExecutorCore(
            store, self._approval_loader, self._approval_verify, self._tolaria_delete
        )
        result = exec_core.process_job(job["job_id"], "delete-worker")
        self.assertEqual(result["result_code"], "COMMIT_MISMATCH")
        self.assertEqual(self.http_calls, [])
        store.close()

    def test_crash_before_http(self):
        # Crash vor HTTP -> Job bleibt CLAIMED, Lease läuft ab, SAVE wird recovered
        store = JobStore(self.save_db, "SAVE")
        job = _make_save_job(provenance_hash=self.provenance)
        store.create_job(job)
        store.mark_ready(job["job_id"])
        store.claim_job(job["job_id"], "worker-1", lease_seconds=0)
        # Crash simuliert: kein HTTP, kein weiterer Schritt
        self.assertEqual(self.http_calls, [])
        # Recovery: SAVE -> READY
        store.recover_stale_claims("worker-2")
        self.assertEqual(store.get_job(job["job_id"])["state"], ST_READY)
        store.close()

    def test_crash_after_uncertain_http(self):
        # Crash nach unklarem HTTP -> OUTCOME_UNKNOWN, kein blinder Retry
        def uncertain_delete(vault_path):
            self.http_calls.append(("DELETE", vault_path))
            return {"status": "error", "code": "TIMEOUT", "uncertain": True}
        job = _make_delete_job()
        self.approvals[job["approval_id"]] = _valid_approval(job)
        store = JobStore(self.delete_db, "DELETE")
        store.create_job(job)
        store.mark_ready(job["job_id"])
        exec_core = DeleteExecutorCore(
            store, self._approval_loader, self._approval_verify, uncertain_delete
        )
        result = exec_core.process_job(job["job_id"], "delete-worker")
        self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN)
        # Kein blinder Retry
        exec_core.process_job(job["job_id"], "delete-worker")
        self.assertEqual(len(self.http_calls), 1)
        store.close()


class SensitivityTests(unittest.TestCase):
    """
    Test Sensitivity (Mission §23): Mutationen A–L müssen relevante Tests ROT machen.

    Diese Tests prüfen die INVARIANTEN direkt. Wenn eine Mutation die Invariante
    entfernt, schlägt der Test fehl (ROT). Die Mutationen selbst werden in
    sensitivity_proof.py simuliert (statische Analyse).
    """

    def test_a_job_type_allowlist_removed(self):
        # Invariante: job_type muss in geschlossener Allowlist sein
        job = _make_save_job(job_type="C5_EXECUTE")
        with self.assertRaises(JobRejectedError):
            validate_job(job)

    def test_b_save_delete_worker_scope_removed(self):
        # Invariante: SAVE-Executor claimt nur SAVE-Jobs
        store = JobStore(self.save_db, "SAVE")
        job = _make_delete_job()
        store.create_job(job)
        store.mark_ready(job["job_id"])
        exec_core = SaveExecutorCore(
            store, self._source_loader, self._tolaria_save
        )
        result = exec_core.process_job(job["job_id"], "save-worker")
        self.assertEqual(result["state"], ST_REJECTED)
        store.close()

    def test_c_atomic_claim_removed(self):
        # Invariante: Claim ist atomar (nur ein Worker gewinnt)
        store = JobStore(self.save_db, "SAVE")
        job = _make_save_job(provenance_hash=self.provenance)
        store.create_job(job)
        store.mark_ready(job["job_id"])
        from job_claim import JobAlreadyClaimedError
        store.claim_job(job["job_id"], "worker-1")
        with self.assertRaises(JobAlreadyClaimedError):
            store.claim_job(job["job_id"], "worker-2")
        store.close()

    def test_d_idempotency_removed(self):
        # Invariante: duplicate job_id idempotent
        store = JobStore(self.save_db, "SAVE")
        job = _make_save_job()
        store.create_job(job)
        store.create_job(job)
        self.assertEqual(len(store.list_jobs()), 1)
        store.close()

    def test_e_approval_requirement_removed(self):
        # Invariante: DELETE ohne Approval denied
        job = _make_delete_job()
        store = JobStore(self.delete_db, "DELETE")
        store.create_job(job)
        store.mark_ready(job["job_id"])
        exec_core = DeleteExecutorCore(
            store, self._approval_loader, self._approval_verify, self._tolaria_delete
        )
        result = exec_core.process_job(job["job_id"], "delete-worker")
        self.assertEqual(result["result_code"], "APPROVAL_MISSING")
        store.close()

    def test_f_signature_verification_removed(self):
        # Invariante: invalid Signature denied
        job = _make_delete_job()
        approval = _valid_approval(job)
        approval["signature_valid"] = False
        self.approvals[job["approval_id"]] = approval
        store = JobStore(self.delete_db, "DELETE")
        store.create_job(job)
        store.mark_ready(job["job_id"])
        exec_core = DeleteExecutorCore(
            store, self._approval_loader, self._approval_verify, self._tolaria_delete
        )
        result = exec_core.process_job(job["job_id"], "delete-worker")
        self.assertEqual(result["result_code"], "APPROVAL_INVALID")
        store.close()

    def test_g_mission_binding_removed(self):
        # Invariante: falsche Mission denied
        job = _make_delete_job()
        approval = _valid_approval(job)
        approval["mission_id"] = _uuid()
        self.approvals[job["approval_id"]] = approval
        store = JobStore(self.delete_db, "DELETE")
        store.create_job(job)
        store.mark_ready(job["job_id"])
        exec_core = DeleteExecutorCore(
            store, self._approval_loader, self._approval_verify, self._tolaria_delete
        )
        result = exec_core.process_job(job["job_id"], "delete-worker")
        self.assertEqual(result["result_code"], "MISSION_MISMATCH")
        store.close()

    def test_h_object_binding_removed(self):
        # Invariante: falsches Object denied
        job = _make_delete_job()
        approval = _valid_approval(job)
        approval["object_id"] = "obj-OTHER"
        self.approvals[job["approval_id"]] = approval
        store = JobStore(self.delete_db, "DELETE")
        store.create_job(job)
        store.mark_ready(job["job_id"])
        exec_core = DeleteExecutorCore(
            store, self._approval_loader, self._approval_verify, self._tolaria_delete
        )
        result = exec_core.process_job(job["job_id"], "delete-worker")
        self.assertEqual(result["result_code"], "OBJECT_MISMATCH")
        store.close()

    def test_i_commit_binding_removed(self):
        # Invariante: falscher Commit denied
        job = _make_delete_job()
        approval = _valid_approval(job)
        approval["expected_commit"] = _sha256("other-commit")
        self.approvals[job["approval_id"]] = approval
        store = JobStore(self.delete_db, "DELETE")
        store.create_job(job)
        store.mark_ready(job["job_id"])
        exec_core = DeleteExecutorCore(
            store, self._approval_loader, self._approval_verify, self._tolaria_delete
        )
        result = exec_core.process_job(job["job_id"], "delete-worker")
        self.assertEqual(result["result_code"], "COMMIT_MISMATCH")
        store.close()

    def test_j_provenance_binding_removed(self):
        # Invariante: falsche Provenance denied
        job = _make_delete_job()
        approval = _valid_approval(job)
        approval["expected_provenance_hash"] = _sha256("other-content")
        self.approvals[job["approval_id"]] = approval
        store = JobStore(self.delete_db, "DELETE")
        store.create_job(job)
        store.mark_ready(job["job_id"])
        exec_core = DeleteExecutorCore(
            store, self._approval_loader, self._approval_verify, self._tolaria_delete
        )
        result = exec_core.process_job(job["job_id"], "delete-worker")
        self.assertEqual(result["result_code"], "PROVENANCE_MISMATCH")
        store.close()

    def test_k_outcome_unknown_blind_retry_removed(self):
        # Invariante: OUTCOME_UNKNOWN -> kein blinder Retry
        def uncertain_save(vault_path, content):
            self.http_calls.append(("SAVE", vault_path))
            return {"status": "error", "code": "TIMEOUT", "uncertain": True}
        store = JobStore(self.save_db, "SAVE")
        job = _make_save_job(provenance_hash=self.provenance)
        store.create_job(job)
        store.mark_ready(job["job_id"])
        exec_core = SaveExecutorCore(store, self._source_loader, uncertain_save)
        exec_core.process_job(job["job_id"], "save-worker")
        exec_core.process_job(job["job_id"], "save-worker")
        self.assertEqual(len(self.http_calls), 1)
        store.close()

    def test_l_credential_fail_closed_removed(self):
        # Invariante: Credential fehlt -> kein HTTP
        def no_cred_save(vault_path, content):
            self.http_calls.append(("SAVE", vault_path))
            return {"status": "error", "code": "CREDENTIAL_MISSING", "uncertain": False}
        store = JobStore(self.save_db, "SAVE")
        job = _make_save_job(provenance_hash=self.provenance)
        store.create_job(job)
        store.mark_ready(job["job_id"])
        exec_core = SaveExecutorCore(store, self._source_loader, no_cred_save)
        result = exec_core.process_job(job["job_id"], "save-worker")
        self.assertEqual(result["result_code"], "CREDENTIAL_MISSING")
        store.close()

    def setUp(self):
        self.tmpdir = tempfile.mkdtemp()
        self.save_db = os.path.join(self.tmpdir, "c5a_save.db")
        self.delete_db = os.path.join(self.tmpdir, "c5a_delete.db")
        self.http_calls = []
        self.approvals = {}
        self.content = "hello world"
        self.provenance = _sha256(self.content)

    def _source_loader(self, commit, object_id):
        return self.content

    def _tolaria_save(self, vault_path, content):
        self.http_calls.append(("SAVE", vault_path))
        return {"status": "ok"}

    def _approval_loader(self, approval_id):
        return self.approvals.get(approval_id)

    def _approval_verify(self, approval):
        return {"valid": approval.get("signature_valid", False), "reason": "ok"}

    def _tolaria_delete(self, vault_path):
        self.http_calls.append(("DELETE", vault_path))
        return {"status": "ok"}


if __name__ == "__main__":
    unittest.main()