Parity close

README.md

@@ -205,6 +205,11 @@ predicate-authority revoke intent --host 127.0.0.1 --port 8787 --hash <intent_ha
 predicate-authorityd --host 127.0.0.1 --port 8787 --mode local_only --policy-file examples/authorityd/policy.json
 ```
 
+Mandate cache behavior:
+
+- default is ephemeral in-memory mandate/revocation cache,
+- set `--mandate-store-file <path>` to enable optional local persistence and restart recovery.
+
 ### Identity mode options (`predicate-authorityd`)
 
 - `--identity-mode local`: deterministic local bridge (default).

docs/authorityd-operations.md

@@ -43,6 +43,23 @@ PYTHONPATH=. predicate-authorityd \
   --credential-store-file ./.predicate-authorityd/credentials.json
 ```
 
+By design, mandate/revocation cache is in-memory (ephemeral) unless you explicitly
+enable persistence with `--mandate-store-file`.
+
+### Optional: enable persisted mandate/revocation cache (parity extension)
+
+Use this only when restart-recovery for local revocations/mandate lineage is required.
+If omitted, default behavior remains ephemeral.
+
+```bash
+PYTHONPATH=. predicate-authorityd \
+  --host 127.0.0.1 \
+  --port 8787 \
+  --mode local_only \
+  --policy-file examples/authorityd/policy.json \
+  --mandate-store-file ./.predicate-authorityd/mandates.json
+```
+
 ### Optional: enable control-plane shipping
 
 To automatically ship proof events and usage records to
