From ae1996bd789eda3f0899dda159127deb04241c34 Mon Sep 17 00:00:00 2001 From: aah20 Date: Mon, 17 Aug 2026 12:29:02 +0300 Subject: [PATCH 1/5] feat(plugins): add GateProve safety boundary & hash-chained action ledger plugin --- plugins/gate_prove/README.md | 36 +++++ plugins/gate_prove/app/gate_prove_svc.py | 150 ++++++++++++++++++++ plugins/gate_prove/app/ledger.py | 79 +++++++++++ plugins/gate_prove/app/schema.py | 46 ++++++ plugins/gate_prove/conf/default.yml | 8 ++ plugins/gate_prove/hook.py | 8 ++ plugins/gate_prove/tests/test_gate_prove.py | 77 ++++++++++ 7 files changed, 404 insertions(+) create mode 100644 plugins/gate_prove/README.md create mode 100644 plugins/gate_prove/app/gate_prove_svc.py create mode 100644 plugins/gate_prove/app/ledger.py create mode 100644 plugins/gate_prove/app/schema.py create mode 100644 plugins/gate_prove/conf/default.yml create mode 100644 plugins/gate_prove/hook.py create mode 100644 plugins/gate_prove/tests/test_gate_prove.py diff --git a/plugins/gate_prove/README.md b/plugins/gate_prove/README.md new file mode 100644 index 000000000..a1c4d0faf --- /dev/null +++ b/plugins/gate_prove/README.md @@ -0,0 +1,36 @@ +# GateProve Plugin for MITRE Caldera + +Deterministic Gate/Prove safety boundary and append-only hash-chained Action Ledger for Caldera operations. + +## Overview + +When running automated adversary emulation against enterprise or staging infrastructure, executing high-blast abilities (such as `T1562` Impair Defenses or `T1485` Data Destruction) without strict safety controls creates severe operational risk. + +**GateProve** introduces a zero-trust safety boundary: +1. **`never_equate_intent_to_approval: true`**: High planner confidence or automated execution does not authorize destructive techniques. +2. **Simulation Fallback**: Unapproved destructive abilities automatically default to safe simulation mode without mutating underlying systems. +3. **HITL Prove Token**: Destructive execution requires an authorized cryptographic token (`CALDERA_PROVE_TOKEN`). +4. **Append-Only Action Ledger**: Every ability evaluation, receipt, and hash is recorded into an append-only JSONL ledger with SHA-256 chain verification for audit compliance (SOC 2, ISO 27001, NIST CSF). +5. **Atomic Kill-Switch**: Immediate freeze of operation ability dispatch via environment variable (`CALDERA_KILL_SWITCH=1`) or file sentinel (`artifacts/KILL`). + +## Configuration + +In `conf/default.yml`: + +```yaml +name: GateProve +enabled: true +prove_token: "your-hitl-secret-token" +ledger_path: "artifacts/caldera_action_ledger.jsonl" +kill_switch: false +``` + +## Running Tests + +```bash +python3 -m unittest plugins/gate_prove/tests/test_gate_prove.py +``` + +## License + +Apache-2.0 diff --git a/plugins/gate_prove/app/gate_prove_svc.py b/plugins/gate_prove/app/gate_prove_svc.py new file mode 100644 index 000000000..cffad21ee --- /dev/null +++ b/plugins/gate_prove/app/gate_prove_svc.py @@ -0,0 +1,150 @@ +from __future__ import annotations + +import hmac +import os +from pathlib import Path +from typing import Any, Dict, Optional + +from plugins.gate_prove.app.ledger import OperationLedger +from plugins.gate_prove.app.schema import ( + DESTRUCTIVE_ATTACK_PATTERNS, + AbilityDecision, + GateDisposition, +) + + +class GateProveService: + """Service to evaluate adversary abilities before dispatch in Caldera operations.""" + + def __init__( + self, + prove_token: str = "", + ledger: OperationLedger | None = None, + ledger_path: str = "artifacts/caldera_action_ledger.jsonl", + ) -> None: + self.prove_token = prove_token or os.environ.get("CALDERA_PROVE_TOKEN", "") + self.ledger = ledger if ledger is not None else OperationLedger(Path(ledger_path)) + + def is_kill_switch_engaged(self) -> bool: + flag = os.environ.get("CALDERA_KILL_SWITCH", "").strip().lower() + if flag in {"1", "true", "yes", "on"}: + return True + kill_file = Path(os.environ.get("CALDERA_KILL_SWITCH_FILE", "artifacts/KILL")) + return kill_file.exists() + + def is_destructive_technique(self, technique_id: str) -> bool: + tech_upper = technique_id.upper().strip() + for pattern in DESTRUCTIVE_ATTACK_PATTERNS: + if tech_upper.startswith(pattern): + return True + return False + + def evaluate_ability( + self, + operation_id: str, + ability_id: str, + technique_id: str, + technique_name: str = "", + approved: bool = False, + offered_token: str = "", + simulate: bool = False, + ) -> AbilityDecision: + # 1. Kill Switch Check + if self.is_kill_switch_engaged(): + lid, rhash = self.ledger.record( + operation_id, ability_id, technique_id, "deny", False, "kill_switch_engaged" + ) + return AbilityDecision( + operation_id=operation_id, + ability_id=ability_id, + technique_id=technique_id, + technique_name=technique_name, + disposition="deny", + allowed=False, + requires_hitl=True, + never_equate_intent_to_approval=True, + reason="kill_switch_engaged", + ledger_id=lid, + receipt_hash=rhash, + kill_switch=True, + ) + + # 2. Simulation Mode Check + if simulate: + lid, rhash = self.ledger.record( + operation_id, ability_id, technique_id, "simulate", True, "simulation_mode_requested" + ) + return AbilityDecision( + operation_id=operation_id, + ability_id=ability_id, + technique_id=technique_id, + technique_name=technique_name, + disposition="simulate", + allowed=True, + requires_hitl=False, + never_equate_intent_to_approval=True, + reason="simulation_mode_requested", + ledger_id=lid, + receipt_hash=rhash, + ) + + # 3. Destructive / High-Blast Techniques Check + if self.is_destructive_technique(technique_id): + token_valid = bool( + self.prove_token + and offered_token + and hmac.compare_digest(self.prove_token.strip(), offered_token.strip()) + ) + if approved and token_valid: + lid, rhash = self.ledger.record( + operation_id, ability_id, technique_id, "allow", True, "hitl_token_verified" + ) + return AbilityDecision( + operation_id=operation_id, + ability_id=ability_id, + technique_id=technique_id, + technique_name=technique_name, + disposition="allow", + allowed=True, + requires_hitl=True, + never_equate_intent_to_approval=True, + reason="hitl_token_verified", + ledger_id=lid, + receipt_hash=rhash, + ) + + # Unapproved destructive ability falls back to safe simulation + lid, rhash = self.ledger.record( + operation_id, ability_id, technique_id, "simulate", True, "unapproved_destructive_simulated" + ) + return AbilityDecision( + operation_id=operation_id, + ability_id=ability_id, + technique_id=technique_id, + technique_name=technique_name, + disposition="simulate", + allowed=True, + requires_hitl=True, + never_equate_intent_to_approval=True, + reason="unapproved_destructive_simulated", + ledger_id=lid, + receipt_hash=rhash, + ) + + # 4. Standard Non-Destructive Abilities (Discovery / Collection / Baseline) + lid, rhash = self.ledger.record( + operation_id, ability_id, technique_id, "allow", True, "safe_emulation_allowed" + ) + return AbilityDecision( + operation_id=operation_id, + ability_id=ability_id, + technique_id=technique_id, + technique_name=technique_name, + disposition="allow", + allowed=True, + requires_hitl=False, + never_equate_intent_to_approval=True, + reason="safe_emulation_allowed", + ledger_id=lid, + receipt_hash=rhash, + ) diff --git a/plugins/gate_prove/app/ledger.py b/plugins/gate_prove/app/ledger.py new file mode 100644 index 000000000..1b199f802 --- /dev/null +++ b/plugins/gate_prove/app/ledger.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +import hashlib +import json +import time +import uuid +from pathlib import Path +from typing import Any, List, Optional, Tuple + + +class OperationLedger: + """Append-only, SHA-256 hash-chained ledger for Caldera operation ability executions.""" + + def __init__(self, path: Path | None = None) -> None: + self.path = path + self.entries: list[dict[str, Any]] = [] + self._last_hash = "0" * 64 + if self.path and self.path.exists(): + self._load() + + def _load(self) -> None: + self.entries = [] + with open(self.path, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if line: + data = json.loads(line) + self.entries.append(data) + self._last_hash = data.get("receipt_hash", self._last_hash) + + def record( + self, + operation_id: str, + ability_id: str, + technique_id: str, + disposition: str, + allowed: bool, + reason: str, + metadata: dict[str, Any] | None = None, + ) -> tuple[str, str]: + ledger_id = str(uuid.uuid4()) + ts = time.time() + payload = { + "ledger_id": ledger_id, + "timestamp": ts, + "operation_id": operation_id, + "ability_id": ability_id, + "technique_id": technique_id, + "disposition": disposition, + "allowed": allowed, + "reason": reason, + "prev_hash": self._last_hash, + "metadata": metadata or {}, + } + raw = json.dumps(payload, sort_keys=True) + receipt_hash = hashlib.sha256(raw.encode("utf-8")).hexdigest() + payload["receipt_hash"] = receipt_hash + self._last_hash = receipt_hash + self.entries.append(payload) + + if self.path: + self.path.parent.mkdir(parents=True, exist_ok=True) + with open(self.path, "a", encoding="utf-8") as f: + f.write(json.dumps(payload, sort_keys=True) + "\n") + + return ledger_id, receipt_hash + + def verify_chain(self) -> bool: + prev = "0" * 64 + for entry in self.entries: + if entry.get("prev_hash") != prev: + return False + payload = {k: v for k, v in entry.items() if k != "receipt_hash"} + raw = json.dumps(payload, sort_keys=True) + calc_hash = hashlib.sha256(raw.encode("utf-8")).hexdigest() + if calc_hash != entry.get("receipt_hash"): + return False + prev = entry["receipt_hash"] + return True diff --git a/plugins/gate_prove/app/schema.py b/plugins/gate_prove/app/schema.py new file mode 100644 index 000000000..8aea622ba --- /dev/null +++ b/plugins/gate_prove/app/schema.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +from dataclasses import asdict, dataclass, field +from typing import Any, List, Literal + +GateDisposition = Literal["allow", "simulate", "deny"] +TechniqueTier = Literal["discovery", "collection", "lateral_movement", "persistence", "destructive", "defense_evasion"] + +DESTRUCTIVE_ATTACK_PATTERNS = { + "T1562", # Impair Defenses + "T1070", # Indicator Removal + "T1485", # Data Destruction + "T1486", # Data Encrypted for Impact + "T1489", # Service Stop + "T1529", # System Shutdown/Reboot + "T1561", # Disk Wipe +} + +INSTANT_AUDIT_CTA = "https://a2zsoc.com/productized-services#caldera-arsenal-tripwire" +CONSULTATION_CTA = "https://a2zsoc.com/consultation" + + +@dataclass(frozen=True) +class AbilityDecision: + """Evaluation result for an adversary ability in a Caldera operation.""" + + operation_id: str + ability_id: str + technique_id: str + technique_name: str + disposition: GateDisposition + allowed: bool + requires_hitl: bool + never_equate_intent_to_approval: bool + reason: str + ledger_id: str + receipt_hash: str + kill_switch: bool = False + compliance_mapping: list[str] = field( + default_factory=lambda: ["NIST_CSF_DE.CM", "SOC2_CC7.2", "ISO_27001_A.12.6.1"] + ) + instant_audit: str = INSTANT_AUDIT_CTA + consultation: str = CONSULTATION_CTA + + def to_dict(self) -> dict[str, Any]: + return asdict(self) diff --git a/plugins/gate_prove/conf/default.yml b/plugins/gate_prove/conf/default.yml new file mode 100644 index 000000000..ede1bcf3a --- /dev/null +++ b/plugins/gate_prove/conf/default.yml @@ -0,0 +1,8 @@ +--- +name: GateProve +description: Deterministic Gate/Prove safety boundary and hash-chained action ledger for Caldera operations. +enabled: true +prove_token: "" +ledger_path: "artifacts/caldera_action_ledger.jsonl" +kill_switch: false +simulation_default_for_destructive: true diff --git a/plugins/gate_prove/hook.py b/plugins/gate_prove/hook.py new file mode 100644 index 000000000..1f6236d90 --- /dev/null +++ b/plugins/gate_prove/hook.py @@ -0,0 +1,8 @@ +name = 'GateProve' +description = 'Deterministic Gate/Prove safety boundary and hash-chained action ledger for Caldera operations.' +address = '/plugin/gate_prove/gui' + + +async def enable(services): + """Enable GateProve safety hook and ledger service in Caldera server.""" + pass diff --git a/plugins/gate_prove/tests/test_gate_prove.py b/plugins/gate_prove/tests/test_gate_prove.py new file mode 100644 index 000000000..0f0de189d --- /dev/null +++ b/plugins/gate_prove/tests/test_gate_prove.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +import sys +import tempfile +import unittest +from pathlib import Path + +# Add plugins path to sys.path +_plugins_dir = Path(__file__).resolve().parent.parent.parent.parent +if str(_plugins_dir) not in sys.path: + sys.path.insert(0, str(_plugins_dir)) + +from plugins.gate_prove.app.gate_prove_svc import GateProveService +from plugins.gate_prove.app.ledger import OperationLedger + + +class TestGateProve(unittest.TestCase): + def setUp(self) -> None: + self.service = GateProveService(prove_token="secret-caldera-token", ledger=OperationLedger()) + + def test_safe_ability_allowed(self) -> None: + decision = self.service.evaluate_ability( + operation_id="op_1", + ability_id="ab_discovery", + technique_id="T1082", + technique_name="System Information Discovery", + ) + self.assertTrue(decision.allowed) + self.assertEqual(decision.disposition, "allow") + self.assertFalse(decision.requires_hitl) + + def test_unapproved_destructive_ability_simulated(self) -> None: + decision = self.service.evaluate_ability( + operation_id="op_1", + ability_id="ab_impair", + technique_id="T1562.001", + technique_name="Disable Security Tools", + ) + self.assertTrue(decision.allowed) + self.assertEqual(decision.disposition, "simulate") + self.assertEqual(decision.reason, "unapproved_destructive_simulated") + self.assertTrue(decision.never_equate_intent_to_approval) + + def test_destructive_ability_allowed_with_valid_hitl_token(self) -> None: + decision = self.service.evaluate_ability( + operation_id="op_1", + ability_id="ab_impair", + technique_id="T1562.001", + approved=True, + offered_token="secret-caldera-token", + ) + self.assertTrue(decision.allowed) + self.assertEqual(decision.disposition, "allow") + self.assertEqual(decision.reason, "hitl_token_verified") + + +class TestOperationLedger(unittest.TestCase): + def test_ledger_hash_chain(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + ledger_file = Path(tmp) / "ledger.jsonl" + ledger = OperationLedger(ledger_file) + + ledger.record("op-1", "ab-1", "T1082", "allow", True, "ok") + ledger.record("op-1", "ab-2", "T1562", "simulate", True, "simulated") + ledger.record("op-1", "ab-3", "T1485", "allow", True, "verified") + + self.assertEqual(len(ledger.entries), 3) + self.assertTrue(ledger.verify_chain()) + + # Verify persistent reload + reloaded = OperationLedger(ledger_file) + self.assertEqual(len(reloaded.entries), 3) + self.assertTrue(reloaded.verify_chain()) + + +if __name__ == "__main__": + unittest.main() From 54483ca2c48bcb5930253cd97a8a4542c9d1c7af Mon Sep 17 00:00:00 2001 From: aah20 Date: Sat, 22 Aug 2026 13:16:08 +0300 Subject: [PATCH 2/5] feat(gate-prove): validate generated ability manifests --- plugins/gate_prove/README.md | 9 ++ plugins/gate_prove/app/ability_manifest.py | 129 ++++++++++++++++++ plugins/gate_prove/app/gate_prove_svc.py | 44 ++++++- plugins/gate_prove/hook.py | 10 +- plugins/gate_prove/tests/test_gate_prove.py | 137 ++++++++++++++++++++ 5 files changed, 325 insertions(+), 4 deletions(-) create mode 100644 plugins/gate_prove/app/ability_manifest.py diff --git a/plugins/gate_prove/README.md b/plugins/gate_prove/README.md index a1c4d0faf..0beca221c 100644 --- a/plugins/gate_prove/README.md +++ b/plugins/gate_prove/README.md @@ -12,6 +12,13 @@ When running automated adversary emulation against enterprise or staging infrast 3. **HITL Prove Token**: Destructive execution requires an authorized cryptographic token (`CALDERA_PROVE_TOKEN`). 4. **Append-Only Action Ledger**: Every ability evaluation, receipt, and hash is recorded into an append-only JSONL ledger with SHA-256 chain verification for audit compliance (SOC 2, ISO 27001, NIST CSF). 5. **Atomic Kill-Switch**: Immediate freeze of operation ability dispatch via environment variable (`CALDERA_KILL_SWITCH=1`) or file sentinel (`artifacts/KILL`). +6. **Generated Ability Manifest**: AI-generated abilities fail closed unless they include parsers, cleanup, bounded scope, generator/model provenance, and a matching digest over execution-relevant fields. + +## AI-generated ability contract + +Call `evaluate_ability` with both `ability_manifest` and `provenance` before dispatching content produced by an LLM ability factory. GateProve validates that the ability is attributable, bounded, reversible, and unchanged since review. A missing manifest field or mismatched digest produces a `deny` decision and a ledger receipt containing the validation errors. + +Required scope fields are `targets`, `expires_at`, and a positive `max_executions`. Required provenance fields are `generator`, `model`, `created_at`, and `content_hash`; compute the latter with `ability_content_hash` from `app/ability_manifest.py`. ## Configuration @@ -25,6 +32,8 @@ ledger_path: "artifacts/caldera_action_ledger.jsonl" kill_switch: false ``` +At plugin enablement, `CALDERA_PROVE_TOKEN` and `CALDERA_GATE_PROVE_LEDGER` configure the registered `gate_prove_svc` service. The kill switch is evaluated for every decision, so it can freeze dispatch without restarting Caldera. + ## Running Tests ```bash diff --git a/plugins/gate_prove/app/ability_manifest.py b/plugins/gate_prove/app/ability_manifest.py new file mode 100644 index 000000000..ce3e4352b --- /dev/null +++ b/plugins/gate_prove/app/ability_manifest.py @@ -0,0 +1,129 @@ +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any, Mapping + + +SAFETY_RELEVANT_FIELDS = ( + "ability_id", + "name", + "description", + "tactic", + "technique_id", + "technique_name", + "platforms", + "privilege", + "parsers", + "cleanup", + "scope", +) + + +def ability_content_hash(ability: Mapping[str, Any]) -> str: + """Return a stable digest over fields that affect execution and cleanup.""" + canonical = {field: ability.get(field) for field in SAFETY_RELEVANT_FIELDS} + raw = json.dumps(canonical, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + return hashlib.sha256(raw.encode("utf-8")).hexdigest() + + +@dataclass(frozen=True) +class ManifestValidation: + valid: bool + errors: tuple[str, ...] = field(default_factory=tuple) + content_hash: str = "" + + +class AbilityManifestValidator: + """Validate the safety envelope attached to an AI-generated ability. + + The validator deliberately does not judge whether a command is malicious or + effective. It proves that generated content has the minimum metadata needed + for bounded, attributable and reversible adversary emulation. + """ + + REQUIRED_ABILITY_FIELDS = ( + "ability_id", + "name", + "description", + "tactic", + "technique_id", + "technique_name", + "platforms", + "parsers", + "cleanup", + "scope", + ) + REQUIRED_PROVENANCE_FIELDS = ("generator", "model", "created_at", "content_hash") + REQUIRED_SCOPE_FIELDS = ("targets", "expires_at", "max_executions") + + def validate( + self, + ability: Mapping[str, Any], + provenance: Mapping[str, Any], + *, + expected_ability_id: str = "", + expected_technique_id: str = "", + ) -> ManifestValidation: + errors: list[str] = [] + + self._require_nonempty(ability, self.REQUIRED_ABILITY_FIELDS, "ability", errors) + self._require_nonempty(provenance, self.REQUIRED_PROVENANCE_FIELDS, "provenance", errors) + + scope = ability.get("scope") + if isinstance(scope, Mapping): + self._require_nonempty(scope, self.REQUIRED_SCOPE_FIELDS, "scope", errors) + max_executions = scope.get("max_executions") + if not isinstance(max_executions, int) or isinstance(max_executions, bool) or max_executions < 1: + errors.append("scope.max_executions must be a positive integer") + targets = scope.get("targets") + if targets and (not isinstance(targets, list) or not all(isinstance(item, str) for item in targets)): + errors.append("scope.targets must be a list of target identifiers") + self._validate_expiry(scope.get("expires_at"), errors) + elif scope is not None: + errors.append("ability.scope must be an object") + + if "platforms" in ability and not isinstance(ability.get("platforms"), Mapping): + errors.append("ability.platforms must be an object") + if "parsers" in ability and not isinstance(ability.get("parsers"), list): + errors.append("ability.parsers must be a list") + if "cleanup" in ability and not isinstance(ability.get("cleanup"), list): + errors.append("ability.cleanup must be a list") + + digest = ability_content_hash(ability) + claimed_hash = provenance.get("content_hash") + if claimed_hash and claimed_hash != digest: + errors.append("provenance.content_hash does not match the ability manifest") + + if expected_ability_id and ability.get("ability_id") != expected_ability_id: + errors.append("ability.ability_id does not match the dispatch request") + if expected_technique_id and ability.get("technique_id") != expected_technique_id: + errors.append("ability.technique_id does not match the dispatch request") + + return ManifestValidation(valid=not errors, errors=tuple(errors), content_hash=digest) + + @staticmethod + def _validate_expiry(value: Any, errors: list[str]) -> None: + if not value: + return + try: + expiry = datetime.fromisoformat(str(value).replace("Z", "+00:00")) + except ValueError: + errors.append("scope.expires_at must be an ISO-8601 timestamp") + return + if expiry.tzinfo is None: + errors.append("scope.expires_at must include a timezone") + return + if expiry <= datetime.now(timezone.utc): + errors.append("scope.expires_at must be in the future") + + @staticmethod + def _require_nonempty( + value: Mapping[str, Any], fields: tuple[str, ...], prefix: str, errors: list[str] + ) -> None: + for field_name in fields: + item = value.get(field_name) + if item is None or item == "" or item == [] or item == {}: + errors.append(f"{prefix}.{field_name} is required") diff --git a/plugins/gate_prove/app/gate_prove_svc.py b/plugins/gate_prove/app/gate_prove_svc.py index cffad21ee..4cd7c3944 100644 --- a/plugins/gate_prove/app/gate_prove_svc.py +++ b/plugins/gate_prove/app/gate_prove_svc.py @@ -5,6 +5,7 @@ from pathlib import Path from typing import Any, Dict, Optional +from plugins.gate_prove.app.ability_manifest import AbilityManifestValidator from plugins.gate_prove.app.ledger import OperationLedger from plugins.gate_prove.app.schema import ( DESTRUCTIVE_ATTACK_PATTERNS, @@ -21,9 +22,11 @@ def __init__( prove_token: str = "", ledger: OperationLedger | None = None, ledger_path: str = "artifacts/caldera_action_ledger.jsonl", + manifest_validator: AbilityManifestValidator | None = None, ) -> None: self.prove_token = prove_token or os.environ.get("CALDERA_PROVE_TOKEN", "") self.ledger = ledger if ledger is not None else OperationLedger(Path(ledger_path)) + self.manifest_validator = manifest_validator or AbilityManifestValidator() def is_kill_switch_engaged(self) -> bool: flag = os.environ.get("CALDERA_KILL_SWITCH", "").strip().lower() @@ -48,6 +51,8 @@ def evaluate_ability( approved: bool = False, offered_token: str = "", simulate: bool = False, + ability_manifest: dict[str, Any] | None = None, + provenance: dict[str, Any] | None = None, ) -> AbilityDecision: # 1. Kill Switch Check if self.is_kill_switch_engaged(): @@ -69,7 +74,40 @@ def evaluate_ability( kill_switch=True, ) - # 2. Simulation Mode Check + # 2. AI-generated abilities must carry a complete, untampered safety envelope. + if ability_manifest is not None or provenance is not None: + validation = self.manifest_validator.validate( + ability_manifest or {}, + provenance or {}, + expected_ability_id=ability_id, + expected_technique_id=technique_id, + ) + if not validation.valid: + reason = "invalid_ability_manifest:" + ";".join(validation.errors) + lid, rhash = self.ledger.record( + operation_id, + ability_id, + technique_id, + "deny", + False, + reason, + metadata={"content_hash": validation.content_hash, "validation_errors": list(validation.errors)}, + ) + return AbilityDecision( + operation_id=operation_id, + ability_id=ability_id, + technique_id=technique_id, + technique_name=technique_name, + disposition="deny", + allowed=False, + requires_hitl=True, + never_equate_intent_to_approval=True, + reason=reason, + ledger_id=lid, + receipt_hash=rhash, + ) + + # 3. Simulation Mode Check if simulate: lid, rhash = self.ledger.record( operation_id, ability_id, technique_id, "simulate", True, "simulation_mode_requested" @@ -88,7 +126,7 @@ def evaluate_ability( receipt_hash=rhash, ) - # 3. Destructive / High-Blast Techniques Check + # 4. Destructive / High-Blast Techniques Check if self.is_destructive_technique(technique_id): token_valid = bool( self.prove_token @@ -131,7 +169,7 @@ def evaluate_ability( receipt_hash=rhash, ) - # 4. Standard Non-Destructive Abilities (Discovery / Collection / Baseline) + # 5. Standard Non-Destructive Abilities (Discovery / Collection / Baseline) lid, rhash = self.ledger.record( operation_id, ability_id, technique_id, "allow", True, "safe_emulation_allowed" ) diff --git a/plugins/gate_prove/hook.py b/plugins/gate_prove/hook.py index 1f6236d90..52c7c9632 100644 --- a/plugins/gate_prove/hook.py +++ b/plugins/gate_prove/hook.py @@ -1,3 +1,8 @@ +import os + +from plugins.gate_prove.app.gate_prove_svc import GateProveService + + name = 'GateProve' description = 'Deterministic Gate/Prove safety boundary and hash-chained action ledger for Caldera operations.' address = '/plugin/gate_prove/gui' @@ -5,4 +10,7 @@ async def enable(services): """Enable GateProve safety hook and ledger service in Caldera server.""" - pass + services['gate_prove_svc'] = GateProveService( + prove_token=os.environ.get('CALDERA_PROVE_TOKEN', ''), + ledger_path=os.environ.get('CALDERA_GATE_PROVE_LEDGER', 'artifacts/caldera_action_ledger.jsonl'), + ) diff --git a/plugins/gate_prove/tests/test_gate_prove.py b/plugins/gate_prove/tests/test_gate_prove.py index 0f0de189d..d552a5c08 100644 --- a/plugins/gate_prove/tests/test_gate_prove.py +++ b/plugins/gate_prove/tests/test_gate_prove.py @@ -3,6 +3,8 @@ import sys import tempfile import unittest +from unittest.mock import patch +from datetime import datetime, timedelta, timezone from pathlib import Path # Add plugins path to sys.path @@ -11,7 +13,9 @@ sys.path.insert(0, str(_plugins_dir)) from plugins.gate_prove.app.gate_prove_svc import GateProveService +from plugins.gate_prove.app.ability_manifest import ability_content_hash from plugins.gate_prove.app.ledger import OperationLedger +from plugins.gate_prove.hook import enable class TestGateProve(unittest.TestCase): @@ -53,6 +57,119 @@ def test_destructive_ability_allowed_with_valid_hitl_token(self) -> None: self.assertEqual(decision.disposition, "allow") self.assertEqual(decision.reason, "hitl_token_verified") + @staticmethod + def generated_ability() -> dict: + return { + "ability_id": "generated-1", + "name": "Collect range canary", + "description": "Read a synthetic canary in the authorized range.", + "tactic": "collection", + "technique_id": "T1005", + "technique_name": "Data from Local System", + "platforms": {"linux": {"sh": {"command": "read-range-canary"}}}, + "privilege": "User", + "parsers": [{"module": "plugins.stockpile.app.parsers.basic"}], + "cleanup": ["remove-range-canary-artifact"], + "scope": { + "targets": ["range-host-1"], + "expires_at": (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat(), + "max_executions": 1, + }, + } + + def test_complete_generated_ability_is_allowed(self) -> None: + ability = self.generated_ability() + provenance = { + "generator": "mitre-mcp", + "model": "test-model", + "created_at": datetime.now(timezone.utc).isoformat(), + "content_hash": ability_content_hash(ability), + } + decision = self.service.evaluate_ability( + operation_id="op_1", + ability_id=ability["ability_id"], + technique_id=ability["technique_id"], + ability_manifest=ability, + provenance=provenance, + ) + self.assertTrue(decision.allowed) + self.assertEqual(decision.disposition, "allow") + + def test_generated_ability_without_cleanup_fails_closed(self) -> None: + ability = self.generated_ability() + ability["cleanup"] = [] + provenance = { + "generator": "mitre-mcp", + "model": "test-model", + "created_at": datetime.now(timezone.utc).isoformat(), + "content_hash": ability_content_hash(ability), + } + decision = self.service.evaluate_ability( + operation_id="op_1", + ability_id=ability["ability_id"], + technique_id=ability["technique_id"], + ability_manifest=ability, + provenance=provenance, + ) + self.assertFalse(decision.allowed) + self.assertEqual(decision.disposition, "deny") + self.assertIn("ability.cleanup is required", decision.reason) + + def test_generated_ability_hash_mismatch_fails_closed(self) -> None: + ability = self.generated_ability() + provenance = { + "generator": "mitre-mcp", + "model": "test-model", + "created_at": datetime.now(timezone.utc).isoformat(), + "content_hash": "0" * 64, + } + decision = self.service.evaluate_ability( + operation_id="op_1", + ability_id=ability["ability_id"], + technique_id=ability["technique_id"], + ability_manifest=ability, + provenance=provenance, + ) + self.assertFalse(decision.allowed) + self.assertIn("content_hash does not match", decision.reason) + + def test_generated_ability_cannot_be_replayed_for_another_dispatch(self) -> None: + ability = self.generated_ability() + provenance = { + "generator": "mitre-mcp", + "model": "test-model", + "created_at": datetime.now(timezone.utc).isoformat(), + "content_hash": ability_content_hash(ability), + } + decision = self.service.evaluate_ability( + operation_id="op_1", + ability_id="different-ability", + technique_id=ability["technique_id"], + ability_manifest=ability, + provenance=provenance, + ) + self.assertFalse(decision.allowed) + self.assertIn("does not match the dispatch request", decision.reason) + + def test_expired_generated_ability_scope_fails_closed(self) -> None: + ability = self.generated_ability() + ability["scope"]["expires_at"] = (datetime.now(timezone.utc) - timedelta(seconds=1)).isoformat() + provenance = { + "generator": "mitre-mcp", + "model": "test-model", + "created_at": datetime.now(timezone.utc).isoformat(), + "content_hash": ability_content_hash(ability), + } + decision = self.service.evaluate_ability( + operation_id="op_1", + ability_id=ability["ability_id"], + technique_id=ability["technique_id"], + ability_manifest=ability, + provenance=provenance, + ) + self.assertFalse(decision.allowed) + self.assertIn("scope.expires_at must be in the future", decision.reason) + class TestOperationLedger(unittest.TestCase): def test_ledger_hash_chain(self) -> None: @@ -73,5 +190,25 @@ def test_ledger_hash_chain(self) -> None: self.assertTrue(reloaded.verify_chain()) +class TestPluginHook(unittest.IsolatedAsyncioTestCase): + async def test_enable_registers_configured_service(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + ledger_path = str(Path(tmp) / "ledger.jsonl") + services = {} + with patch.dict( + "os.environ", + { + "CALDERA_PROVE_TOKEN": "configured-token", + "CALDERA_GATE_PROVE_LEDGER": ledger_path, + }, + clear=False, + ): + await enable(services) + + service = services["gate_prove_svc"] + self.assertEqual(service.prove_token, "configured-token") + self.assertEqual(service.ledger.path, Path(ledger_path)) + + if __name__ == "__main__": unittest.main() From f2360ef299a252d04b3a50eb8fb8337f930ab941 Mon Sep 17 00:00:00 2001 From: aah20 Date: Sat, 22 Aug 2026 13:38:01 +0300 Subject: [PATCH 3/5] feat(gate-prove): enforce scoped dispatch authorization --- app/objects/c_operation.py | 14 +- app/service/rest_svc.py | 2 +- plugins/gate_prove/README.md | 8 +- plugins/gate_prove/app/authorization_lease.py | 132 +++++++++++++ plugins/gate_prove/app/gate_prove_svc.py | 92 +++++++-- plugins/gate_prove/conf/default.yml | 1 + plugins/gate_prove/hook.py | 4 +- plugins/gate_prove/tests/test_gate_prove.py | 180 +++++++++++++++++- 8 files changed, 412 insertions(+), 21 deletions(-) create mode 100644 plugins/gate_prove/app/authorization_lease.py diff --git a/app/objects/c_operation.py b/app/objects/c_operation.py index 9170b472a..6e268aa7e 100644 --- a/app/objects/c_operation.py +++ b/app/objects/c_operation.py @@ -214,6 +214,18 @@ def ran_ability_id(self, ability_id): return ability_id in [link.ability.ability_id for link in self.chain if link.finish] async def apply(self, link): + gate_prove_svc = BaseService.get_service('gate_prove_svc') + if gate_prove_svc: + decision = gate_prove_svc.govern_link(self, link) + if decision.disposition != 'allow': + self.add_link(link) + logging.warning( + 'GateProve prevented dispatch of ability %s in operation %s: %s', + link.ability.ability_id, + self.id, + decision.reason, + ) + return link.id while self.state != self.states['RUNNING']: if self.state == self.states['RUN_ONE_LINK']: self.add_link(link) @@ -438,7 +450,7 @@ async def _cleanup_operation(self, services): cleanup_count = 0 for member in self.agents: for link in await services.get('planning_svc').get_cleanup_links(self, member): - self.add_link(link) + await self.apply(link) cleanup_count += 1 if cleanup_count: self.state = self.states['CLEANUP'] diff --git a/app/service/rest_svc.py b/app/service/rest_svc.py index f879ca99d..df7514f0f 100644 --- a/app/service/rest_svc.py +++ b/app/service/rest_svc.py @@ -246,7 +246,7 @@ async def add_manual_command(self, access, data): link = Link.load(dict(command=encoded_command, paw=agent.paw, cleanup=0, ability=ability, score=0, jitter=2, executor=executor, status=operation.link_status())) link.apply_id(agent.host) - operation.add_link(link) + await operation.apply(link) return dict(link=link.unique) diff --git a/plugins/gate_prove/README.md b/plugins/gate_prove/README.md index 0beca221c..5470c42aa 100644 --- a/plugins/gate_prove/README.md +++ b/plugins/gate_prove/README.md @@ -9,10 +9,11 @@ When running automated adversary emulation against enterprise or staging infrast **GateProve** introduces a zero-trust safety boundary: 1. **`never_equate_intent_to_approval: true`**: High planner confidence or automated execution does not authorize destructive techniques. 2. **Simulation Fallback**: Unapproved destructive abilities automatically default to safe simulation mode without mutating underlying systems. -3. **HITL Prove Token**: Destructive execution requires an authorized cryptographic token (`CALDERA_PROVE_TOKEN`). +3. **Scoped Authorization Lease**: High-blast execution requires an integrity-protected, expiring lease bound to the operation, ability, exact command digest, target, approver, and execution budget. 4. **Append-Only Action Ledger**: Every ability evaluation, receipt, and hash is recorded into an append-only JSONL ledger with SHA-256 chain verification for audit compliance (SOC 2, ISO 27001, NIST CSF). 5. **Atomic Kill-Switch**: Immediate freeze of operation ability dispatch via environment variable (`CALDERA_KILL_SWITCH=1`) or file sentinel (`artifacts/KILL`). 6. **Generated Ability Manifest**: AI-generated abilities fail closed unless they include parsers, cleanup, bounded scope, generator/model provenance, and a matching digest over execution-relevant fields. +7. **Central Dispatch Enforcement**: `Operation.apply()` evaluates every planner, REST, scheduled, and direct link through GateProve. Denied or simulation-only links are retained as discarded audit records but never become executable agent instructions. ## AI-generated ability contract @@ -28,12 +29,17 @@ In `conf/default.yml`: name: GateProve enabled: true prove_token: "your-hitl-secret-token" +authorization_key: "use-CALDERA_AUTHORIZATION_KEY-in-production" ledger_path: "artifacts/caldera_action_ledger.jsonl" kill_switch: false ``` At plugin enablement, `CALDERA_PROVE_TOKEN` and `CALDERA_GATE_PROVE_LEDGER` configure the registered `gate_prove_svc` service. The kill switch is evaluated for every decision, so it can freeze dispatch without restarting Caldera. +Set `CALDERA_AUTHORIZATION_KEY` to a high-entropy server-side key used to issue and verify scoped authorization leases. Lease IDs and consumption counts are stored in the action ledger, so a one-execution approval cannot be replayed after a service restart. `CALDERA_PROVE_TOKEN` remains accepted as configuration for migration but no longer authorizes high-blast execution. + +Cleanup is context, not authority: setting CALDERA's cleanup flag never bypasses manifest, technique, target, lease, budget, or kill-switch enforcement. High-blast cleanup requires its own scoped authorization lease. + ## Running Tests ```bash diff --git a/plugins/gate_prove/app/authorization_lease.py b/plugins/gate_prove/app/authorization_lease.py new file mode 100644 index 000000000..9a9487d36 --- /dev/null +++ b/plugins/gate_prove/app/authorization_lease.py @@ -0,0 +1,132 @@ +from __future__ import annotations + +import base64 +import binascii +import hashlib +import hmac +import json +import secrets +import time +from dataclasses import dataclass, field +from typing import Any + + +def command_digest(command: Any) -> str: + """Bind authorization to the exact command bytes queued for dispatch.""" + if isinstance(command, bytes): + raw = command + else: + raw = str(command or "").encode("utf-8") + return hashlib.sha256(raw).hexdigest() + + +@dataclass(frozen=True) +class LeaseValidation: + valid: bool + reason: str + claims: dict[str, Any] = field(default_factory=dict) + + +class AuthorizationLeaseIssuer: + """Issue and verify scoped, expiring HMAC authorization leases.""" + + def __init__(self, secret: str, issuer: str = "gate-prove") -> None: + self._secret = secret.encode("utf-8") + self.issuer = issuer + + @property + def configured(self) -> bool: + return bool(self._secret) + + def issue( + self, + *, + operation_id: str, + ability_id: str, + technique_id: str, + ability_digest: str, + target: str, + approver: str, + ttl_seconds: int = 300, + max_executions: int = 1, + now: int | None = None, + ) -> str: + if not self.configured: + raise ValueError("authorization lease key is not configured") + if ttl_seconds < 1 or max_executions < 1: + raise ValueError("ttl_seconds and max_executions must be positive") + issued_at = int(time.time() if now is None else now) + claims = { + "iss": self.issuer, + "jti": secrets.token_urlsafe(16), + "operation_id": operation_id, + "ability_id": ability_id, + "technique_id": technique_id, + "ability_digest": ability_digest, + "target": target, + "approver": approver, + "iat": issued_at, + "exp": issued_at + ttl_seconds, + "max_executions": max_executions, + } + payload = self._encode(json.dumps(claims, sort_keys=True, separators=(",", ":")).encode("utf-8")) + signature = self._encode(hmac.new(self._secret, payload.encode("ascii"), hashlib.sha256).digest()) + return f"{payload}.{signature}" + + def verify( + self, + token: str, + *, + operation_id: str, + ability_id: str, + technique_id: str, + ability_digest: str, + target: str, + now: int | None = None, + ) -> LeaseValidation: + if not self.configured: + return LeaseValidation(False, "authorization_lease_key_unconfigured") + if not token: + return LeaseValidation(False, "authorization_lease_missing") + try: + payload, supplied_signature = token.split(".", 1) + expected_signature = self._encode( + hmac.new(self._secret, payload.encode("ascii"), hashlib.sha256).digest() + ) + if not hmac.compare_digest(expected_signature, supplied_signature): + return LeaseValidation(False, "authorization_lease_signature_invalid") + claims = json.loads(self._decode(payload)) + except (ValueError, TypeError, UnicodeDecodeError, binascii.Error, json.JSONDecodeError): + return LeaseValidation(False, "authorization_lease_malformed") + if not isinstance(claims, dict): + return LeaseValidation(False, "authorization_lease_malformed") + + expected = { + "iss": self.issuer, + "operation_id": operation_id, + "ability_id": ability_id, + "technique_id": technique_id, + "ability_digest": ability_digest, + "target": target, + } + for field_name, expected_value in expected.items(): + if claims.get(field_name) != expected_value: + return LeaseValidation(False, f"authorization_lease_{field_name}_mismatch", claims) + + current_time = int(time.time() if now is None else now) + if not isinstance(claims.get("exp"), int) or claims["exp"] <= current_time: + return LeaseValidation(False, "authorization_lease_expired", claims) + if not isinstance(claims.get("max_executions"), int) or claims["max_executions"] < 1: + return LeaseValidation(False, "authorization_lease_budget_invalid", claims) + if not claims.get("jti") or not claims.get("approver"): + return LeaseValidation(False, "authorization_lease_claims_incomplete", claims) + return LeaseValidation(True, "authorization_lease_verified", claims) + + @staticmethod + def _encode(value: bytes) -> str: + return base64.urlsafe_b64encode(value).rstrip(b"=").decode("ascii") + + @staticmethod + def _decode(value: str) -> str: + padding = "=" * (-len(value) % 4) + return base64.urlsafe_b64decode(value + padding).decode("utf-8") diff --git a/plugins/gate_prove/app/gate_prove_svc.py b/plugins/gate_prove/app/gate_prove_svc.py index 4cd7c3944..d207eb1f3 100644 --- a/plugins/gate_prove/app/gate_prove_svc.py +++ b/plugins/gate_prove/app/gate_prove_svc.py @@ -1,11 +1,11 @@ from __future__ import annotations -import hmac import os from pathlib import Path from typing import Any, Dict, Optional from plugins.gate_prove.app.ability_manifest import AbilityManifestValidator +from plugins.gate_prove.app.authorization_lease import AuthorizationLeaseIssuer, command_digest from plugins.gate_prove.app.ledger import OperationLedger from plugins.gate_prove.app.schema import ( DESTRUCTIVE_ATTACK_PATTERNS, @@ -23,10 +23,13 @@ def __init__( ledger: OperationLedger | None = None, ledger_path: str = "artifacts/caldera_action_ledger.jsonl", manifest_validator: AbilityManifestValidator | None = None, + authorization_key: str = "", ) -> None: self.prove_token = prove_token or os.environ.get("CALDERA_PROVE_TOKEN", "") self.ledger = ledger if ledger is not None else OperationLedger(Path(ledger_path)) self.manifest_validator = manifest_validator or AbilityManifestValidator() + lease_key = authorization_key or os.environ.get("CALDERA_AUTHORIZATION_KEY", "") + self.lease_issuer = AuthorizationLeaseIssuer(lease_key) def is_kill_switch_engaged(self) -> bool: flag = os.environ.get("CALDERA_KILL_SWITCH", "").strip().lower() @@ -53,6 +56,10 @@ def evaluate_ability( simulate: bool = False, ability_manifest: dict[str, Any] | None = None, provenance: dict[str, Any] | None = None, + cleanup: bool = False, + authorization_lease: str = "", + ability_digest: str = "", + target: str = "", ) -> AbilityDecision: # 1. Kill Switch Check if self.is_kill_switch_engaged(): @@ -128,14 +135,31 @@ def evaluate_ability( # 4. Destructive / High-Blast Techniques Check if self.is_destructive_technique(technique_id): - token_valid = bool( - self.prove_token - and offered_token - and hmac.compare_digest(self.prove_token.strip(), offered_token.strip()) + lease = self.lease_issuer.verify( + authorization_lease, + operation_id=operation_id, + ability_id=ability_id, + technique_id=technique_id, + ability_digest=ability_digest, + target=target, ) - if approved and token_valid: + executions = self._lease_execution_count(lease.claims.get("jti", "")) + if lease.valid and executions < lease.claims["max_executions"]: lid, rhash = self.ledger.record( - operation_id, ability_id, technique_id, "allow", True, "hitl_token_verified" + operation_id, + ability_id, + technique_id, + "allow", + True, + "authorization_lease_verified", + metadata={ + "lease_id": lease.claims["jti"], + "approver": lease.claims["approver"], + "ability_digest": ability_digest, + "target": target, + "execution": executions + 1, + "max_executions": lease.claims["max_executions"], + }, ) return AbilityDecision( operation_id=operation_id, @@ -146,14 +170,17 @@ def evaluate_ability( allowed=True, requires_hitl=True, never_equate_intent_to_approval=True, - reason="hitl_token_verified", + reason="authorization_lease_verified", ledger_id=lid, receipt_hash=rhash, ) - # Unapproved destructive ability falls back to safe simulation + reason = lease.reason + if lease.valid: + reason = "authorization_lease_budget_exhausted" + # Unapproved or invalid high-blast activity falls back to safe simulation. lid, rhash = self.ledger.record( - operation_id, ability_id, technique_id, "simulate", True, "unapproved_destructive_simulated" + operation_id, ability_id, technique_id, "simulate", True, reason ) return AbilityDecision( operation_id=operation_id, @@ -164,7 +191,7 @@ def evaluate_ability( allowed=True, requires_hitl=True, never_equate_intent_to_approval=True, - reason="unapproved_destructive_simulated", + reason=reason, ledger_id=lid, receipt_hash=rhash, ) @@ -186,3 +213,46 @@ def evaluate_ability( ledger_id=lid, receipt_hash=rhash, ) + + def issue_authorization_lease(self, **claims: Any) -> str: + """Create a lease for an approval UI or other trusted control-plane caller.""" + return self.lease_issuer.issue(**claims) + + def _lease_execution_count(self, lease_id: str) -> int: + if not lease_id: + return 0 + return sum( + 1 + for entry in self.ledger.entries + if entry.get("reason") == "authorization_lease_verified" + and entry.get("metadata", {}).get("lease_id") == lease_id + ) + + def evaluate_link(self, operation: Any, link: Any) -> AbilityDecision: + """Evaluate a Caldera link immediately before it enters the chain.""" + ability = link.ability + digest = command_digest(getattr(link, "command", "")) + target = str(getattr(link, "paw", "") or getattr(link, "host", "")) + return self.evaluate_ability( + operation_id=operation.id, + ability_id=ability.ability_id, + technique_id=ability.technique_id, + technique_name=ability.technique_name, + approved=bool(getattr(link, "gate_prove_approved", False)), + offered_token=str(getattr(link, "gate_prove_token", "")), + simulate=bool(getattr(link, "gate_prove_simulate", False)), + ability_manifest=getattr(ability, "gate_prove_manifest", None), + provenance=getattr(ability, "gate_prove_provenance", None), + cleanup=bool(getattr(link, "cleanup", False)), + authorization_lease=str(getattr(link, "gate_prove_authorization_lease", "")), + ability_digest=digest, + target=target, + ) + + def govern_link(self, operation: Any, link: Any) -> AbilityDecision: + """Apply a gate decision to a link before Caldera queues it for an agent.""" + decision = self.evaluate_link(operation, link) + link.gate_prove_decision = decision.to_dict() + if decision.disposition != "allow": + link.status = link.states["DISCARD"] + return decision diff --git a/plugins/gate_prove/conf/default.yml b/plugins/gate_prove/conf/default.yml index ede1bcf3a..90581879e 100644 --- a/plugins/gate_prove/conf/default.yml +++ b/plugins/gate_prove/conf/default.yml @@ -3,6 +3,7 @@ name: GateProve description: Deterministic Gate/Prove safety boundary and hash-chained action ledger for Caldera operations. enabled: true prove_token: "" +authorization_key: "" ledger_path: "artifacts/caldera_action_ledger.jsonl" kill_switch: false simulation_default_for_destructive: true diff --git a/plugins/gate_prove/hook.py b/plugins/gate_prove/hook.py index 52c7c9632..6101cabbf 100644 --- a/plugins/gate_prove/hook.py +++ b/plugins/gate_prove/hook.py @@ -10,7 +10,9 @@ async def enable(services): """Enable GateProve safety hook and ledger service in Caldera server.""" - services['gate_prove_svc'] = GateProveService( + service = GateProveService( prove_token=os.environ.get('CALDERA_PROVE_TOKEN', ''), + authorization_key=os.environ.get('CALDERA_AUTHORIZATION_KEY', ''), ledger_path=os.environ.get('CALDERA_GATE_PROVE_LEDGER', 'artifacts/caldera_action_ledger.jsonl'), ) + services['gate_prove_svc'] = service diff --git a/plugins/gate_prove/tests/test_gate_prove.py b/plugins/gate_prove/tests/test_gate_prove.py index d552a5c08..d505ce5ea 100644 --- a/plugins/gate_prove/tests/test_gate_prove.py +++ b/plugins/gate_prove/tests/test_gate_prove.py @@ -6,6 +6,7 @@ from unittest.mock import patch from datetime import datetime, timedelta, timezone from pathlib import Path +from types import SimpleNamespace # Add plugins path to sys.path _plugins_dir = Path(__file__).resolve().parent.parent.parent.parent @@ -14,13 +15,16 @@ from plugins.gate_prove.app.gate_prove_svc import GateProveService from plugins.gate_prove.app.ability_manifest import ability_content_hash +from plugins.gate_prove.app.authorization_lease import AuthorizationLeaseIssuer, command_digest from plugins.gate_prove.app.ledger import OperationLedger from plugins.gate_prove.hook import enable class TestGateProve(unittest.TestCase): def setUp(self) -> None: - self.service = GateProveService(prove_token="secret-caldera-token", ledger=OperationLedger()) + self.service = GateProveService( + authorization_key="lease-signing-key", ledger=OperationLedger() + ) def test_safe_ability_allowed(self) -> None: decision = self.service.evaluate_ability( @@ -42,20 +46,98 @@ def test_unapproved_destructive_ability_simulated(self) -> None: ) self.assertTrue(decision.allowed) self.assertEqual(decision.disposition, "simulate") - self.assertEqual(decision.reason, "unapproved_destructive_simulated") + self.assertEqual(decision.reason, "authorization_lease_missing") self.assertTrue(decision.never_equate_intent_to_approval) - def test_destructive_ability_allowed_with_valid_hitl_token(self) -> None: + def test_destructive_ability_allowed_with_scoped_lease(self) -> None: + digest = command_digest("disable-range-control") + lease = self.service.issue_authorization_lease( + operation_id="op_1", + ability_id="ab_impair", + technique_id="T1562.001", + ability_digest=digest, + target="range-host-1", + approver="security-lead@example.test", + ) decision = self.service.evaluate_ability( operation_id="op_1", ability_id="ab_impair", technique_id="T1562.001", - approved=True, - offered_token="secret-caldera-token", + authorization_lease=lease, + ability_digest=digest, + target="range-host-1", ) self.assertTrue(decision.allowed) self.assertEqual(decision.disposition, "allow") - self.assertEqual(decision.reason, "hitl_token_verified") + self.assertEqual(decision.reason, "authorization_lease_verified") + + def test_authorization_lease_cannot_move_to_another_target(self) -> None: + digest = command_digest("disable-range-control") + lease = self.service.issue_authorization_lease( + operation_id="op_1", + ability_id="ab_impair", + technique_id="T1562.001", + ability_digest=digest, + target="range-host-1", + approver="security-lead@example.test", + ) + decision = self.service.evaluate_ability( + operation_id="op_1", + ability_id="ab_impair", + technique_id="T1562.001", + authorization_lease=lease, + ability_digest=digest, + target="range-host-2", + ) + self.assertEqual(decision.disposition, "simulate") + self.assertEqual(decision.reason, "authorization_lease_target_mismatch") + + def test_authorization_lease_execution_budget_is_enforced(self) -> None: + digest = command_digest("disable-range-control") + lease = self.service.issue_authorization_lease( + operation_id="op_1", + ability_id="ab_impair", + technique_id="T1562.001", + ability_digest=digest, + target="range-host-1", + approver="security-lead@example.test", + max_executions=1, + ) + request = { + "operation_id": "op_1", + "ability_id": "ab_impair", + "technique_id": "T1562.001", + "authorization_lease": lease, + "ability_digest": digest, + "target": "range-host-1", + } + first = self.service.evaluate_ability(**request) + second = self.service.evaluate_ability(**request) + self.assertEqual(first.disposition, "allow") + self.assertEqual(second.disposition, "simulate") + self.assertEqual(second.reason, "authorization_lease_budget_exhausted") + + def test_kill_switch_also_blocks_cleanup_link(self) -> None: + with patch.dict("os.environ", {"CALDERA_KILL_SWITCH": "1"}, clear=False): + decision = self.service.evaluate_ability( + operation_id="op_1", + ability_id="cleanup-1", + technique_id="T1562.001", + cleanup=True, + ) + self.assertFalse(decision.allowed) + self.assertEqual(decision.reason, "kill_switch_engaged") + self.assertTrue(decision.kill_switch) + + def test_cleanup_flag_does_not_bypass_high_blast_authorization(self) -> None: + decision = self.service.evaluate_ability( + operation_id="op_1", + ability_id="cleanup-1", + technique_id="T1562.001", + cleanup=True, + ) + self.assertEqual(decision.disposition, "simulate") + self.assertEqual(decision.reason, "authorization_lease_missing") @staticmethod def generated_ability() -> dict: @@ -190,6 +272,35 @@ def test_ledger_hash_chain(self) -> None: self.assertTrue(reloaded.verify_chain()) +class TestAuthorizationLease(unittest.TestCase): + def setUp(self) -> None: + self.issuer = AuthorizationLeaseIssuer("test-signing-key") + self.claims = { + "operation_id": "op-1", + "ability_id": "ability-1", + "technique_id": "T1562.001", + "ability_digest": command_digest("range-command"), + "target": "range-host-1", + "approver": "security-lead@example.test", + } + + def test_expired_lease_fails_closed(self) -> None: + token = self.issuer.issue(**self.claims, ttl_seconds=30, now=100) + result = self.issuer.verify(token, **{k: v for k, v in self.claims.items() if k != "approver"}, now=131) + self.assertFalse(result.valid) + self.assertEqual(result.reason, "authorization_lease_expired") + + def test_tampered_lease_fails_closed(self) -> None: + token = self.issuer.issue(**self.claims) + payload, signature = token.split(".", 1) + replacement = "A" if payload[-1] != "A" else "B" + result = self.issuer.verify( + f"{payload[:-1]}{replacement}.{signature}", + **{k: v for k, v in self.claims.items() if k != "approver"}, + ) + self.assertFalse(result.valid) + + class TestPluginHook(unittest.IsolatedAsyncioTestCase): async def test_enable_registers_configured_service(self) -> None: with tempfile.TemporaryDirectory() as tmp: @@ -210,5 +321,62 @@ async def test_enable_registers_configured_service(self) -> None: self.assertEqual(service.ledger.path, Path(ledger_path)) +class TestDispatchBoundary(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.service = GateProveService(authorization_key="dispatch-key", ledger=OperationLedger()) + self.operation = SimpleNamespace(id="governed-operation") + + @staticmethod + def link(technique_id: str, cleanup: int = 0) -> SimpleNamespace: + return SimpleNamespace( + id=f"link-{technique_id}", + ability=SimpleNamespace( + ability_id=f"ability-{technique_id}", + technique_id=technique_id, + technique_name="Test technique", + ), + cleanup=cleanup, + command=f"command-{technique_id}", + paw="range-host-1", + status=-3, + states={"DISCARD": -2}, + ) + + async def test_governor_discards_unapproved_high_blast_link(self) -> None: + link = self.link("T1562.001") + + decision = self.service.govern_link(self.operation, link) + + self.assertEqual(link.status, link.states["DISCARD"]) + self.assertEqual(decision.disposition, "simulate") + self.assertEqual(link.gate_prove_decision["disposition"], "simulate") + + async def test_governor_allows_standard_link(self) -> None: + link = self.link("T1082") + + decision = self.service.govern_link(self.operation, link) + + self.assertEqual(link.status, -3) + self.assertEqual(decision.disposition, "allow") + self.assertEqual(link.gate_prove_decision["disposition"], "allow") + + async def test_governor_discards_cleanup_during_kill_switch(self) -> None: + link = self.link("T1562.001", cleanup=1) + with patch.dict("os.environ", {"CALDERA_KILL_SWITCH": "1"}, clear=False): + decision = self.service.govern_link(self.operation, link) + + self.assertEqual(link.status, link.states["DISCARD"]) + self.assertEqual(decision.disposition, "deny") + self.assertEqual(link.gate_prove_decision["reason"], "kill_switch_engaged") + + async def test_cleanup_flag_cannot_bypass_high_blast_policy(self) -> None: + link = self.link("T1562.001", cleanup=1) + + decision = self.service.govern_link(self.operation, link) + + self.assertEqual(link.status, link.states["DISCARD"]) + self.assertEqual(decision.reason, "authorization_lease_missing") + + if __name__ == "__main__": unittest.main() From 8ddef1123cfb83ff7be46a71ac5985d8fd1329d3 Mon Sep 17 00:00:00 2001 From: aah20 Date: Sat, 22 Aug 2026 13:45:28 +0300 Subject: [PATCH 4/5] feat(gate-prove): attest governed operations --- plugins/gate_prove/README.md | 3 + plugins/gate_prove/app/attestation.py | 188 ++++++++++++++++++++ plugins/gate_prove/app/gate_prove_svc.py | 11 ++ plugins/gate_prove/conf/default.yml | 1 + plugins/gate_prove/hook.py | 1 + plugins/gate_prove/tests/test_gate_prove.py | 108 +++++++++++ 6 files changed, 312 insertions(+) create mode 100644 plugins/gate_prove/app/attestation.py diff --git a/plugins/gate_prove/README.md b/plugins/gate_prove/README.md index 5470c42aa..ba3e342a0 100644 --- a/plugins/gate_prove/README.md +++ b/plugins/gate_prove/README.md @@ -14,6 +14,7 @@ When running automated adversary emulation against enterprise or staging infrast 5. **Atomic Kill-Switch**: Immediate freeze of operation ability dispatch via environment variable (`CALDERA_KILL_SWITCH=1`) or file sentinel (`artifacts/KILL`). 6. **Generated Ability Manifest**: AI-generated abilities fail closed unless they include parsers, cleanup, bounded scope, generator/model provenance, and a matching digest over execution-relevant fields. 7. **Central Dispatch Enforcement**: `Operation.apply()` evaluates every planner, REST, scheduled, and direct link through GateProve. Denied or simulation-only links are retained as discarded audit records but never become executable agent instructions. +8. **Operation Attestation**: Completed operations can emit an integrity-protected evidence bundle containing input provenance, target and command digests, gate decisions, cleanup state, detection outcomes, and the ledger root without disclosing command contents. ## AI-generated ability contract @@ -38,6 +39,8 @@ At plugin enablement, `CALDERA_PROVE_TOKEN` and `CALDERA_GATE_PROVE_LEDGER` conf Set `CALDERA_AUTHORIZATION_KEY` to a high-entropy server-side key used to issue and verify scoped authorization leases. Lease IDs and consumption counts are stored in the action ledger, so a one-execution approval cannot be replayed after a service restart. `CALDERA_PROVE_TOKEN` remains accepted as configuration for migration but no longer authorizes high-blast execution. +Set a separate `CALDERA_ATTESTATION_KEY` to protect portable operation evidence. `attest_operation()` classifies results as `completed_verified`, `completed_with_detection_gaps`, `cleanup_incomplete`, `evidence_incomplete`, `evidence_invalid`, or `in_progress`. Detection evidence distinguishes execution failure, missing telemetry, missing detection, and failed SOC correlation. + Cleanup is context, not authority: setting CALDERA's cleanup flag never bypasses manifest, technique, target, lease, budget, or kill-switch enforcement. High-blast cleanup requires its own scoped authorization lease. ## Running Tests diff --git a/plugins/gate_prove/app/attestation.py b/plugins/gate_prove/app/attestation.py new file mode 100644 index 000000000..2c8eb42e0 --- /dev/null +++ b/plugins/gate_prove/app/attestation.py @@ -0,0 +1,188 @@ +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +from datetime import datetime, timezone +from typing import Any, Iterable + +from plugins.gate_prove.app.authorization_lease import command_digest +from plugins.gate_prove.app.ledger import OperationLedger + + +DETECTION_OUTCOMES = { + "detected", + "visibility_gap", + "detection_gap", + "soc_workflow_gap", + "invalid_test", +} + + +class OperationAttestor: + """Build and verify portable evidence for a governed Caldera operation.""" + + SCHEMA_VERSION = "1.0.0" + + def __init__(self, secret: str, ledger: OperationLedger) -> None: + self._secret = secret.encode("utf-8") + self.ledger = ledger + + def build( + self, + operation: Any, + *, + detection_results: Iterable[dict[str, Any]] = (), + input_provenance: Iterable[dict[str, Any]] = (), + ) -> dict[str, Any]: + if not self._secret: + raise ValueError("attestation key is not configured") + + links = [self._link_evidence(link) for link in getattr(operation, "chain", [])] + link_ids = {item["link_id"] for item in links} + detections = [self._normalize_detection(item, link_ids) for item in detection_results] + ledger_valid = self.ledger.verify_chain() + bundle = { + "schema_version": self.SCHEMA_VERSION, + "generated_at": datetime.now(timezone.utc).isoformat(), + "operation": { + "id": str(operation.id), + "name": str(getattr(operation, "name", "")), + "state": str(getattr(operation, "state", "")), + "started_at": self._timestamp(getattr(operation, "start", None)), + "finished_at": self._timestamp(getattr(operation, "finish", None)), + }, + "input_provenance": list(input_provenance), + "links": links, + "detection_results": detections, + "ledger": { + "valid": ledger_valid, + "entries": len(self.ledger.entries), + "root_hash": self.ledger.entries[-1]["receipt_hash"] if self.ledger.entries else "0" * 64, + }, + } + bundle["summary"] = self._summary(operation, links, detections, ledger_valid) + canonical = self._canonical(bundle) + bundle["bundle_hash"] = hashlib.sha256(canonical).hexdigest() + bundle["signature"] = self._encode(hmac.new(self._secret, canonical, hashlib.sha256).digest()) + return bundle + + def verify(self, bundle: dict[str, Any]) -> bool: + if not self._secret: + return False + supplied_hash = bundle.get("bundle_hash", "") + supplied_signature = bundle.get("signature", "") + unsigned = {key: value for key, value in bundle.items() if key not in {"bundle_hash", "signature"}} + canonical = self._canonical(unsigned) + expected_hash = hashlib.sha256(canonical).hexdigest() + expected_signature = self._encode(hmac.new(self._secret, canonical, hashlib.sha256).digest()) + return hmac.compare_digest(supplied_hash, expected_hash) and hmac.compare_digest( + supplied_signature, expected_signature + ) + + @staticmethod + def _link_evidence(link: Any) -> dict[str, Any]: + ability = link.ability + decision = getattr(link, "gate_prove_decision", None) + return { + "link_id": str(link.id), + "target": str(getattr(link, "paw", "") or getattr(link, "host", "")), + "ability_id": str(ability.ability_id), + "technique_id": str(ability.technique_id), + "command_digest": command_digest(getattr(link, "command", "")), + "cleanup": bool(getattr(link, "cleanup", False)), + "cleanup_state": OperationAttestor._cleanup_state(link, decision), + "status": getattr(link, "status", None), + "finished_at": OperationAttestor._timestamp(getattr(link, "finish", None)), + "gate_decision": decision, + } + + @staticmethod + def _cleanup_state(link: Any, decision: dict[str, Any] | None) -> str: + if not getattr(link, "cleanup", False): + return "not_applicable" + if not decision or decision.get("disposition") != "allow": + return "blocked" + states = getattr(link, "states", {}) + if getattr(link, "status", None) == states.get("SUCCESS", 0) and getattr(link, "finish", None): + return "verified" + if getattr(link, "status", None) in {states.get("ERROR", 1), states.get("TIMEOUT", 124)}: + return "failed" + return "pending" + + @staticmethod + def _normalize_detection(result: dict[str, Any], link_ids: set[str]) -> dict[str, Any]: + outcome = result.get("outcome") + if outcome not in DETECTION_OUTCOMES: + raise ValueError(f"unsupported detection outcome: {outcome}") + link_id = str(result.get("link_id", "")) + if link_id not in link_ids: + raise ValueError(f"detection evidence references unknown link: {link_id}") + if outcome == "detected" and not all( + result.get(field_name) for field_name in ("telemetry_source", "detection_id", "evidence_digest") + ): + raise ValueError("detected outcome requires telemetry_source, detection_id, and evidence_digest") + return { + "link_id": link_id, + "outcome": outcome, + "telemetry_source": str(result.get("telemetry_source", "")), + "detection_id": str(result.get("detection_id", "")), + "evidence_digest": str(result.get("evidence_digest", "")), + } + + @staticmethod + def _summary( + operation: Any, + links: list[dict[str, Any]], + detections: list[dict[str, Any]], + ledger_valid: bool, + ) -> dict[str, Any]: + cleanup_states = [item["cleanup_state"] for item in links if item["cleanup"]] + missing_decisions = sum(1 for item in links if not item["gate_decision"]) + expected_detection_links = { + item["link_id"] + for item in links + if not item["cleanup"] + and item["gate_decision"] + and item["gate_decision"].get("disposition") == "allow" + } + classified_detection_links = {result["link_id"] for result in detections} + missing_detection_links = sorted(expected_detection_links - classified_detection_links) + if not ledger_valid: + disposition = "evidence_invalid" + elif not getattr(operation, "finish", None): + disposition = "in_progress" + elif missing_decisions or missing_detection_links: + disposition = "evidence_incomplete" + elif any(state in {"pending", "failed", "blocked"} for state in cleanup_states): + disposition = "cleanup_incomplete" + elif any(result["outcome"] != "detected" for result in detections): + disposition = "completed_with_detection_gaps" + else: + disposition = "completed_verified" + return { + "disposition": disposition, + "links": len(links), + "missing_gate_decisions": missing_decisions, + "missing_detection_links": missing_detection_links, + "cleanup": {state: cleanup_states.count(state) for state in sorted(set(cleanup_states))}, + "detections": { + outcome: sum(1 for result in detections if result["outcome"] == outcome) + for outcome in sorted(DETECTION_OUTCOMES) + }, + } + + @staticmethod + def _canonical(value: dict[str, Any]) -> bytes: + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8") + + @staticmethod + def _timestamp(value: Any) -> str | None: + if value is None: + return None + return value.isoformat() if hasattr(value, "isoformat") else str(value) + + @staticmethod + def _encode(value: bytes) -> str: + return base64.urlsafe_b64encode(value).rstrip(b"=").decode("ascii") diff --git a/plugins/gate_prove/app/gate_prove_svc.py b/plugins/gate_prove/app/gate_prove_svc.py index d207eb1f3..4c0801126 100644 --- a/plugins/gate_prove/app/gate_prove_svc.py +++ b/plugins/gate_prove/app/gate_prove_svc.py @@ -6,6 +6,7 @@ from plugins.gate_prove.app.ability_manifest import AbilityManifestValidator from plugins.gate_prove.app.authorization_lease import AuthorizationLeaseIssuer, command_digest +from plugins.gate_prove.app.attestation import OperationAttestor from plugins.gate_prove.app.ledger import OperationLedger from plugins.gate_prove.app.schema import ( DESTRUCTIVE_ATTACK_PATTERNS, @@ -24,12 +25,15 @@ def __init__( ledger_path: str = "artifacts/caldera_action_ledger.jsonl", manifest_validator: AbilityManifestValidator | None = None, authorization_key: str = "", + attestation_key: str = "", ) -> None: self.prove_token = prove_token or os.environ.get("CALDERA_PROVE_TOKEN", "") self.ledger = ledger if ledger is not None else OperationLedger(Path(ledger_path)) self.manifest_validator = manifest_validator or AbilityManifestValidator() lease_key = authorization_key or os.environ.get("CALDERA_AUTHORIZATION_KEY", "") self.lease_issuer = AuthorizationLeaseIssuer(lease_key) + evidence_key = attestation_key or os.environ.get("CALDERA_ATTESTATION_KEY", "") + self.attestor = OperationAttestor(evidence_key, self.ledger) def is_kill_switch_engaged(self) -> bool: flag = os.environ.get("CALDERA_KILL_SWITCH", "").strip().lower() @@ -218,6 +222,13 @@ def issue_authorization_lease(self, **claims: Any) -> str: """Create a lease for an approval UI or other trusted control-plane caller.""" return self.lease_issuer.issue(**claims) + def attest_operation(self, operation: Any, **evidence: Any) -> dict[str, Any]: + """Build an integrity-protected operation evidence bundle.""" + return self.attestor.build(operation, **evidence) + + def verify_attestation(self, bundle: dict[str, Any]) -> bool: + return self.attestor.verify(bundle) + def _lease_execution_count(self, lease_id: str) -> int: if not lease_id: return 0 diff --git a/plugins/gate_prove/conf/default.yml b/plugins/gate_prove/conf/default.yml index 90581879e..eaf2aafd4 100644 --- a/plugins/gate_prove/conf/default.yml +++ b/plugins/gate_prove/conf/default.yml @@ -4,6 +4,7 @@ description: Deterministic Gate/Prove safety boundary and hash-chained action le enabled: true prove_token: "" authorization_key: "" +attestation_key: "" ledger_path: "artifacts/caldera_action_ledger.jsonl" kill_switch: false simulation_default_for_destructive: true diff --git a/plugins/gate_prove/hook.py b/plugins/gate_prove/hook.py index 6101cabbf..7a4a8da4f 100644 --- a/plugins/gate_prove/hook.py +++ b/plugins/gate_prove/hook.py @@ -13,6 +13,7 @@ async def enable(services): service = GateProveService( prove_token=os.environ.get('CALDERA_PROVE_TOKEN', ''), authorization_key=os.environ.get('CALDERA_AUTHORIZATION_KEY', ''), + attestation_key=os.environ.get('CALDERA_ATTESTATION_KEY', ''), ledger_path=os.environ.get('CALDERA_GATE_PROVE_LEDGER', 'artifacts/caldera_action_ledger.jsonl'), ) services['gate_prove_svc'] = service diff --git a/plugins/gate_prove/tests/test_gate_prove.py b/plugins/gate_prove/tests/test_gate_prove.py index d505ce5ea..fc42a4b9f 100644 --- a/plugins/gate_prove/tests/test_gate_prove.py +++ b/plugins/gate_prove/tests/test_gate_prove.py @@ -378,5 +378,113 @@ async def test_cleanup_flag_cannot_bypass_high_blast_policy(self) -> None: self.assertEqual(decision.reason, "authorization_lease_missing") +class TestOperationAttestation(unittest.TestCase): + def setUp(self) -> None: + self.service = GateProveService( + authorization_key="authorization-key", + attestation_key="attestation-key", + ledger=OperationLedger(), + ) + + @staticmethod + def link(link_id: str, cleanup: int = 0) -> SimpleNamespace: + return SimpleNamespace( + id=link_id, + ability=SimpleNamespace( + ability_id=f"ability-{link_id}", + technique_id="T1082", + technique_name="System Information Discovery", + ), + cleanup=cleanup, + command=f"secret-command-{link_id}", + paw="range-host-1", + status=-3, + finish=None, + states={"DISCARD": -2, "SUCCESS": 0, "ERROR": 1, "TIMEOUT": 124}, + ) + + def completed_operation(self) -> SimpleNamespace: + link = self.link("one") + operation = SimpleNamespace( + id="operation-1", + name="Attested range operation", + state="finished", + start=datetime.now(timezone.utc) - timedelta(minutes=1), + finish=datetime.now(timezone.utc), + chain=[link], + ) + self.service.govern_link(operation, link) + link.status = link.states["SUCCESS"] + link.finish = datetime.now(timezone.utc) + return operation + + def test_completed_operation_produces_verifiable_bundle_without_commands(self) -> None: + operation = self.completed_operation() + bundle = self.service.attest_operation( + operation, + detection_results=[ + { + "link_id": "one", + "outcome": "detected", + "telemetry_source": "sysmon", + "detection_id": "sigma-test-1", + "evidence_digest": "evidence-sha256", + } + ], + input_provenance=[{"kind": "stix", "digest": "cti-sha256"}], + ) + + self.assertEqual(bundle["summary"]["disposition"], "completed_verified") + self.assertTrue(self.service.verify_attestation(bundle)) + self.assertNotIn("secret-command", str(bundle)) + self.assertEqual(len(bundle["links"][0]["command_digest"]), 64) + + def test_tampered_attestation_fails_verification(self) -> None: + operation = self.completed_operation() + bundle = self.service.attest_operation( + operation, + detection_results=[ + { + "link_id": "one", + "outcome": "detected", + "telemetry_source": "sysmon", + "detection_id": "sigma-test-1", + "evidence_digest": "evidence-sha256", + } + ], + ) + bundle["operation"]["id"] = "different-operation" + self.assertFalse(self.service.verify_attestation(bundle)) + + def test_detection_gap_is_distinct_from_visibility_gap(self) -> None: + operation = self.completed_operation() + bundle = self.service.attest_operation( + operation, + detection_results=[ + { + "link_id": "one", + "outcome": "detection_gap", + "telemetry_source": "sysmon", + } + ], + ) + self.assertEqual(bundle["summary"]["disposition"], "completed_with_detection_gaps") + self.assertEqual(bundle["summary"]["detections"]["detection_gap"], 1) + + def test_missing_link_detection_keeps_attestation_incomplete(self) -> None: + operation = self.completed_operation() + bundle = self.service.attest_operation(operation) + self.assertEqual(bundle["summary"]["disposition"], "evidence_incomplete") + self.assertEqual(bundle["summary"]["missing_detection_links"], ["one"]) + + def test_detection_cannot_reference_unknown_link(self) -> None: + operation = self.completed_operation() + with self.assertRaisesRegex(ValueError, "unknown link"): + self.service.attest_operation( + operation, + detection_results=[{"link_id": "unknown", "outcome": "visibility_gap"}], + ) + + if __name__ == "__main__": unittest.main() From a3c3abdf083963316dfa59bd311edd986b0db8cd Mon Sep 17 00:00:00 2001 From: aah20 Date: Sat, 22 Aug 2026 13:59:28 +0300 Subject: [PATCH 5/5] feat(gate-prove): bind agent intents to dispatch --- plugins/gate_prove/README.md | 9 + plugins/gate_prove/app/gate_prove_svc.py | 83 +++++++- plugins/gate_prove/app/trust_contract.py | 182 ++++++++++++++++++ .../schemas/ability-manifest.schema.json | 47 +++++ .../schemas/operation-intent.schema.json | 63 ++++++ plugins/gate_prove/tests/test_gate_prove.py | 86 +++++++++ 6 files changed, 469 insertions(+), 1 deletion(-) create mode 100644 plugins/gate_prove/app/trust_contract.py create mode 100644 plugins/gate_prove/schemas/ability-manifest.schema.json create mode 100644 plugins/gate_prove/schemas/operation-intent.schema.json diff --git a/plugins/gate_prove/README.md b/plugins/gate_prove/README.md index ba3e342a0..6b599a992 100644 --- a/plugins/gate_prove/README.md +++ b/plugins/gate_prove/README.md @@ -15,6 +15,7 @@ When running automated adversary emulation against enterprise or staging infrast 6. **Generated Ability Manifest**: AI-generated abilities fail closed unless they include parsers, cleanup, bounded scope, generator/model provenance, and a matching digest over execution-relevant fields. 7. **Central Dispatch Enforcement**: `Operation.apply()` evaluates every planner, REST, scheduled, and direct link through GateProve. Denied or simulation-only links are retained as discarded audit records but never become executable agent instructions. 8. **Operation Attestation**: Completed operations can emit an integrity-protected evidence bundle containing input provenance, target and command digests, gate decisions, cleanup state, detection outcomes, and the ledger root without disclosing command contents. +9. **Operation Trust Contract**: A versioned, digest-protected operation intent binds an autonomous planner's inputs and objective to the exact ability manifests, range targets, privilege ceiling, and expiration accepted for dispatch. ## AI-generated ability contract @@ -22,6 +23,14 @@ Call `evaluate_ability` with both `ability_manifest` and `provenance` before dis Required scope fields are `targets`, `expires_at`, and a positive `max_executions`. Required provenance fields are `generator`, `model`, `created_at`, and `content_hash`; compute the latter with `ability_content_hash` from `app/ability_manifest.py`. +Machine-readable contracts are published in `schemas/ability-manifest.schema.json` and `schemas/operation-intent.schema.json`. + +## MITRE MCP trust contract + +An autonomous planner must submit an `OperationIntent` through `register_operation_intent()` before dispatch. GateProve validates and records the intent, then revalidates it at every link dispatch. Each link must target an asset named by the intent and carry an `AbilityManifest` whose deterministic digest exactly matches the intent's ability reference. Expired, modified, unlisted, or out-of-range work fails closed and receives a hash-chained denial receipt. + +The boundary is intentionally transport-neutral: an MCP server, REST client, or CALDERA plugin can build the same version `1.0.0` contract with `OperationIntentValidator.build()`. This keeps model planning outside the trusted computing base; only deterministic schema, digest, scope, and policy checks authorize execution. After CALDERA finishes, `attest_operation()` provides the evidence-bearing terminal result that an MCP workflow can return to its caller. + ## Configuration In `conf/default.yml`: diff --git a/plugins/gate_prove/app/gate_prove_svc.py b/plugins/gate_prove/app/gate_prove_svc.py index 4c0801126..eb644c0e3 100644 --- a/plugins/gate_prove/app/gate_prove_svc.py +++ b/plugins/gate_prove/app/gate_prove_svc.py @@ -4,7 +4,7 @@ from pathlib import Path from typing import Any, Dict, Optional -from plugins.gate_prove.app.ability_manifest import AbilityManifestValidator +from plugins.gate_prove.app.ability_manifest import AbilityManifestValidator, ability_content_hash from plugins.gate_prove.app.authorization_lease import AuthorizationLeaseIssuer, command_digest from plugins.gate_prove.app.attestation import OperationAttestor from plugins.gate_prove.app.ledger import OperationLedger @@ -13,6 +13,7 @@ AbilityDecision, GateDisposition, ) +from plugins.gate_prove.app.trust_contract import IntentValidation, OperationIntentValidator class GateProveService: @@ -34,6 +35,7 @@ def __init__( self.lease_issuer = AuthorizationLeaseIssuer(lease_key) evidence_key = attestation_key or os.environ.get("CALDERA_ATTESTATION_KEY", "") self.attestor = OperationAttestor(evidence_key, self.ledger) + self.intent_validator = OperationIntentValidator() def is_kill_switch_engaged(self) -> bool: flag = os.environ.get("CALDERA_KILL_SWITCH", "").strip().lower() @@ -229,6 +231,30 @@ def attest_operation(self, operation: Any, **evidence: Any) -> dict[str, Any]: def verify_attestation(self, bundle: dict[str, Any]) -> bool: return self.attestor.verify(bundle) + def register_operation_intent( + self, operation: Any, intent: dict[str, Any] + ) -> IntentValidation: + """Validate and bind an AI planner's immutable intent to an operation.""" + validation = self.intent_validator.validate(intent, expected_operation_id=str(operation.id)) + disposition = "allow" if validation.valid else "deny" + reason = "operation_intent_registered" if validation.valid else "operation_intent_invalid" + self.ledger.record( + str(operation.id), + "operation-intent", + "trust-contract", + disposition, + validation.valid, + reason, + metadata={ + "intent_id": str(intent.get("intent_id", "")), + "intent_digest": validation.intent_digest, + "validation_errors": list(validation.errors), + }, + ) + if validation.valid: + operation.gate_prove_intent = dict(intent) + return validation + def _lease_execution_count(self, lease_id: str) -> int: if not lease_id: return 0 @@ -244,6 +270,34 @@ def evaluate_link(self, operation: Any, link: Any) -> AbilityDecision: ability = link.ability digest = command_digest(getattr(link, "command", "")) target = str(getattr(link, "paw", "") or getattr(link, "host", "")) + intent = getattr(operation, "gate_prove_intent", None) + if intent: + binding_errors = list( + self.intent_validator.validate( + intent, expected_operation_id=str(operation.id) + ).errors + ) + references = { + item.get("ability_id"): item.get("manifest_digest") + for item in intent.get("abilities", []) + if isinstance(item, dict) + } + manifest = getattr(ability, "gate_prove_manifest", None) + expected_manifest_digest = references.get(ability.ability_id) + if not expected_manifest_digest: + binding_errors.append("ability is not authorized by the operation intent") + elif not manifest: + binding_errors.append("ability manifest is missing for an intent-bound operation") + elif ability_content_hash(manifest) != expected_manifest_digest: + binding_errors.append("ability manifest does not match the operation intent") + elif str(manifest.get("privilege", "")) != str( + intent.get("range", {}).get("privilege_ceiling", "") + ): + binding_errors.append("ability privilege does not match the operation intent ceiling") + if target not in intent.get("range", {}).get("targets", []): + binding_errors.append("target is outside the operation intent range") + if binding_errors: + return self._deny_intent_binding(operation, ability, binding_errors) return self.evaluate_ability( operation_id=operation.id, ability_id=ability.ability_id, @@ -260,6 +314,33 @@ def evaluate_link(self, operation: Any, link: Any) -> AbilityDecision: target=target, ) + def _deny_intent_binding( + self, operation: Any, ability: Any, errors: list[str] + ) -> AbilityDecision: + reason = "operation_intent_binding_failed:" + ";".join(errors) + ledger_id, receipt_hash = self.ledger.record( + str(operation.id), + str(ability.ability_id), + str(ability.technique_id), + "deny", + False, + reason, + metadata={"binding_errors": errors}, + ) + return AbilityDecision( + operation_id=str(operation.id), + ability_id=str(ability.ability_id), + technique_id=str(ability.technique_id), + technique_name=str(ability.technique_name), + disposition="deny", + allowed=False, + requires_hitl=True, + never_equate_intent_to_approval=True, + reason=reason, + ledger_id=ledger_id, + receipt_hash=receipt_hash, + ) + def govern_link(self, operation: Any, link: Any) -> AbilityDecision: """Apply a gate decision to a link before Caldera queues it for an agent.""" decision = self.evaluate_link(operation, link) diff --git a/plugins/gate_prove/app/trust_contract.py b/plugins/gate_prove/app/trust_contract.py new file mode 100644 index 000000000..14dfd162f --- /dev/null +++ b/plugins/gate_prove/app/trust_contract.py @@ -0,0 +1,182 @@ +from __future__ import annotations + +import hashlib +import json +import uuid +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from typing import Any, Iterable, Mapping + + +TRUST_CONTRACT_VERSION = "1.0.0" + + +def operation_intent_digest(intent: Mapping[str, Any]) -> str: + canonical = {key: value for key, value in intent.items() if key != "intent_digest"} + raw = json.dumps(canonical, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + return hashlib.sha256(raw.encode("utf-8")).hexdigest() + + +@dataclass(frozen=True) +class IntentValidation: + valid: bool + errors: tuple[str, ...] = field(default_factory=tuple) + intent_digest: str = "" + + +class OperationIntentValidator: + """Validate the versioned planning contract submitted before execution.""" + + def validate( + self, intent: Mapping[str, Any], *, expected_operation_id: str = "" + ) -> IntentValidation: + errors: list[str] = [] + required = ( + "schema_version", + "intent_id", + "operation_id", + "objective", + "created_at", + "expires_at", + "planner", + "input_provenance", + "range", + "abilities", + "intent_digest", + ) + self._require_nonempty(intent, required, "intent", errors) + if intent.get("schema_version") != TRUST_CONTRACT_VERSION: + errors.append(f"intent.schema_version must be {TRUST_CONTRACT_VERSION}") + if expected_operation_id and intent.get("operation_id") != expected_operation_id: + errors.append("intent.operation_id does not match the operation") + + planner = intent.get("planner") + if isinstance(planner, Mapping): + self._require_nonempty(planner, ("name", "version", "model"), "planner", errors) + elif planner is not None: + errors.append("intent.planner must be an object") + + range_scope = intent.get("range") + if isinstance(range_scope, Mapping): + self._require_nonempty( + range_scope, ("spec_digest", "targets", "privilege_ceiling"), "range", errors + ) + self._validate_digest(range_scope.get("spec_digest"), "range.spec_digest", errors) + targets = range_scope.get("targets") + if targets and ( + not isinstance(targets, list) + or not all(isinstance(target, str) and target for target in targets) + ): + errors.append("range.targets must be a list of target identifiers") + elif range_scope is not None: + errors.append("intent.range must be an object") + + abilities = intent.get("abilities") + if isinstance(abilities, list) and abilities: + seen: set[str] = set() + for index, ability in enumerate(abilities): + if not isinstance(ability, Mapping): + errors.append(f"abilities[{index}] must be an object") + continue + self._require_nonempty( + ability, ("ability_id", "manifest_digest"), f"abilities[{index}]", errors + ) + ability_id = str(ability.get("ability_id", "")) + if ability_id in seen: + errors.append(f"abilities contains duplicate ability_id: {ability_id}") + seen.add(ability_id) + self._validate_digest( + ability.get("manifest_digest"), f"abilities[{index}].manifest_digest", errors + ) + elif abilities is not None: + errors.append("intent.abilities must be a non-empty list") + + provenance = intent.get("input_provenance") + if isinstance(provenance, list) and provenance: + for index, item in enumerate(provenance): + if not isinstance(item, Mapping): + errors.append(f"input_provenance[{index}] must be an object") + continue + self._require_nonempty( + item, ("kind", "digest"), f"input_provenance[{index}]", errors + ) + self._validate_digest(item.get("digest"), f"input_provenance[{index}].digest", errors) + elif provenance is not None: + errors.append("intent.input_provenance must be a non-empty list") + + self._validate_expiry(intent.get("expires_at"), errors) + digest = operation_intent_digest(intent) + claimed_digest = intent.get("intent_digest") + if claimed_digest and claimed_digest != digest: + errors.append("intent.intent_digest does not match the operation intent") + return IntentValidation(not errors, tuple(errors), digest) + + @staticmethod + def build( + *, + operation_id: str, + objective: str, + planner: Mapping[str, str], + input_provenance: Iterable[Mapping[str, str]], + range_spec_digest: str, + targets: Iterable[str], + privilege_ceiling: str, + abilities: Iterable[Mapping[str, str]], + ttl_seconds: int = 3600, + now: datetime | None = None, + ) -> dict[str, Any]: + if ttl_seconds < 1: + raise ValueError("ttl_seconds must be positive") + created_at = now or datetime.now(timezone.utc) + if created_at.tzinfo is None: + raise ValueError("now must include a timezone") + intent: dict[str, Any] = { + "schema_version": TRUST_CONTRACT_VERSION, + "intent_id": str(uuid.uuid4()), + "operation_id": operation_id, + "objective": objective, + "created_at": created_at.isoformat(), + "expires_at": (created_at + timedelta(seconds=ttl_seconds)).isoformat(), + "planner": dict(planner), + "input_provenance": [dict(item) for item in input_provenance], + "range": { + "spec_digest": range_spec_digest, + "targets": list(targets), + "privilege_ceiling": privilege_ceiling, + }, + "abilities": [dict(item) for item in abilities], + } + intent["intent_digest"] = operation_intent_digest(intent) + return intent + + @staticmethod + def _validate_expiry(value: Any, errors: list[str]) -> None: + if not value: + return + try: + expiry = datetime.fromisoformat(str(value).replace("Z", "+00:00")) + except ValueError: + errors.append("intent.expires_at must be an ISO-8601 timestamp") + return + if expiry.tzinfo is None: + errors.append("intent.expires_at must include a timezone") + elif expiry <= datetime.now(timezone.utc): + errors.append("intent.expires_at must be in the future") + + @staticmethod + def _validate_digest(value: Any, field_name: str, errors: list[str]) -> None: + if value and ( + not isinstance(value, str) + or len(value) != 64 + or any(character not in "0123456789abcdef" for character in value) + ): + errors.append(f"{field_name} must be a lowercase SHA-256 digest") + + @staticmethod + def _require_nonempty( + value: Mapping[str, Any], fields: tuple[str, ...], prefix: str, errors: list[str] + ) -> None: + for field_name in fields: + item = value.get(field_name) + if item is None or item == "" or item == [] or item == {}: + errors.append(f"{prefix}.{field_name} is required") diff --git a/plugins/gate_prove/schemas/ability-manifest.schema.json b/plugins/gate_prove/schemas/ability-manifest.schema.json new file mode 100644 index 000000000..5adcad06d --- /dev/null +++ b/plugins/gate_prove/schemas/ability-manifest.schema.json @@ -0,0 +1,47 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://caldera.apache.org/gate-prove/ability-manifest.schema.json", + "title": "GateProve AbilityManifest", + "description": "Safety-relevant execution contract for an AI-generated CALDERA ability.", + "type": "object", + "required": [ + "ability_id", + "name", + "description", + "tactic", + "technique_id", + "technique_name", + "platforms", + "parsers", + "cleanup", + "scope" + ], + "properties": { + "ability_id": { "type": "string", "minLength": 1 }, + "name": { "type": "string", "minLength": 1 }, + "description": { "type": "string", "minLength": 1 }, + "tactic": { "type": "string", "minLength": 1 }, + "technique_id": { "type": "string", "pattern": "^T[0-9]{4}(\\.[0-9]{3})?$" }, + "technique_name": { "type": "string", "minLength": 1 }, + "platforms": { "type": "object", "minProperties": 1 }, + "privilege": { "type": "string" }, + "parsers": { "type": "array", "minItems": 1 }, + "cleanup": { "type": "array", "minItems": 1 }, + "scope": { + "type": "object", + "required": ["targets", "expires_at", "max_executions"], + "properties": { + "targets": { + "type": "array", + "minItems": 1, + "uniqueItems": true, + "items": { "type": "string", "minLength": 1 } + }, + "expires_at": { "type": "string", "format": "date-time" }, + "max_executions": { "type": "integer", "minimum": 1 } + }, + "additionalProperties": true + } + }, + "additionalProperties": true +} diff --git a/plugins/gate_prove/schemas/operation-intent.schema.json b/plugins/gate_prove/schemas/operation-intent.schema.json new file mode 100644 index 000000000..fda8776c6 --- /dev/null +++ b/plugins/gate_prove/schemas/operation-intent.schema.json @@ -0,0 +1,63 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://caldera.apache.org/gate-prove/operation-intent.schema.json", + "title": "GateProve Operation Intent", + "type": "object", + "additionalProperties": false, + "required": ["schema_version", "intent_id", "operation_id", "objective", "created_at", "expires_at", "planner", "input_provenance", "range", "abilities", "intent_digest"], + "properties": { + "schema_version": {"const": "1.0.0"}, + "intent_id": {"type": "string", "minLength": 1}, + "operation_id": {"type": "string", "minLength": 1}, + "objective": {"type": "string", "minLength": 1}, + "created_at": {"type": "string", "format": "date-time"}, + "expires_at": {"type": "string", "format": "date-time"}, + "intent_digest": {"type": "string", "pattern": "^[0-9a-f]{64}$"}, + "planner": { + "type": "object", + "additionalProperties": false, + "required": ["name", "version", "model"], + "properties": { + "name": {"type": "string", "minLength": 1}, + "version": {"type": "string", "minLength": 1}, + "model": {"type": "string", "minLength": 1} + } + }, + "input_provenance": { + "type": "array", + "minItems": 1, + "items": { + "type": "object", + "additionalProperties": true, + "required": ["kind", "digest"], + "properties": { + "kind": {"type": "string", "minLength": 1}, + "digest": {"type": "string", "pattern": "^[0-9a-f]{64}$"} + } + } + }, + "range": { + "type": "object", + "additionalProperties": false, + "required": ["spec_digest", "targets", "privilege_ceiling"], + "properties": { + "spec_digest": {"type": "string", "pattern": "^[0-9a-f]{64}$"}, + "targets": {"type": "array", "minItems": 1, "items": {"type": "string", "minLength": 1}}, + "privilege_ceiling": {"type": "string", "minLength": 1} + } + }, + "abilities": { + "type": "array", + "minItems": 1, + "items": { + "type": "object", + "additionalProperties": false, + "required": ["ability_id", "manifest_digest"], + "properties": { + "ability_id": {"type": "string", "minLength": 1}, + "manifest_digest": {"type": "string", "pattern": "^[0-9a-f]{64}$"} + } + } + } + } +} diff --git a/plugins/gate_prove/tests/test_gate_prove.py b/plugins/gate_prove/tests/test_gate_prove.py index fc42a4b9f..30e29be5a 100644 --- a/plugins/gate_prove/tests/test_gate_prove.py +++ b/plugins/gate_prove/tests/test_gate_prove.py @@ -18,6 +18,7 @@ from plugins.gate_prove.app.authorization_lease import AuthorizationLeaseIssuer, command_digest from plugins.gate_prove.app.ledger import OperationLedger from plugins.gate_prove.hook import enable +from plugins.gate_prove.app.trust_contract import OperationIntentValidator class TestGateProve(unittest.TestCase): @@ -301,6 +302,91 @@ def test_tampered_lease_fails_closed(self) -> None: self.assertFalse(result.valid) +class TestOperationIntentContract(unittest.TestCase): + def setUp(self) -> None: + self.service = GateProveService(authorization_key="contract-key", ledger=OperationLedger()) + self.operation = SimpleNamespace(id="operation-contract-1") + self.manifest = TestGateProve.generated_ability() + self.manifest_digest = ability_content_hash(self.manifest) + + def intent(self, targets: list[str] | None = None, now: datetime | None = None) -> dict: + digest = "a" * 64 + return OperationIntentValidator.build( + operation_id=self.operation.id, + objective="Validate identity-to-container detection coverage", + planner={"name": "mitre-mcp", "version": "1.0", "model": "test-model"}, + input_provenance=[{"kind": "stix", "digest": digest}], + range_spec_digest="b" * 64, + targets=targets or ["range-host-1"], + privilege_ceiling="User", + abilities=[ + { + "ability_id": self.manifest["ability_id"], + "manifest_digest": self.manifest_digest, + } + ], + ttl_seconds=60, + now=now, + ) + + def link(self, target: str = "range-host-1") -> SimpleNamespace: + provenance = { + "generator": "mitre-mcp", + "model": "test-model", + "created_at": datetime.now(timezone.utc).isoformat(), + "content_hash": self.manifest_digest, + } + return SimpleNamespace( + id="intent-link-1", + ability=SimpleNamespace( + ability_id=self.manifest["ability_id"], + technique_id=self.manifest["technique_id"], + technique_name=self.manifest["technique_name"], + gate_prove_manifest=self.manifest, + gate_prove_provenance=provenance, + ), + cleanup=0, + command="read-range-canary", + paw=target, + status=-3, + states={"DISCARD": -2}, + ) + + def test_valid_intent_binds_exact_manifest_and_target(self) -> None: + validation = self.service.register_operation_intent(self.operation, self.intent()) + link = self.link() + decision = self.service.govern_link(self.operation, link) + self.assertTrue(validation.valid) + self.assertEqual(decision.disposition, "allow") + + def test_intent_rejects_out_of_range_target_at_dispatch(self) -> None: + self.service.register_operation_intent(self.operation, self.intent()) + link = self.link(target="range-host-2") + decision = self.service.govern_link(self.operation, link) + self.assertEqual(decision.disposition, "deny") + self.assertIn("target is outside", decision.reason) + self.assertEqual(link.status, link.states["DISCARD"]) + + def test_intent_rejects_manifest_above_declared_privilege_ceiling(self) -> None: + self.manifest["privilege"] = "Elevated" + self.manifest_digest = ability_content_hash(self.manifest) + intent = self.intent() + validation = self.service.register_operation_intent(self.operation, intent) + decision = self.service.govern_link(self.operation, self.link()) + self.assertTrue(validation.valid) + self.assertEqual(decision.disposition, "deny") + self.assertIn("privilege", decision.reason) + + def test_expired_intent_is_not_registered(self) -> None: + past = datetime.now(timezone.utc) - timedelta(hours=1) + validation = self.service.register_operation_intent( + self.operation, self.intent(now=past) + ) + self.assertFalse(validation.valid) + self.assertFalse(hasattr(self.operation, "gate_prove_intent")) + self.assertIn("intent.expires_at must be in the future", validation.errors) + + class TestPluginHook(unittest.IsolatedAsyncioTestCase): async def test_enable_registers_configured_service(self) -> None: with tempfile.TemporaryDirectory() as tmp: