address code review

Author: rcholic

predicate_authority/client.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import os
 from dataclasses import dataclass
 
 from predicate_authority.guard import ActionGuard
@@ -63,6 +64,27 @@ def from_policy_file(
             policy_file=policy_file,
         )
 
+    @classmethod
+    def from_env(cls) -> LocalAuthorizationContext:
+        policy_file = os.getenv("PREDICATE_AUTHORITY_POLICY_FILE")
+        secret_key = os.getenv("PREDICATE_AUTHORITY_SIGNING_KEY")
+        ttl_seconds_raw = os.getenv("PREDICATE_AUTHORITY_MANDATE_TTL_SECONDS", "300")
+        if policy_file is None or policy_file.strip() == "":
+            raise RuntimeError("PREDICATE_AUTHORITY_POLICY_FILE is required.")
+        if secret_key is None or secret_key.strip() == "":
+            raise RuntimeError("PREDICATE_AUTHORITY_SIGNING_KEY is required.")
+        try:
+            ttl_seconds = int(ttl_seconds_raw)
+        except ValueError as exc:
+            raise RuntimeError(
+                "PREDICATE_AUTHORITY_MANDATE_TTL_SECONDS must be an integer."
+            ) from exc
+        return cls.from_policy_file(
+            policy_file=policy_file,
+            secret_key=secret_key,
+            ttl_seconds=ttl_seconds,
+        )
+
     def authorize(
         self,
         request: ActionRequest,

predicate_authority/control_plane.py

@@ -9,7 +9,7 @@
 from datetime import datetime, timezone
 from urllib.parse import urlsplit
 
-from predicate_contracts import ProofEvent, TraceEmitter
+from predicate_contracts import ProofEvent
 
 
 @dataclass(frozen=True)
@@ -138,7 +138,7 @@ def _new_connection(self) -> http.client.HTTPConnection:
 
 
 @dataclass
-class ControlPlaneTraceEmitter(TraceEmitter):
+class ControlPlaneTraceEmitter:
     client: ControlPlaneClient
     trace_id: str | None = None
     emit_usage_credits: bool = True

predicate_authority/daemon.py

@@ -53,6 +53,7 @@ class DaemonConfig:
     host: str = "127.0.0.1"
     port: int = 8787
     policy_poll_interval_s: float = 2.0
+    max_request_body_bytes: int = 1_048_576
 
 
 @dataclass(frozen=True)
@@ -167,6 +168,7 @@ def do_POST(self) -> None:  # noqa: N802
             "/policy/reload": self._handle_policy_reload,
             "/revoke/principal": self._handle_revoke_principal,
             "/revoke/intent": self._handle_revoke_intent,
+            "/revoke/mandate": self._handle_revoke_mandate,
             "/identity/task": self._handle_identity_task,
             "/identity/revoke": self._handle_identity_revoke,
             "/ledger/flush-ack": self._handle_ledger_flush_ack,
@@ -201,6 +203,15 @@ def _handle_revoke_intent(self) -> None:
         self.server.daemon_ref.revoke_intent(intent_hash.strip())  # type: ignore[attr-defined]
         self._send_json(200, {"ok": True, "intent_hash": intent_hash.strip()})
 
+    def _handle_revoke_mandate(self) -> None:
+        payload = self._read_json_body()
+        mandate_id = payload.get("mandate_id")
+        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()})
+
     def _handle_identity_task(self) -> None:
         payload = self._read_json_body()
         principal_id = payload.get("principal_id")
@@ -273,6 +284,9 @@ def _read_json_body(self) -> dict[str, Any]:
             return {}
         if content_length <= 0:
             return {}
+        max_body = self.server.daemon_ref.max_request_body_bytes()  # type: ignore[attr-defined]
+        if content_length > max_body:
+            return {}
         payload = self.rfile.read(content_length).decode("utf-8")
         try:
             loaded = json.loads(payload)
