#!/usr/bin/env python3
"""
Red Queen — C5 DELETE-Execution FRESH CHECKER (Todo 12).

Unabhaengiger, adversarialer Checker. Versucht AUSDRUECKLICH, den
Human-Gated-DELETE-Contract zu brechen. Nutzt NUR die Produktionsmodule
(rq_c5a/rq_c5c/rq_c5_delete/rq_c5d) + isolierte Fakes — KEINE Maker-Testdateien,
KEINE Maker-Argumentation. Jeder Check ist ein eigenstaendiger Angriffsversuch.

Angriffsvektoren (Todo 12):
  wrong commit / wrong object / wrong path / wrong nonce
  missing approval / used approval / stale approval
  already absent / unexpected drift / crash after delete
  restart / search mismatch
  cross-object approval / cross-commit approval
  path traversal / arbitrary path deletion
  no auto-approval / no SQL bypass / no direct APPLIED bypass
  DELETE bleibt HUMAN_GATED

Ergebnis: PASS (kein Defekt) oder FAIL (Defekt reproduziert).
"""

from __future__ import annotations

import json
import os
import sys
import tempfile
import unittest
from typing import Any, Dict, List, Optional

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from rq_c5a import (
    C5AStore, ST_READY, ST_HUMAN_REVIEW_REQUIRED, ST_DELETE_APPROVED,
    ST_DELETING, ST_UPDATING_SEARCH, ST_VERIFYING_SEARCH, ST_APPLIED,
    RC_DELETE_APPROVAL_MISSING, RC_DELETE_APPROVAL_MISMATCH,
    RC_UNEXPECTED_TOLARIA_DRIFT,
    OP_DELETE_REQUEST, OP_CREATE,
)
from rq_c5c import TolariaClient, TolariaUnavailableError, RC_TOLARIA_UNAVAILABLE, VAULT_PREFIX
from rq_c5_delete import (
    DeleteExecutor, DeleteApprovalError, DeleteExecutionError,
    DELETE_ALREADY_AT_TARGET,
)
from rq_c5d import C5DEngine, SearchSourceBuilder, SearchClient
from rq_c5b import KnowledgeScope

UUID_A = "object/77b02661-d67b-4bae-b612-01d1287cea6b"
UUID_B = "object/48bd264f-607b-15f1-5f73-3e922af9b19d"


def _fm(id_: str, **extra) -> str:
    lines = ["---", "knowledge_schema: 1", f"id: {id_}"]
    for k, v in extra.items():
        if isinstance(v, list):
            lines.append(f"{k}: [{', '.join(v)}]")
        else:
            lines.append(f"{k}: {v}")
    lines.append("---")
    return "\n".join(lines) + "\n"


class FakeTolaria:
    def __init__(self):
        self.vault: Dict[str, str] = {}
        self.delete_count = 0
        self.unavailable = False

    def read(self, p: str) -> Optional[str]:
        if self.unavailable:
            raise TolariaUnavailableError("down", RC_TOLARIA_UNAVAILABLE)
        return self.vault.get(p)

    def write(self, p: str, c: str) -> Dict[str, Any]:
        self.vault[p] = c
        return {"ok": True}

    def delete(self, p: str) -> Dict[str, Any]:
        if p in self.vault:
            del self.vault[p]
            self.delete_count += 1
        return {"ok": True}

    def list(self, p: str = "/app/vault") -> List[Dict[str, Any]]:
        return [{"path": x} for x in self.vault]


class FakeTolariaClient(TolariaClient):
    def __init__(self, fake: FakeTolaria):
        super().__init__(base_url="http://fake")
        self.fake = fake

    def read(self, vault_path: str) -> Optional[str]:
        return self.fake.read(vault_path)

    def write(self, vault_path: str, content: str) -> Dict[str, Any]:
        return self.fake.write(vault_path, content)

    def delete(self, vault_path: str) -> Dict[str, Any]:
        return self.fake.delete(vault_path)

    def list(self, vault_path: str = "/app/vault") -> List[Dict[str, Any]]:
        return self.fake.list(vault_path)


class FakeSearch:
    def __init__(self, source_path: str):
        self.source_path = source_path
        self.rebuild_count = 0
        self.health_override: Optional[Dict[str, Any]] = None
        self.indexed_ids: List[str] = []
        self.indexed_paths: List[str] = []

    def _load(self) -> Dict[str, Any]:
        with open(self.source_path, "r", encoding="utf-8") as f:
            return json.load(f)

    def rebuild(self) -> Dict[str, Any]:
        self.rebuild_count += 1
        src = self._load()
        objs = src.get("objects", [])
        self.indexed_ids = [o["id"] for o in objs if o.get("id")]
        self.indexed_paths = [o["path"] for o in objs]
        return {"status": "ok", "indexed": len(objs)}

    def health(self) -> Dict[str, Any]:
        if self.health_override is not None:
            return self.health_override
        return {
            "index_built": True, "object_count": len(self.indexed_ids),
            "failed_objects": [], "integrity_ok": True,
            "indexed_object_ids": self.indexed_ids,
            "indexed_paths": self.indexed_paths,
            "secret_blocked_objects": 0,
        }


