Phase 1: SDK only guard

Author: rcholic

.pre-commit-config.yaml

@@ -1,4 +1,4 @@
-# Pre-commit hooks for AgentIdentity repository
+# Pre-commit hooks for predicate-authority repository
 # Baseline adapted from /Code/Sentience/sdk-python/.pre-commit-config.yaml
 
 repos:

Makefile

@@ -1,11 +1,14 @@
-.PHONY: hooks lint format format-python format-docs lint-docs
+.PHONY: hooks lint test format format-python format-docs lint-docs
 
 hooks:
 	pre-commit install
 
 lint:
 	pre-commit run --all-files
 
+test:
+	python -m pytest -q
+
 format: format-python format-docs
 
 format-python:

docs/better-sdk-opportunity-proposal.md

@@ -15,6 +15,18 @@ This proposal answers the "Better SDK Opportunity" in `northstar.md` by combinin
 - Caracal's strongest ideas (short-lived mandates, scope checks, fail-closed gateway-style enforcement, immutable authority ledger),
 - A bridge-first strategy (works with Azure AD/Okta/Auth0 and existing agent stacks).
 
+## Progress Dashboard
+
+Status snapshot date: 2026-02-16
+
+| Phase | Status | ETA | Owner |
+| --- | --- | --- | --- |
+| Phase 0: Architecture and Spec Lock | Partially complete | 1-2 weeks total (remaining: sign-off + schema/process formalization) | SDK + Platform + Security |
+| Phase 1: Local SDK Guard (MVP) | In progress | 3-5 weeks total (remaining: `sdk-python` hooks + OTel export + examples + CI publish flow) | SDK |
+| Phase 2: Sidecar + Identity Bridge | Not started (design only) | 4-6 weeks | Platform + Identity |
+| Phase 3: Hosted Governance Control Plane | Not started (design only) | 6-8 weeks | Platform + Product |
+| Phase 4: Enterprise Hardening and Scale | Not started (design only) | Ongoing (first 4-6 weeks) | Platform + Security + GTM |
+
 
 ## TL;DR Design
 
@@ -511,6 +523,21 @@ async def web_search_tool(query: str):
 - Basic policy DSL.
 - Trace/proof event emission to existing tracer.
 
+Status (as of 2026-02-16): **in progress (MVP scaffold implemented in this `predicate-authority` repository)**
+
+- Completed in repo:
+  - `predicate-contracts` package scaffold with typed contracts and protocols.
+  - `predicate-authority` local `ActionGuard.authorize(...)` + `enforce(...)`.
+  - Signed local mandates with TTL + verification.
+  - Local policy evaluation and normalized deny reasons.
+  - In-memory proof ledger with optional trace emitter interface.
+  - pytest coverage for policy, mandate signing, and proof emission paths.
+- Pending for full Phase 1 exit:
+  - direct `sdk-python` integration hooks (pre-action + postcondition linkage),
+  - OpenTelemetry-native event export (beyond protocol-level trace emitter),
+  - developer quickstart/examples for browser/MCP/HTTP guard patterns,
+  - package publishing pipeline verification (`predicate-contracts` -> `predicate-authority`).
+
 ## Phase 2: Sidecar and IdP bridge (4-8 weeks)
 
 - `predicate-authorityd`.
@@ -609,6 +636,14 @@ Exit criteria:
 - compatibility mapping to existing `sdk-python` step lifecycle approved.
 - release orchestration design approved for multi-package PyPI publishing (`predicate-contracts` then `predicate-authority`).
 
+Current status: **partially complete**
+
+- [x] dependency graph/import boundaries documented in this proposal.
+- [x] package scaffolding started in this `predicate-authority` repository (`predicate-contracts`, `predicate-authority`).
+- [ ] formal design sign-off from SDK/platform/security.
+- [ ] versioned schema docs publication process.
+- [ ] approved compatibility mapping with `sdk-python` lifecycle owners.
+
 ## Phase 1: Local SDK Guard (MVP) (3-5 weeks)
 
 Objective: deliver immediate value with in-process pre-execution authority.
@@ -632,6 +667,18 @@ Exit criteria:
 - developer quickstart validated end-to-end on local-only mode.
 - CI release pipeline can publish and verify `predicate-contracts` and `predicate-authority` in dependency order.
 
+Current status: **in progress**
+
+- [x] local `ActionGuard.authorize(...)`.
+- [x] signed local mandates.
+- [x] local policy evaluation.
+- [x] fail-closed deny path with normalized reason enums.
+- [x] deterministic regression tests for authorize/deny paths.
+- [ ] `sdk-python` runtime integration hooks.
+- [ ] OpenTelemetry-native authority event export.
+- [ ] quickstart/examples for browser/MCP/outbound HTTP.
+- [ ] dependency-ordered package publish pipeline in CI.
+
 ## Phase 2: Sidecar + Identity Bridge (4-6 weeks)
 
 Objective: production-ready token lifecycle and enterprise identity compatibility.
