feat(plugins): add GateProve safety boundary & hash-chained action ledger plugin

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']

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)
 

plugins/gate_prove/README.md

@@ -0,0 +1,63 @@
+# 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. **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.
+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
+
+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`.
+
+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`:
+
+```yaml
+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.
+
+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
+
+```bash
+python3 -m unittest plugins/gate_prove/tests/test_gate_prove.py
+```
+
+## License
+
+Apache-2.0

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")