class FakeSearchClient(SearchClient):
    def __init__(self, fake: FakeSearch):
        super().__init__(base_url="http://fake-search")
        self.fake = fake

    def rebuild(self) -> Dict[str, Any]:
        return self.fake.rebuild()

    def health(self) -> Dict[str, Any]:
        return self.fake.health()


def _make_store() -> C5AStore:
    db = os.path.join(tempfile.mkdtemp(prefix="c5checker_db_"), "c5a.db")
    return C5AStore(db)


def _seed_delete(store: C5AStore, sha: str, oid: str, path: str,
                 status: str = ST_HUMAN_REVIEW_REQUIRED) -> int:
    store.upsert_commit({
        "commit_sha": sha, "parent_sha": "p", "sequence": 1,
        "status": status, "retry_count": 0,
    })
    oc = store.add_object_change({
        "commit_sha": sha, "object_id": oid, "operation": OP_DELETE_REQUEST,
        "path_before": path, "path_after": None,
    })
    if oc is None or oc.get("id") is None:
        raise RuntimeError("keine id")
    return oc["id"]


def _approve(store: C5AStore, sha: str, cid: int, oid: str, path: str) -> Dict[str, Any]:
    return store.create_delete_approval(
        workflow_commit=sha, object_change_id=cid, object_id=oid,
        path=path, approved_by="human:christian")


class CheckerFixture:
    def __init__(self):
        self.store = _make_store()
        self.fake = FakeTolaria()
        self.client = FakeTolariaClient(self.fake)
        self.tmpdir = tempfile.mkdtemp(prefix="c5checker_src_")
        self.source_path = os.path.join(self.tmpdir, "index_source.json")
        self.builder = SearchSourceBuilder(self.client, self.source_path,
                                           scope=KnowledgeScope(VAULT_PREFIX))
        self.search = FakeSearch(self.source_path)
        self.search_client = FakeSearchClient(self.search)
        self.engine = C5DEngine(self.store, search=self.search_client,
                                source_builder=self.builder)

    def seed(self, objs: Dict[str, str]) -> None:
        for p, c in objs.items():
            self.fake.vault[p] = c

    def cleanup(self) -> None:
        import shutil
        shutil.rmtree(self.tmpdir, ignore_errors=True)