@@ -657,6 +704,8 @@ Exit criteria:
 - bridge token exchange validated against at least one enterprise IdP.
 - sidecar survives restart/network partition with fail-closed guarantees.
 
+Current status: **not started (design only)**
+
 ## Phase 3: Hosted Governance Control Plane (6-8 weeks)
 
 Objective: ship monetizable cloud governance capabilities.
@@ -675,6 +724,8 @@ Exit criteria:
 - kill-switch propagation meets incident response target.
 - billable usage pipeline reconciles authority + snapshot credits accurately.
 
+Current status: **not started (design only)**
+
 ## Phase 4: Enterprise Hardening and Scale (ongoing, first 4-6 weeks)
 
 Objective: make it enterprise-ready for regulated production.
@@ -693,6 +744,8 @@ Exit criteria:
 - defined SLOs met in staging/load tests.
 - enterprise onboarding playbook validated with pilot accounts.
 
+Current status: **not started (design only)**
+
 ## Cross-Phase Dependencies
 
 - `sdk-python` runtime contract stability (snapshot schema, assertion labels, step metadata).

predicate_authority/__init__.py

@@ -0,0 +1,18 @@
+from predicate_authority.bridge import IdentityBridge, TokenExchangeResult
+from predicate_authority.errors import AuthorizationDeniedError
+from predicate_authority.guard import ActionExecutionResult, ActionGuard
+from predicate_authority.mandate import LocalMandateSigner
+from predicate_authority.policy import PolicyEngine, PolicyMatchResult
+from predicate_authority.proof import InMemoryProofLedger
+
+__all__ = [
+    "ActionExecutionResult",
+    "ActionGuard",
+    "AuthorizationDeniedError",
+    "IdentityBridge",
+    "InMemoryProofLedger",
+    "LocalMandateSigner",
+    "PolicyEngine",
+    "PolicyMatchResult",
+    "TokenExchangeResult",
+]

predicate_authority/bridge.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+import hashlib
+import time
+from dataclasses import dataclass
+
+from predicate_contracts import PrincipalRef, StateEvidence
+
+
+@dataclass(frozen=True)
+class TokenExchangeResult:
+    access_token: str
+    expires_at_epoch_s: int
+    token_type: str = "Bearer"
+
+
+class IdentityBridge:
+    """Local placeholder bridge for Phase 1.
+
+    This keeps an explicit interface so Phase 2 can swap in a real OIDC/Entra bridge.
+    """
+
+    def __init__(self, token_ttl_seconds: int = 300) -> None:
+        self._token_ttl_seconds = token_ttl_seconds
+
+    def exchange_token(
+        self, subject: PrincipalRef, state_evidence: StateEvidence
+    ) -> TokenExchangeResult:
+        expires_at = int(time.time()) + self._token_ttl_seconds
+        token_seed = f"{subject.principal_id}|{state_evidence.state_hash}|{expires_at}"
+        token_hash = hashlib.sha256(token_seed.encode("utf-8")).hexdigest()
+        return TokenExchangeResult(
+            access_token=f"local.{token_hash}", expires_at_epoch_s=expires_at
+        )

predicate_authority/errors.py

@@ -0,0 +1,9 @@
+from __future__ import annotations
+
+from predicate_contracts import AuthorizationDecision
+
+
+class AuthorizationDeniedError(RuntimeError):
+    def __init__(self, decision: AuthorizationDecision) -> None:
+        self.decision = decision
+        super().__init__(f"Authorization denied: {decision.reason.value}")

predicate_authority/guard.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Generic, TypeVar
+
+from predicate_authority.errors import AuthorizationDeniedError
+from predicate_authority.mandate import LocalMandateSigner
+from predicate_authority.policy import PolicyEngine
+from predicate_authority.proof import InMemoryProofLedger
+from predicate_contracts import (
+    ActionRequest,
+    AuthorizationDecision,
+    AuthorizationReason,
+    SignedMandate,
+)
+
+T = TypeVar("T")
+
+
+@dataclass(frozen=True)
+class ActionExecutionResult(Generic[T]):
+    value: T
+    decision: AuthorizationDecision
+    mandate: SignedMandate
+
+
+class ActionGuard:
+    def __init__(
+        self,
+        policy_engine: PolicyEngine,
+        mandate_signer: LocalMandateSigner,
+        proof_ledger: InMemoryProofLedger,
+    ) -> None:
+        self._policy_engine = policy_engine
+        self._mandate_signer = mandate_signer
+        self._proof_ledger = proof_ledger
+
+    def authorize(self, request: ActionRequest) -> AuthorizationDecision:
+        evaluation = self._policy_engine.evaluate(request)
+        if not evaluation.allowed:
+            decision = AuthorizationDecision(
+                allowed=False,
+                reason=evaluation.reason,
+                violated_rule=evaluation.matched_rule,
+                missing_labels=evaluation.missing_labels,
+            )
+            self._proof_ledger.record(decision, request)
+            return decision
+
+        mandate = self._mandate_signer.issue(request)
+        decision = AuthorizationDecision(
+            allowed=True,
+            reason=AuthorizationReason.ALLOWED,
+            mandate=mandate,
+            violated_rule=evaluation.matched_rule,
+        )
+        self._proof_ledger.record(decision, request)
+        return decision
+
+    def enforce(
+        self, action_callable: Callable[[], T], request: ActionRequest
+    ) -> ActionExecutionResult[T]:
+        decision = self.authorize(request)
+        if not decision.allowed or decision.mandate is None:
+            raise AuthorizationDeniedError(decision)
+        value = action_callable()
+        return ActionExecutionResult(value=value, decision=decision, mandate=decision.mandate)