@@ -388,6 +402,12 @@ 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 max_request_body_bytes(self) -> int:
+        return max(0, int(self._config.max_request_body_bytes))
+
     def issue_task_identity(
         self,
         principal_id: str,
@@ -592,6 +612,7 @@ def _build_default_sidecar(
     control_plane_config: ControlPlaneBootstrapConfig | None = None,
     local_identity_config: LocalIdentityBootstrapConfig | None = None,
     identity_bridge: ExchangeTokenBridge | None = None,
+    mandate_signing_key: str | None = None,
 ) -> PredicateAuthoritySidecar:
     policy_rules: tuple[PolicyRule, ...] = ()
     global_max_delegation_depth: int | None = None
@@ -647,7 +668,7 @@ def _build_default_sidecar(
 
     guard = ActionGuard(
         policy_engine=policy_engine,
-        mandate_signer=LocalMandateSigner(secret_key=secrets.token_hex(32)),
+        mandate_signer=LocalMandateSigner(secret_key=mandate_signing_key or secrets.token_hex(32)),
         proof_ledger=proof_ledger,
     )
     return PredicateAuthoritySidecar(
@@ -709,6 +730,22 @@ def _build_identity_bridge_from_args(args: argparse.Namespace) -> ExchangeTokenB
     raise SystemExit(f"Unsupported identity mode: {mode}")
 
 
+def _resolve_mandate_signing_key(
+    signing_key_file: str | None,
+    signing_key_env: str,
+) -> str:
+    if signing_key_file is not None and str(signing_key_file).strip() != "":
+        key_path = Path(signing_key_file)
+        if key_path.exists():
+            loaded = key_path.read_text(encoding="utf-8").strip()
+            if loaded != "":
+                return loaded
+    env_value = os.getenv(signing_key_env)
+    if env_value is not None and env_value.strip() != "":
+        return env_value.strip()
+    return secrets.token_hex(32)
+
+
 def main() -> None:
     parser = argparse.ArgumentParser(description="predicate-authorityd sidecar daemon")
     parser.add_argument("--host", default="127.0.0.1")
@@ -816,6 +853,16 @@ def main() -> None:
     )
     parser.set_defaults(control_plane_fail_open=True)
     parser.add_argument("--control-plane-usage-credits-per-decision", type=int, default=1)
+    parser.add_argument(
+        "--mandate-signing-key-env",
+        default="PREDICATE_AUTHORITY_SIGNING_KEY",
+        help="Env var name for mandate signing key.",
+    )
+    parser.add_argument(
+        "--mandate-signing-key-file",
+        default=None,
+        help="Optional file path containing mandate signing key.",
+    )
     args = parser.parse_args()
 
     mode = AuthorityMode(args.mode)
@@ -851,20 +898,26 @@ def main() -> None:
         default_ttl_seconds=max(1, int(args.local_identity_default_ttl_s)),
     )
     identity_bridge = _build_identity_bridge_from_args(args)
+    mandate_signing_key = _resolve_mandate_signing_key(
+        signing_key_file=args.mandate_signing_key_file,
+        signing_key_env=args.mandate_signing_key_env,
+    )
     sidecar = _build_default_sidecar(
         mode=mode,
         policy_file=args.policy_file,
         credential_store_file=args.credential_store_file,
         control_plane_config=control_plane_bootstrap,
         local_identity_config=local_identity_bootstrap,
         identity_bridge=identity_bridge,
+        mandate_signing_key=mandate_signing_key,
     )
     daemon = PredicateAuthorityDaemon(
         sidecar=sidecar,
         config=DaemonConfig(
             host=args.host,
             port=args.port,
             policy_poll_interval_s=args.policy_poll_interval_s,
+            max_request_body_bytes=1_048_576,
         ),
         flush_worker=FlushWorkerConfig(
             enabled=bool(args.flush_worker_enabled),

predicate_authority/local_identity.py

@@ -259,11 +259,12 @@ def quarantine_queue_item(self, queue_item_id: str, reason: str) -> bool:
             return True
 
     def list_dead_letter_queue(self, limit: int | None = None) -> list[LedgerQueueItem]:
-        return self.list_flush_queue(
+        items = self.list_flush_queue(
             include_flushed=True,
             include_quarantined=True,
             limit=limit,
         )
+        return [item for item in items if item.quarantined]
 
     def requeue_item(self, queue_item_id: str, reset_attempts: bool = True) -> bool:
         with self._lock:
@@ -317,7 +318,10 @@ def _read_all_unlocked(self) -> dict[str, Any]:
         content = self._file_path.read_text(encoding="utf-8").strip()
         if content == "":
             return {"identities": {}, "flush_queue": {}}
-        loaded = json.loads(content)
+        try:
+            loaded = json.loads(content)
+        except json.JSONDecodeError:
+            return {"identities": {}, "flush_queue": {}}
         if isinstance(loaded, dict):
             if "identities" not in loaded:
                 loaded["identities"] = {}
@@ -356,7 +360,9 @@ def _parse_queue_item(self, raw: dict[str, Any]) -> LedgerQueueItem | None:
             return None
 
     def _write_all_unlocked(self, payload: dict[str, Any]) -> None:
-        self._file_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
+        tmp_path = self._file_path.with_name(f"{self._file_path.name}.{uuid.uuid4().hex}.tmp")
+        tmp_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
+        os.replace(tmp_path, self._file_path)
         self._chmod_file_safe()
 
     def _ensure_store_path(self) -> None:
@@ -366,10 +372,12 @@ def _ensure_store_path(self) -> None:
         except OSError:
             pass
         if not self._file_path.exists():
-            self._file_path.write_text(
+            tmp_path = self._file_path.with_name(f"{self._file_path.name}.{uuid.uuid4().hex}.tmp")
+            tmp_path.write_text(
                 json.dumps({"identities": {}, "flush_queue": {}}, indent=2),
                 encoding="utf-8",
             )
+            os.replace(tmp_path, self._file_path)
         self._chmod_file_safe()
 
     def _chmod_file_safe(self) -> None:
@@ -380,7 +388,7 @@ def _chmod_file_safe(self) -> None:
 
 
 @dataclass
-class LocalLedgerQueueEmitter(TraceEmitter):
+class LocalLedgerQueueEmitter:
     registry: LocalIdentityRegistry
     source: str = "predicate-authorityd"
 
@@ -389,7 +397,7 @@ def emit(self, event: ProofEvent) -> None:
 
 
 @dataclass
-class CompositeTraceEmitter(TraceEmitter):
+class CompositeTraceEmitter:
     emitters: tuple[TraceEmitter, ...]
 
     def emit(self, event: ProofEvent) -> None:

predicate_authority/policy.py

@@ -2,6 +2,7 @@
 
 from dataclasses import dataclass
 from fnmatch import fnmatch
+from threading import Lock
 
 from predicate_contracts import ActionRequest, AuthorizationReason, PolicyEffect, PolicyRule
 
@@ -22,15 +23,31 @@ def __init__(
     ) -> None:
         self._rules = rules
         self._global_max_delegation_depth = global_max_delegation_depth
+        self._lock = Lock()
 
     def replace_rules(self, rules: tuple[PolicyRule, ...]) -> None:
-        self._rules = rules
+        with self._lock:
+            self._rules = rules
 
     def set_global_max_delegation_depth(self, max_depth: int | None) -> None:
-        self._global_max_delegation_depth = max_depth
+        with self._lock:
+            self._global_max_delegation_depth = max_depth
+
+    def replace_policy(
+        self,
+        rules: tuple[PolicyRule, ...],
+        global_max_delegation_depth: int | None,
+    ) -> None:
+        with self._lock:
+            self._rules = rules
+            self._global_max_delegation_depth = global_max_delegation_depth
 
     def evaluate(self, request: ActionRequest, delegation_depth: int = 0) -> PolicyMatchResult:
-        matching_rules = [rule for rule in self._rules if self._matches_rule(rule, request)]
+        with self._lock:
+            rules = self._rules
+            global_max_delegation_depth = self._global_max_delegation_depth
+
+        matching_rules = [rule for rule in rules if self._matches_rule(rule, request)]
         if not matching_rules:
             return PolicyMatchResult(
                 allowed=False,
@@ -50,7 +67,10 @@ def evaluate(self, request: ActionRequest, delegation_depth: int = 0) -> PolicyM
             if rule.effect != PolicyEffect.ALLOW:
                 continue
 
-            effective_max_depth = self._effective_max_delegation_depth(rule)
+            effective_max_depth = self._effective_max_delegation_depth(
+                global_max_delegation_depth,
+                rule.max_delegation_depth,
+            )
             if effective_max_depth is not None and delegation_depth > effective_max_depth:
                 failure = PolicyMatchResult(
                     allowed=False,
@@ -102,9 +122,11 @@ def _matches_rule(rule: PolicyRule, request: ActionRequest) -> bool:
         )
         return principal_ok and action_ok and resource_ok
 
-    def _effective_max_delegation_depth(self, rule: PolicyRule) -> int | None:
-        global_max = self._global_max_delegation_depth
-        rule_max = rule.max_delegation_depth
+    @staticmethod
+    def _effective_max_delegation_depth(
+        global_max: int | None,
+        rule_max: int | None,
+    ) -> int | None:
         if global_max is None:
             return rule_max
         if rule_max is None:

predicate_authority/proof.py

@@ -2,6 +2,7 @@
 
 import time
 from dataclasses import dataclass, field
+from threading import Lock
 
 from predicate_contracts import ActionRequest, AuthorizationDecision, ProofEvent, TraceEmitter
 
@@ -10,6 +11,7 @@
 class InMemoryProofLedger:
     trace_emitter: TraceEmitter | None = None
     events: list[ProofEvent] = field(default_factory=list)
+    _lock: Lock = field(default_factory=Lock)
 
     def record(self, decision: AuthorizationDecision, request: ActionRequest) -> ProofEvent:
         event = ProofEvent(
@@ -22,7 +24,13 @@ def record(self, decision: AuthorizationDecision, request: ActionRequest) -> Pro
             mandate_id=decision.mandate.claims.mandate_id if decision.mandate else None,
             emitted_at_epoch_s=int(time.time()),
         )
-        self.events.append(event)
-        if self.trace_emitter is not None:
-            self.trace_emitter.emit(event)
+        with self._lock:
+            self.events.append(event)
+            trace_emitter = self.trace_emitter
+        if trace_emitter is not None:
+            trace_emitter.emit(event)
         return event
+
+    def event_count(self) -> int:
+        with self._lock:
+            return len(self.events)