"""Non-production reference gate for intent-bound, tenant-scoped agent authority. This module authorizes a declared action; it never executes a tool. Receipts are SHA-256 integrity checksums, not signatures or proof of principal identity. Production use requires authenticated identity, durable revocation/replay state, a cryptographic approval verifier, and a separate sandboxed tool executor. """ from __future__ import annotations from dataclasses import asdict, dataclass import hashlib import json import re import threading from typing import Callable, Optional POLICY_VERSION = "darkoza-intent-authority-v0.2" MAX_GRANT_TTL_SECONDS = 3600 MAX_APPROVAL_TTL_SECONDS = 900 _ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$") _IDEMPOTENCY = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$") _DIGEST = re.compile(r"^[0-9a-f]{64}$") # Risk is assigned by the policy, never by the model/request payload. ACTION_RISK = { "read": "low", "draft": "low", "write": "medium", "external_send": "high", "delete": "high", "payment": "critical", "credential_change": "critical", "network_change": "critical", } APPROVAL_REQUIRED = frozenset({"high", "critical"}) IDEMPOTENCY_REQUIRED = frozenset(ACTION_RISK) - {"read", "draft"} @dataclass(frozen=True) class AuthorityGrant: grant_id: str tenant_id: str principal_id: str task_id: str audience: str issued_at: int expires_at: int tools: tuple[str, ...] capabilities: tuple[str, ...] actions: tuple[str, ...] resources: tuple[str, ...] revoked: bool = False @dataclass(frozen=True) class ActionIntent: intent_id: str tenant_id: str principal_id: str task_id: str audience: str tool_id: str effect_digest: str capability: str action: str resource_id: str idempotency_key: Optional[str] = None @dataclass(frozen=True) class ApprovalAssertion: approval_id: str tenant_id: str principal_id: str task_id: str intent_digest: str approver_id: str issued_at: int expires_at: int ApprovalVerifier = Callable[[ApprovalAssertion], bool] def _canonical_digest(value: object) -> str: encoded = json.dumps( value, sort_keys=True, separators=(",", ":"), ensure_ascii=False ).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def intent_digest(intent: ActionIntent) -> str: return _canonical_digest(asdict(intent)) def _safe_intent_digest(intent: ActionIntent) -> Optional[str]: try: return intent_digest(intent) except (TypeError, ValueError): return None def verify_receipt(receipt: object) -> bool: """Check receipt integrity only; this does not authenticate its creator.""" required = { "policy_version", "decision", "reason", "risk", "intent_id", "intent_digest", "evaluated_at", "approval_id", "receipt_digest", } if not isinstance(receipt, dict) or set(receipt) != required: return False if receipt.get("policy_version") != POLICY_VERSION: return False if receipt.get("decision") not in {"allow", "deny"}: return False body = {k: v for k, v in receipt.items() if k != "receipt_digest"} try: return receipt["receipt_digest"] == _canonical_digest(body) except (TypeError, ValueError): return False class AuthorityGate: """Fail-closed policy evaluator with in-process one-time approval replay defense.""" def __init__( self, approval_verifier: Optional[ApprovalVerifier] = None, *, max_grant_ttl_seconds: int = MAX_GRANT_TTL_SECONDS, max_approval_ttl_seconds: int = MAX_APPROVAL_TTL_SECONDS, ) -> None: self._approval_verifier = approval_verifier self._max_grant_ttl = max_grant_ttl_seconds self._max_approval_ttl = max_approval_ttl_seconds self._used_approvals: set[tuple[str, str, str]] = set() self._used_idempotency_keys: set[tuple[str, str, str, str]] = set() self._lock = threading.Lock() @staticmethod def _valid_id(value: object) -> bool: return isinstance(value, str) and bool(_ID.fullmatch(value)) @staticmethod def _valid_time(value: object) -> bool: return type(value) is int and value >= 0 def _receipt( self, *, decision: str, reason: str, risk: str, intent: ActionIntent, now: int, approval_id: Optional[str] = None, ) -> dict: body = { "policy_version": POLICY_VERSION, "decision": decision, "reason": reason, "risk": risk, "intent_id": getattr(intent, "intent_id", None), "intent_digest": _safe_intent_digest(intent) if isinstance(intent, ActionIntent) else None, "evaluated_at": now, "approval_id": approval_id, } return {**body, "receipt_digest": _canonical_digest(body)} def evaluate( self, grant: AuthorityGrant, intent: ActionIntent, *, now: int, approval: Optional[ApprovalAssertion] = None, ) -> dict: """Return a deterministic allow/deny receipt. No tool is executed.""" if not isinstance(grant, AuthorityGrant) or not isinstance(intent, ActionIntent): placeholder = intent if isinstance(intent, ActionIntent) else ActionIntent( "invalid", "invalid", "invalid", "invalid", "invalid", "invalid", "invalid", "invalid", "invalid", "invalid" ) return self._receipt( decision="deny", reason="input_invalid", risk="unknown", intent=placeholder, now=now if self._valid_time(now) else 0, ) risk = ACTION_RISK.get(intent.action, "unknown") if isinstance(intent.action, str) else "unknown" evaluated_at = now if self._valid_time(now) else 0 if not self._valid_time(now): return self._receipt(decision="deny", reason="clock_invalid", risk=risk, intent=intent, now=evaluated_at) identifiers = ( grant.grant_id, grant.tenant_id, grant.principal_id, grant.task_id, grant.audience, intent.intent_id, intent.tenant_id, intent.principal_id, intent.task_id, intent.audience, intent.tool_id, intent.capability, intent.resource_id, ) if not all(self._valid_id(x) for x in identifiers): return self._receipt(decision="deny", reason="identifier_invalid", risk=risk, intent=intent, now=evaluated_at) if not isinstance(intent.effect_digest, str) or not _DIGEST.fullmatch(intent.effect_digest): return self._receipt(decision="deny", reason="effect_digest_invalid", risk=risk, intent=intent, now=evaluated_at) if grant.revoked: return self._receipt(decision="deny", reason="grant_revoked", risk=risk, intent=intent, now=evaluated_at) if not all(self._valid_time(x) for x in (grant.issued_at, grant.expires_at)): return self._receipt(decision="deny", reason="grant_time_invalid", risk=risk, intent=intent, now=evaluated_at) if grant.expires_at <= grant.issued_at or grant.expires_at - grant.issued_at > self._max_grant_ttl: return self._receipt(decision="deny", reason="grant_ttl_invalid", risk=risk, intent=intent, now=evaluated_at) if evaluated_at < grant.issued_at: return self._receipt(decision="deny", reason="grant_not_yet_valid", risk=risk, intent=intent, now=evaluated_at) if evaluated_at >= grant.expires_at: return self._receipt(decision="deny", reason="grant_expired", risk=risk, intent=intent, now=evaluated_at) # Explicit context binding prevents cross-tenant, cross-principal, # cross-task and cross-service confused-deputy use. bindings = ( (intent.tenant_id, grant.tenant_id, "tenant_mismatch"), (intent.principal_id, grant.principal_id, "principal_mismatch"), (intent.task_id, grant.task_id, "task_mismatch"), (intent.audience, grant.audience, "audience_mismatch"), ) for actual, expected, reason in bindings: if actual != expected: return self._receipt(decision="deny", reason=reason, risk=risk, intent=intent, now=evaluated_at) for field in (grant.tools, grant.capabilities, grant.actions, grant.resources): if not isinstance(field, tuple) or any(not self._valid_id(v) for v in field): return self._receipt(decision="deny", reason="grant_scope_invalid", risk=risk, intent=intent, now=evaluated_at) if (len(set(grant.tools)) != len(grant.tools) or len(set(grant.capabilities)) != len(grant.capabilities) or len(set(grant.actions)) != len(grant.actions) or len(set(grant.resources)) != len(grant.resources)): return self._receipt(decision="deny", reason="grant_scope_duplicate", risk=risk, intent=intent, now=evaluated_at) if risk == "unknown": return self._receipt(decision="deny", reason="action_unknown", risk=risk, intent=intent, now=evaluated_at) if intent.tool_id not in grant.tools: return self._receipt(decision="deny", reason="tool_not_granted", risk=risk, intent=intent, now=evaluated_at) if intent.capability not in grant.capabilities: return self._receipt(decision="deny", reason="capability_not_granted", risk=risk, intent=intent, now=evaluated_at) if intent.action not in grant.actions: return self._receipt(decision="deny", reason="action_not_granted", risk=risk, intent=intent, now=evaluated_at) if intent.resource_id not in grant.resources: return self._receipt(decision="deny", reason="resource_not_granted", risk=risk, intent=intent, now=evaluated_at) if intent.action in IDEMPOTENCY_REQUIRED and ( not isinstance(intent.idempotency_key, str) or not _IDEMPOTENCY.fullmatch(intent.idempotency_key) ): return self._receipt(decision="deny", reason="idempotency_key_required", risk=risk, intent=intent, now=evaluated_at) if risk not in APPROVAL_REQUIRED: if intent.action in IDEMPOTENCY_REQUIRED: idempotency_scope = (intent.tenant_id, intent.principal_id, intent.tool_id, intent.idempotency_key) with self._lock: if idempotency_scope in self._used_idempotency_keys: return self._receipt(decision="deny", reason="idempotency_replayed", risk=risk, intent=intent, now=evaluated_at) self._used_idempotency_keys.add(idempotency_scope) return self._receipt(decision="allow", reason="scope_match", risk=risk, intent=intent, now=evaluated_at) if not isinstance(approval, ApprovalAssertion): return self._receipt(decision="deny", reason="human_approval_required", risk=risk, intent=intent, now=evaluated_at) approval_ids = (approval.approval_id, approval.tenant_id, approval.principal_id, approval.task_id, approval.approver_id) if not all(self._valid_id(x) for x in approval_ids): return self._receipt(decision="deny", reason="approval_invalid", risk=risk, intent=intent, now=evaluated_at) if approval.tenant_id != intent.tenant_id or approval.principal_id != intent.principal_id or approval.task_id != intent.task_id: return self._receipt(decision="deny", reason="approval_binding_mismatch", risk=risk, intent=intent, now=evaluated_at) expected_intent_digest = intent_digest(intent) if approval.intent_digest != expected_intent_digest: return self._receipt(decision="deny", reason="approval_intent_mismatch", risk=risk, intent=intent, now=evaluated_at) if not self._valid_time(approval.issued_at) or not self._valid_time(approval.expires_at): return self._receipt(decision="deny", reason="approval_time_invalid", risk=risk, intent=intent, now=evaluated_at) if approval.expires_at <= approval.issued_at or approval.expires_at - approval.issued_at > self._max_approval_ttl or evaluated_at < approval.issued_at or evaluated_at >= approval.expires_at: return self._receipt(decision="deny", reason="approval_expired_or_ttl_invalid", risk=risk, intent=intent, now=evaluated_at) if self._approval_verifier is None: return self._receipt(decision="deny", reason="approval_verifier_unavailable", risk=risk, intent=intent, now=evaluated_at) # Verification must be provided by a trusted identity/signature layer. try: verified = self._approval_verifier(approval) is True except Exception: verified = False if not verified: return self._receipt(decision="deny", reason="approval_authentication_failed", risk=risk, intent=intent, now=evaluated_at) approval_scope = (approval.tenant_id, approval.principal_id, approval.approval_id) idempotency_scope = (intent.tenant_id, intent.principal_id, intent.tool_id, intent.idempotency_key) with self._lock: if approval_scope in self._used_approvals: return self._receipt(decision="deny", reason="approval_replayed", risk=risk, intent=intent, now=evaluated_at, approval_id=approval.approval_id) if idempotency_scope in self._used_idempotency_keys: return self._receipt(decision="deny", reason="idempotency_replayed", risk=risk, intent=intent, now=evaluated_at, approval_id=approval.approval_id) self._used_approvals.add(approval_scope) self._used_idempotency_keys.add(idempotency_scope) return self._receipt( decision="allow", reason="scope_match_and_approval_verified", risk=risk, intent=intent, now=evaluated_at, approval_id=approval.approval_id, )