predicate_authority/mandate.py

@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import json
+import time
+from dataclasses import asdict
+
+from predicate_contracts import ActionRequest, MandateClaims, SignedMandate
+
+
+class LocalMandateSigner:
+    def __init__(self, secret_key: str, ttl_seconds: int = 300) -> None:
+        if ttl_seconds <= 0:
+            raise ValueError("ttl_seconds must be > 0")
+        self._secret_key = secret_key.encode("utf-8")
+        self._ttl_seconds = ttl_seconds
+
+    def issue(self, request: ActionRequest) -> SignedMandate:
+        issued_at = int(time.time())
+        expires_at = issued_at + self._ttl_seconds
+        intent_hash = hashlib.sha256(request.action_spec.intent.encode("utf-8")).hexdigest()
+        mandate_id_seed = (
+            f"{request.principal.principal_id}|"
+            f"{request.action_spec.action}|"
+            f"{request.action_spec.resource}|"
+            f"{intent_hash}|"
+            f"{request.state_evidence.state_hash}|"
+            f"{issued_at}"
+        )
+        mandate_id = hashlib.sha256(mandate_id_seed.encode("utf-8")).hexdigest()[:24]
+
+        claims = MandateClaims(
+            mandate_id=mandate_id,
+            principal_id=request.principal.principal_id,
+            action=request.action_spec.action,
+            resource=request.action_spec.resource,
+            intent_hash=intent_hash,
+            state_hash=request.state_evidence.state_hash,
+            issued_at_epoch_s=issued_at,
+            expires_at_epoch_s=expires_at,
+        )
+        token, signature = self._sign_claims(claims)
+        return SignedMandate(token=token, claims=claims, signature=signature)
+
+    def verify(self, token: str) -> SignedMandate | None:
+        parts = token.split(".")
+        if len(parts) != 3:
+            return None
+
+        encoded_header, encoded_payload, encoded_signature = parts
+        signing_input = f"{encoded_header}.{encoded_payload}".encode()
+        expected_signature = self._hmac(signing_input)
+        expected_signature_encoded = self._base64url_encode(expected_signature)
+        if not hmac.compare_digest(expected_signature_encoded, encoded_signature):
+            return None
+
+        try:
+            payload_json = self._base64url_decode(encoded_payload).decode("utf-8")
+            payload = json.loads(payload_json)
+            claims = MandateClaims(**payload)
+        except (ValueError, TypeError, json.JSONDecodeError):
+            return None
+
+        now_epoch = int(time.time())
+        if claims.expires_at_epoch_s < now_epoch:
+            return None
+        return SignedMandate(token=token, claims=claims, signature=encoded_signature)
+
+    def _sign_claims(self, claims: MandateClaims) -> tuple[str, str]:
+        header_json = json.dumps(
+            {"alg": "HS256", "typ": "JWT"}, separators=(",", ":"), sort_keys=True
+        )
+        payload_json = json.dumps(asdict(claims), separators=(",", ":"), sort_keys=True)
+
+        encoded_header = self._base64url_encode(header_json.encode("utf-8"))
+        encoded_payload = self._base64url_encode(payload_json.encode("utf-8"))
+        signing_input = f"{encoded_header}.{encoded_payload}".encode()
+        signature = self._base64url_encode(self._hmac(signing_input))
+        token = f"{encoded_header}.{encoded_payload}.{signature}"
+        return token, signature
+
+    def _hmac(self, payload: bytes) -> bytes:
+        return hmac.new(self._secret_key, payload, hashlib.sha256).digest()
+
+    @staticmethod
+    def _base64url_encode(value: bytes) -> str:
+        return base64.urlsafe_b64encode(value).rstrip(b"=").decode("ascii")
+
+    @staticmethod
+    def _base64url_decode(value: str) -> bytes:
+        padding = "=" * ((4 - len(value) % 4) % 4)
+        return base64.urlsafe_b64decode(value + padding)