class TestFreshChecker(unittest.TestCase):
    """Adversarialer Checker: versucht den Contract zu brechen."""

    def setUp(self):
        self.fx = CheckerFixture()
        self.fx.seed({
            "/app/vault/a.md": _fm(UUID_A, title="A"),
            "/app/vault/b.md": _fm(UUID_B, title="B"),
        })

    def tearDown(self):
        self.fx.cleanup()

    def _exec(self) -> DeleteExecutor:
        return DeleteExecutor(self.fx.store, self.fx.client)

    # -- wrong commit / object / path / nonce --------------------------------

    def test_wrong_commit_breaks(self):
        """Cross-commit: Approval fuer c2 darf c1 nicht freigeben."""
        cid = _seed_delete(self.fx.store, "c1", UUID_A, "a.md")
        _approve(self.fx.store, "c2", cid, UUID_A, "a.md")
        self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
        with self.assertRaises(DeleteApprovalError):
            self._exec().execute("c1")
        self.assertEqual(self.fx.fake.delete_count, 0)

    def test_wrong_object_breaks(self):
        """Cross-object: Approval fuer B darf A nicht freigeben."""
        cid = _seed_delete(self.fx.store, "c1", UUID_A, "a.md")
        _approve(self.fx.store, "c1", cid, UUID_B, "a.md")
        self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
        with self.assertRaises(DeleteApprovalError):
            self._exec().execute("c1")
        self.assertEqual(self.fx.fake.delete_count, 0)

    def test_wrong_path_breaks(self):
        """Cross-path: Approval fuer b.md darf a.md nicht freigeben."""
        cid = _seed_delete(self.fx.store, "c1", UUID_A, "a.md")
        _approve(self.fx.store, "c1", cid, UUID_A, "b.md")
        self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
        with self.assertRaises(DeleteApprovalError):
            self._exec().execute("c1")
        self.assertEqual(self.fx.fake.delete_count, 0)

    def test_wrong_nonce_breaks(self):
        """Manipulierter nonce darf keinen zusaetzlichen Pfad oeffnen."""
        cid = _seed_delete(self.fx.store, "c1", UUID_A, "a.md")
        _approve(self.fx.store, "c1", cid, UUID_A, "a.md")
        # nonce wird nicht als Autorisierung genutzt; Approval bleibt korrekt
        # gebunden. Der Checker verifiziert, dass nonce keinen Replay oeffnet.
        self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
        r = self._exec().execute("c1")
        self.assertEqual(r["status"], ST_UPDATING_SEARCH)
        self.assertEqual(self.fx.fake.delete_count, 1)

    # -- missing / used / stale approval ------------------------------------

    def test_missing_approval_breaks(self):
        """Kein Delete ohne persistierte Approval."""
        cid = _seed_delete(self.fx.store, "c1", UUID_A, "a.md")
        self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
        with self.assertRaises(DeleteApprovalError) as ctx:
            self._exec().execute("c1")
        self.assertEqual(ctx.exception.reason_code, RC_DELETE_APPROVAL_MISSING)
        self.assertEqual(self.fx.fake.delete_count, 0)

    def test_used_approval_breaks(self):
        """USED-Approval darf nicht erneut einen DELETE autorisieren."""
        cid = _seed_delete(self.fx.store, "c1", UUID_A, "a.md")
        _approve(self.fx.store, "c1", cid, UUID_A, "a.md")
        self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
        self._exec().execute("c1")
        # Approval ist USED. Neuer Commit c2 mit gleicher Approval-id -> MISSING.
        self.fx.store.upsert_commit({
            "commit_sha": "c2", "parent_sha": "p", "sequence": 2,
            "status": ST_DELETE_APPROVED, "retry_count": 0,
        })
        cid2 = self.fx.store.add_object_change({
            "commit_sha": "c2", "object_id": UUID_A, "operation": OP_DELETE_REQUEST,
            "path_before": "a.md", "path_after": None,
        })
        self.assertIsNotNone(cid2)
        with self.assertRaises(DeleteApprovalError):
            self._exec().execute("c2")
        self.assertEqual(self.fx.fake.delete_count, 1)

    def test_stale_approval_breaks(self):
        """Stale Approval (Commit nicht DELETE_APPROVED) -> kein DELETE."""
        cid = _seed_delete(self.fx.store, "c1", UUID_A, "a.md")
        _approve(self.fx.store, "c1", cid, UUID_A, "a.md")
        # Commit bleibt HUMAN_REVIEW_REQUIRED (nicht DELETE_APPROVED)
        with self.assertRaises(DeleteExecutionError):
            self._exec().execute("c1")
        self.assertEqual(self.fx.fake.delete_count, 0)

    # -- already absent / unexpected drift / crash --------------------------

    def test_already_absent_no_second_delete(self):
        """Bereits absent -> kein zweiter destruktiver Write."""
        cid = _seed_delete(self.fx.store, "c1", UUID_A, "a.md")
        _approve(self.fx.store, "c1", cid, UUID_A, "a.md")
        self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
        del self.fx.fake.vault["/app/vault/a.md"]
        r = self._exec().execute("c1")
        self.assertEqual(r["idempotency"], DELETE_ALREADY_AT_TARGET)
        self.assertEqual(self.fx.fake.delete_count, 0)

    def test_unexpected_drift_breaks(self):
        """Unerwartete neue Objekte nach DELETE -> FAIL CLOSED."""
        cid = _seed_delete(self.fx.store, "c1", UUID_A, "a.md")
        _approve(self.fx.store, "c1", cid, UUID_A, "a.md")
        self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
        # Nach dem DELETE fuegt der Fake ein neues Objekt hinzu (Drift).
        orig_delete = self.fx.fake.delete

        def _delete_with_drift(p: str) -> Dict[str, Any]:
            r = orig_delete(p)
            self.fx.fake.vault["/app/vault/evil.md"] = _fm(UUID_B, title="EVIL")
            return r

        self.fx.fake.delete = _delete_with_drift
        with self.assertRaises(DeleteExecutionError):
            self._exec().execute("c1")
        self.assertEqual(self.fx.store.commit_status("c1"), ST_HUMAN_REVIEW_REQUIRED)

    def test_crash_after_delete_replay(self):
        """Crash nach DELETE: Replay -> DELETE_ALREADY_AT_TARGET, kein 2. Delete."""
        cid = _seed_delete(self.fx.store, "c1", UUID_A, "a.md")
        _approve(self.fx.store, "c1", cid, UUID_A, "a.md")
        self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
        self._exec().execute("c1")
        # Commit ist ST_UPDATING_SEARCH. Replay -> nichts zu tun.
        r = self._exec().replay("c1")
        self.assertEqual(r["idempotency"], "ALREADY_AT_TARGET")
        self.assertEqual(self.fx.fake.delete_count, 1)

    def test_restart_after_approval(self):
        """Restart nach Approval: Approval persistent, DELETE ausfuehrbar."""
        cid = _seed_delete(self.fx.store, "c1", UUID_A, "a.md")
        _approve(self.fx.store, "c1", cid, UUID_A, "a.md")
        self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
        store2 = C5AStore(self.fx.store.db_path)
        r = DeleteExecutor(store2, self.fx.client).execute("c1")
        self.assertEqual(r["status"], ST_UPDATING_SEARCH)
        self.assertEqual(self.fx.fake.delete_count, 1)

    # -- search mismatch ----------------------------------------------------

    def test_search_mismatch_no_applied(self):
        """Search-Mismatch -> KEIN APPLIED."""
        cid = _seed_delete(self.fx.store, "c1", UUID_A, "a.md")
        _approve(self.fx.store, "c1", cid, UUID_A, "a.md")
        self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
        self._exec().execute("c1")
        # Stale Index mit geloeschtem Objekt
        self.fx.search.health_override = {
            "index_built": True, "object_count": 2,
            "failed_objects": [], "integrity_ok": True,
            "indexed_object_ids": [UUID_A, UUID_B],
            "indexed_paths": ["a.md", "b.md"],
        }
        r = self.fx.engine.apply_commit("c1")
        self.assertNotEqual(r["status"], ST_APPLIED)

    # -- path traversal / arbitrary path ------------------------------------

    def test_path_traversal_breaks(self):
        """Path-Traversal -> BLOCK, kein DELETE."""
        cid = _seed_delete(self.fx.store, "c1", UUID_A, "../../etc/passwd")
        _approve(self.fx.store, "c1", cid, UUID_A, "../../etc/passwd")
        self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
        with self.assertRaises(DeleteExecutionError):
            self._exec().execute("c1")
        self.assertEqual(self.fx.fake.delete_count, 0)

    def test_arbitrary_path_no_object_change(self):
        """Freie Object-ID/Pfad ohne ObjectChange kann keinen Delete triggern."""
        # Kein ObjectChange angelegt. Executor findet keinen DELETE-Change.
        self.fx.store.upsert_commit({
            "commit_sha": "c1", "parent_sha": "p", "sequence": 1,
            "status": ST_DELETE_APPROVED, "retry_count": 0,
        })
        with self.assertRaises(DeleteApprovalError):
            self._exec().execute("c1")
        self.assertEqual(self.fx.fake.delete_count, 0)

    # -- no auto-approval / no SQL bypass / no APPLIED bypass ---------------

    def test_no_auto_approval(self):
        """Kein Auto-Approval: ohne CLI-approve gibt es keine Approval."""
        cid = _seed_delete(self.fx.store, "c1", UUID_A, "a.md")
        self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
        approvals = self.fx.store.list_delete_approvals("c1")
        self.assertEqual(len(approvals), 0)
        with self.assertRaises(DeleteApprovalError):
            self._exec().execute("c1")

    def test_no_direct_applied_bypass(self):
        """Kein direkter APPLIED-Bypass: Executor setzt nie APPLIED."""
        cid = _seed_delete(self.fx.store, "c1", UUID_A, "a.md")
        _approve(self.fx.store, "c1", cid, UUID_A, "a.md")
        self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
        r = self._exec().execute("c1")
        self.assertEqual(r["status"], ST_UPDATING_SEARCH)
        self.assertNotEqual(self.fx.store.commit_status("c1"), ST_APPLIED)

    def test_delete_stays_human_gated(self):
        """DELETE bleibt HUMAN_GATED: neuer DELETE-Request -> HUMAN_REVIEW_REQUIRED."""
        # C5B klassifiziert DELETE -> HUMAN_REVIEW_REQUIRED (kein Auto-Approval).
        # Der Executor verlangt ST_DELETE_APPROVED; ohne explizite Approval
        # bleibt der Commit HUMAN_REVIEW_REQUIRED.
        cid = _seed_delete(self.fx.store, "c1", UUID_A, "a.md")
        self.assertEqual(self.fx.store.commit_status("c1"), ST_HUMAN_REVIEW_REQUIRED)
        with self.assertRaises(DeleteExecutionError):
            self._exec().execute("c1")
        self.assertEqual(self.fx.fake.delete_count, 0)


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