@@ -506,6 +523,7 @@ Example response:
 {
   "mode": "local_only",
   "policy_hot_reload_enabled": true,
+  "mandate_store_persistence_enabled": false,
   "revoked_principal_count": 0,
   "revoked_intent_count": 0,
   "revoked_mandate_count": 0,

predicate_authority/client.py

@@ -113,12 +113,15 @@ def authorize(
                 reason=AuthorizationReason.INVALID_MANDATE,
                 violated_rule="revocation_cache",
             )
+        if decision.allowed and decision.mandate is not None:
+            self._revocation_cache.register_mandate(decision.mandate)
         return decision
 
     def verify_token(self, token: str) -> SignedMandate | None:
         mandate = self._mandate_signer.verify(token)
         if mandate is None:
             return None
+        self._revocation_cache.register_mandate(mandate)
         if self._revocation_cache.is_mandate_revoked(mandate):
             return None
         return mandate
@@ -142,5 +145,5 @@ def verify_delegation_chain(
     def revoke_principal(self, principal_id: str) -> None:
         self._revocation_cache.revoke_principal(principal_id)
 
-    def revoke_mandate(self, mandate_id: str) -> None:
-        self._revocation_cache.revoke_mandate_id(mandate_id)
+    def revoke_mandate(self, mandate_id: str, cascade: bool = False) -> int:
+        return self._revocation_cache.revoke_mandate_id(mandate_id, cascade=cascade)

predicate_authority/control_plane.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import hashlib
+import hmac
 import http.client
 import json
 import time
@@ -27,6 +28,7 @@ class ControlPlaneClientConfig:
     sync_poll_interval_ms: int = 200
     sync_project_id: str | None = None
     sync_environment: str | None = None
+    replay_signing_secret: str | None = None
 
 
 @dataclass(frozen=True)
@@ -212,10 +214,11 @@ def poll_authority_updates(
         return AuthoritySyncSnapshot.from_payload(payload)
 
     def _post_json(self, path: str, payload: Mapping[str, object]) -> bool:
+        replay_headers = self._build_replay_headers(path)
         attempts = self.config.max_retries + 1
         for attempt in range(attempts):
             try:
-                self._post_json_once(path, payload)
+                self._post_json_once(path, payload, replay_headers=replay_headers)
                 return True
             except Exception as exc:
                 is_last_attempt = attempt == attempts - 1
@@ -240,12 +243,19 @@ def _get_json(self, path: str) -> Mapping[str, object]:
                 time.sleep(self.config.backoff_initial_s * (2**attempt))
         return {}
 
-    def _post_json_once(self, path: str, payload: Mapping[str, object]) -> None:
+    def _post_json_once(
+        self,
+        path: str,
+        payload: Mapping[str, object],
+        *,
+        replay_headers: Mapping[str, str],
+    ) -> None:
         target_path = path if path.startswith("/") else f"/{path}"
         connection = self._new_connection()
         headers = {"Content-Type": "application/json"}
         if self.config.auth_token:
             headers["Authorization"] = f"Bearer {self.config.auth_token}"
+        headers.update(replay_headers)
         body = json.dumps(payload)
         try:
             connection.request("POST", target_path, body=body, headers=headers)
@@ -280,6 +290,26 @@ def _new_connection(self) -> http.client.HTTPConnection:
             return http.client.HTTPSConnection(self._base.netloc, timeout=self.config.timeout_s)
         return http.client.HTTPConnection(self._base.netloc, timeout=self.config.timeout_s)
 
+    def _build_replay_headers(self, path: str) -> dict[str, str]:
+        timestamp = str(int(time.time()))
+        nonce = hashlib.sha256(
+            f"{self.config.tenant_id}|{path}|{time.time_ns()}".encode()
+        ).hexdigest()[:32]
+        headers = {
+            "X-PA-Nonce": nonce,
+            "X-PA-Timestamp": timestamp,
+            "X-PA-Idempotency-Token": hashlib.sha256(
+                f"{nonce}|{timestamp}|{path}".encode()
+            ).hexdigest()[:32],
+        }
+        if self.config.replay_signing_secret is not None:
+            message = f"{nonce}:{timestamp}:POST:{path}".encode()
+            signature = hmac.new(
+                self.config.replay_signing_secret.encode("utf-8"), message, hashlib.sha256
+            ).hexdigest()
+            headers["X-PA-Signature"] = signature
+        return headers
+
 
 @dataclass
 class ControlPlaneTraceEmitter:

predicate_authority/daemon.py

@@ -370,11 +370,23 @@ def _handle_revoke_intent(self) -> None:
     def _handle_revoke_mandate(self) -> None:
         payload = self._read_json_body()
         mandate_id = payload.get("mandate_id")
+        cascade_raw = payload.get("cascade")
+        cascade = bool(cascade_raw) if isinstance(cascade_raw, bool) else False
         if not isinstance(mandate_id, str) or mandate_id.strip() == "":
             self._send_json(400, {"error": "mandate_id is required"})
             return
-        self.server.daemon_ref.revoke_mandate(mandate_id.strip())  # type: ignore[attr-defined]
-        self._send_json(200, {"ok": True, "mandate_id": mandate_id.strip()})
+        revoked_count = self.server.daemon_ref.revoke_mandate(  # type: ignore[attr-defined]
+            mandate_id.strip(), cascade=cascade
+        )
+        self._send_json(
+            200,
+            {
+                "ok": True,
+                "mandate_id": mandate_id.strip(),
+                "cascade": cascade,
+                "revoked_count": int(revoked_count),
+            },
+        )
 
     def _handle_identity_task(self) -> None:
         payload = self._read_json_body()
@@ -585,8 +597,8 @@ def revoke_principal(self, principal_id: str) -> None:
     def revoke_intent(self, intent_hash: str) -> None:
         self._sidecar.revoke_intent_hash(intent_hash)
 
-    def revoke_mandate(self, mandate_id: str) -> None:
-        self._sidecar.revoke_mandate_id(mandate_id)
+    def revoke_mandate(self, mandate_id: str, cascade: bool = False) -> int:
+        return self._sidecar.revoke_mandate_id(mandate_id, cascade=cascade)
 
     def max_request_body_bytes(self) -> int:
         return max(0, int(self._config.max_request_body_bytes))
@@ -826,15 +838,16 @@ def _apply_sync_snapshot(self, snapshot: AuthoritySyncSnapshot) -> None:
             elif item.type == "intent" and item.intent_hash is not None:
                 self._sidecar.revoke_intent_hash(item.intent_hash)
             elif item.type == "tags":
-                # Tag revocation support is modeled in control-plane API but not yet represented in
-                # sidecar's revocation cache keys.
-                continue
+                tags = {tag.strip().lower() for tag in item.tags if tag.strip() != ""}
+                if "global_kill_switch" in tags:
+                    self._sidecar.activate_global_kill_switch()
 
 
 def _build_default_sidecar(
     mode: AuthorityMode,
     policy_file: str | None,
     credential_store_file: str,
+    mandate_store_file: str | None = None,
     control_plane_config: ControlPlaneBootstrapConfig | None = None,
     local_identity_config: LocalIdentityBootstrapConfig | None = None,
     identity_bridge: ExchangeTokenBridge | None = None,
@@ -912,7 +925,7 @@ def _build_default_sidecar(
         proof_ledger=proof_ledger,
         identity_bridge=identity_bridge or IdentityBridge(),
         credential_store=LocalCredentialStore(credential_store_file),
-        revocation_cache=LocalRevocationCache(),
+        revocation_cache=LocalRevocationCache(store_file_path=mandate_store_file),
         policy_engine=policy_engine,
         local_identity_registry=local_identity_registry,
     )
@@ -1070,6 +1083,14 @@ def main() -> None:
         "--credential-store-file",
         default=str(Path.home() / ".predicate-authorityd" / "credentials.json"),
     )
+    parser.add_argument(
+        "--mandate-store-file",
+        default=None,
+        help=(
+            "Optional path for persisted local revocation/mandate cache. "
+            "If omitted, mandate cache remains in-memory (ephemeral default)."
+        ),
+    )
     parser.add_argument(
         "--local-identity-enabled",
         action="store_true",
@@ -1307,6 +1328,7 @@ def main() -> None:
         mode=mode,
         policy_file=args.policy_file,
         credential_store_file=args.credential_store_file,
+        mandate_store_file=args.mandate_store_file,
         control_plane_config=control_plane_bootstrap,
         local_identity_config=local_identity_bootstrap,
         identity_bridge=identity_bridge,