"""Opaque, exact, one-use approvals for tainted model-requested actions. The model may propose an action after untrusted context, but only the server stores and later executes the exact approved tool input. Browser-visible fields are display copies, never authority. """ from __future__ import annotations import hashlib import json import os import secrets import threading import time from dataclasses import dataclass, field from typing import Any from src.tool_capabilities import ToolCapabilities, capabilities_for_action DEFAULT_APPROVAL_TTL_SECONDS = 10 * 60 DEFAULT_MAX_PENDING_APPROVALS = 2048 def _normalized_owner(owner: Any) -> str: return str(owner or "").strip().casefold() def _normalized_workspace(workspace: Any) -> str: if not isinstance(workspace, str) or not workspace.strip(): return "" return os.path.realpath(os.path.expanduser(workspace)) def _canonical_digest(payload: dict[str, Any]) -> str: encoded = json.dumps( payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False, ).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def document_content_digest(content: Any) -> str: """Return the stable server-side fingerprint used to seal a document.""" return hashlib.sha256(str(content or "").encode("utf-8")).hexdigest() def _binding_payload( *, owner: Any, session_id: Any, origin_run_id: Any, tool_name: Any, content: Any, workspace: Any, document_id: Any, document_version: Any, document_digest: Any, external_untrusted_context_seen: bool, effects: tuple[str, ...], result_integrity: str, ) -> dict[str, Any]: return { "owner": _normalized_owner(owner), "session_id": str(session_id or ""), "origin_run_id": str(origin_run_id or ""), "tool_name": str(tool_name or ""), "content": str(content or ""), "workspace": _normalized_workspace(workspace), "document_id": str(document_id or ""), "document_version": ( int(document_version) if document_version is not None else None ), "document_digest": str(document_digest or "").strip().lower(), "external_untrusted_context_seen": bool(external_untrusted_context_seen), "effects": list(effects), "result_integrity": str(result_integrity), } @dataclass(frozen=True) class PendingToolApproval: approval_id: str owner: str session_id: str origin_run_id: str tool_name: str content: str workspace: str document_id: str document_version: int | None document_digest: str external_untrusted_context_seen: bool effects: tuple[str, ...] result_integrity: str digest: str created_at: float expires_at: float def public_payload(self, *, reason: str | None = None) -> dict[str, Any]: return { "kind": "tool_approval", "approval_id": self.approval_id, "question": "Allow this exact action once?", "description": reason or ( "Untrusted context influenced this run, so this action needs " "your explicit approval." ), "options": [ { "label": "Allow once", "value": "approve", "description": "Execute only the sealed action shown here.", }, { "label": "Deny", "value": "deny", "description": "Do not execute it.", }, ], "action": { "tool": self.tool_name, # Show the complete sealed input so approval never hides # trailing lines. This is not read back as authority. "content": self.content, "digest": self.digest[:16], "effects": list(self.effects), "workspace": self.workspace or None, "document_id": self.document_id or None, "document_version": self.document_version, }, } @dataclass class ExactToolApproval: """A consumed grant that the dispatcher can claim exactly once.""" pending: PendingToolApproval _claimed: bool = field(default=False, init=False, repr=False) _lock: threading.Lock = field(default_factory=threading.Lock, init=False, repr=False) def _matches_unlocked( self, *, owner: Any, session_id: Any, tool_name: Any, content: Any, workspace: Any, ) -> bool: if self._claimed: return False capabilities = capabilities_for_action(tool_name, content) effects = tuple(sorted(effect.value for effect in capabilities.effects)) result_integrity = capabilities.result_integrity.value if ( effects != self.pending.effects or result_integrity != self.pending.result_integrity ): return False expected = _binding_payload( owner=owner, session_id=session_id, origin_run_id=self.pending.origin_run_id, tool_name=tool_name, content=content, workspace=workspace, document_id=self.pending.document_id, document_version=self.pending.document_version, document_digest=self.pending.document_digest, external_untrusted_context_seen=( self.pending.external_untrusted_context_seen ), effects=effects, result_integrity=result_integrity, ) return _canonical_digest(expected) == self.pending.digest def matches( self, *, owner: Any, session_id: Any, tool_name: Any, content: Any, workspace: Any, ) -> bool: with self._lock: return self._matches_unlocked( owner=owner, session_id=session_id, tool_name=tool_name, content=content, workspace=workspace, ) def claim( self, *, owner: Any, session_id: Any, tool_name: Any, content: Any, workspace: Any, ) -> bool: with self._lock: if not self._matches_unlocked( owner=owner, session_id=session_id, tool_name=tool_name, content=content, workspace=workspace, ): return False self._claimed = True return True class ToolApprovalStore: """Thread-safe pending approval registry with destructive consumption.""" def __init__( self, *, ttl_seconds: int = DEFAULT_APPROVAL_TTL_SECONDS, max_pending: int = DEFAULT_MAX_PENDING_APPROVALS, ): self._ttl_seconds = max(1, int(ttl_seconds)) self._max_pending = max(1, int(max_pending)) self._pending: dict[str, PendingToolApproval] = {} self._lock = threading.Lock() def _purge_expired_locked(self, now: float) -> None: expired = [ approval_id for approval_id, pending in self._pending.items() if pending.expires_at <= now ] for approval_id in expired: self._pending.pop(approval_id, None) def create( self, *, owner: Any, session_id: Any, origin_run_id: Any, tool_name: Any, content: Any, workspace: Any, document_id: Any = None, document_version: Any = None, document_digest: Any = None, external_untrusted_context_seen: bool, capabilities: ToolCapabilities, ) -> PendingToolApproval: now = time.time() effects = tuple(sorted(effect.value for effect in capabilities.effects)) result_integrity = capabilities.result_integrity.value payload = _binding_payload( owner=owner, session_id=session_id, origin_run_id=origin_run_id, tool_name=tool_name, content=content, workspace=workspace, document_id=document_id, document_version=document_version, document_digest=document_digest, external_untrusted_context_seen=external_untrusted_context_seen, effects=effects, result_integrity=result_integrity, ) pending = PendingToolApproval( approval_id=secrets.token_urlsafe(32), owner=payload["owner"], session_id=payload["session_id"], origin_run_id=payload["origin_run_id"], tool_name=payload["tool_name"], content=payload["content"], workspace=payload["workspace"], document_id=payload["document_id"], document_version=payload["document_version"], document_digest=payload["document_digest"], external_untrusted_context_seen=payload[ "external_untrusted_context_seen" ], effects=effects, result_integrity=result_integrity, digest=_canonical_digest(payload), created_at=now, expires_at=now + self._ttl_seconds, ) with self._lock: self._purge_expired_locked(now) # The chat UI exposes one pending card per session, so supersede an # older action there. Headless/manual-test callers use an empty # session id; keep independent origin runs separate so two skill # tests owned by the same user cannot invalidate each other. superseded = [ approval_id for approval_id, existing in self._pending.items() if ( existing.owner == pending.owner and existing.session_id == pending.session_id and ( bool(pending.session_id) or existing.origin_run_id == pending.origin_run_id ) ) ] for approval_id in superseded: self._pending.pop(approval_id, None) while len(self._pending) >= self._max_pending: oldest_id = min( self._pending, key=lambda approval_id: self._pending[approval_id].created_at, ) self._pending.pop(oldest_id, None) self._pending[pending.approval_id] = pending return pending def consume( self, approval_id: Any, *, decision: Any, owner: Any, session_id: Any, ) -> ExactToolApproval | None: now = time.time() with self._lock: self._purge_expired_locked(now) approval_key = str(approval_id or "") pending = self._pending.get(approval_key) if pending is None: return None if ( pending.owner != _normalized_owner(owner) or pending.session_id != str(session_id or "") ): # Authentication is checked before destructive consumption so # a leaked/guessed opaque id cannot be used to invalidate # another owner's pending action. return None self._pending.pop(approval_key, None) if str(decision or "").strip().lower() != "approve": return None return ExactToolApproval(pending) def peek(self, approval_id: Any) -> PendingToolApproval | None: now = time.time() with self._lock: self._purge_expired_locked(now) return self._pending.get(str(approval_id or "")) tool_approval_store = ToolApprovalStore()