"""Permission Gateway — security boundary between agents and tools. The Permission Gateway is the single point through which all agent tool calls must pass. It combines: 1. Agent identity scope checks (does the agent have the required permission?) 2. Policy engine evaluation (does the system policy allow this action?) 3. Approval routing (does this action require human approval?) Core principle: DEFAULT DENY. No tool call is permitted unless explicitly allowed by both identity scope AND policy engine. """ from __future__ import annotations from dataclasses import dataclass, field from enum import Enum from typing import Any, ClassVar from cyberbrain26.security_plane.actions import ( ACTION_SCOPE_REQUIREMENTS, KNOWN_SAFE_ACTIONS, SENSITIVE_ACTIONS, normalize_action, ) from cyberbrain26.security_plane.audit import AuditEventType, AuditLog from cyberbrain26.security_plane.identity import AgentIdentity, TrustLevel from cyberbrain26.security_plane.policy import ( PolicyContext, PolicyDecision, PolicyEngine, ) class PermissionDecision(str, Enum): """Outcome of a permission evaluation.""" ALLOW = "allow" DENY = "deny" REQUIRE_APPROVAL = "require_approval" class SecurityError(Exception): """Raised when a security boundary check fails. Distinguishes deliberate security denials and approval requirements from operational/system errors. """ @dataclass class PermissionRequest: """A request for permission to perform an action.""" agent: AgentIdentity action: str resource: str | None = None context: dict[str, Any] = field(default_factory=dict) risk_score: float = 0.0 @dataclass class PermissionResult: """Result of a permission evaluation.""" decision: PermissionDecision reason: str policy_decision: PolicyDecision | None = None scope_required: str | None = None approval_required: bool = False request: PermissionRequest | None = None class PermissionGateway: """Security boundary that gates all agent tool calls. Evaluates whether an agent identity is permitted to perform an action by checking both identity scopes and system policy. Usage: gateway = PermissionGateway() result = gateway.evaluate(identity, "execute_shell", "/bin/ls") if result.decision == PermissionDecision.DENY: raise SecurityError(result.reason) """ # Mapping from action to required permission scope (class-level constant). # The vocabulary itself lives in ``security_plane.actions`` so the scope gate # and the policy gate cannot drift apart; this copy stays a public, # overridable class attribute for callers that subclass the gateway. ACTION_SCOPE_MAP: ClassVar[dict[str, str]] = dict(ACTION_SCOPE_REQUIREMENTS) def __init__(self, policy_engine: PolicyEngine | None = None, audit_log: AuditLog | None = None) -> None: self.policy_engine = policy_engine or PolicyEngine() self.audit_log = audit_log def evaluate( self, agent: AgentIdentity, action: str, resource: str | None = None, context: dict[str, Any] | None = None, risk_score: float = 0.0, ) -> PermissionResult: """Evaluate whether an agent can perform an action. Args: agent: The agent identity requesting the action. action: The action being requested. resource: Optional resource being accessed. context: Optional additional context (environment, mode, etc.). risk_score: Risk score of the request (0.0 to 1.0). Returns: PermissionResult with decision, reason, and metadata. Security rationale: the action is canonicalized with ``normalize_action`` before *any* lookup, so ``" read_file "``, ``"READ_FILE"`` and ``"read_file"`` are one action and reach one verdict. An action that is neither scope-mapped nor on the explicitly-named ``KNOWN_SAFE_ACTIONS`` allowlist is denied here rather than passed on with no scope requirement — the previous behaviour skipped the scope gate entirely for such a name, so any invented string was treated as harmless. A sensitive action that is classified but carries no scope requirement gets its own denial with its own reason, so "known, risky, and unreachable" can never be mistaken for "unknown name". """ canonical_action = normalize_action(action) request = PermissionRequest( agent=agent, action=canonical_action, resource=resource, context=context or {}, risk_score=risk_score, ) # Step 0: Trust check — untrusted agents are always denied if agent.trust_level == TrustLevel.UNTRUSTED: if self.audit_log: self.audit_log.record( AuditEventType.IDENTITY_CHECK, actor=agent.agent_id, agent_id=agent.agent_id, action=canonical_action, resource=resource or "", result="denied", error="untrusted_agent", ) return PermissionResult( decision=PermissionDecision.DENY, reason=f"Agent {agent.agent_id} is UNTRUSTED — denied by trust policy", request=request, ) # Step 1: Unknown action names are denied before anything is consulted. # A name outside the canonical vocabulary carries no scope requirement # and no policy rule, so continuing would let it through every gate. scope_required = self.ACTION_SCOPE_MAP.get(canonical_action) # Step 1a: A known *sensitive* action with no scope requirement is denied # explicitly and with its own reason. Deny-by-default would swallow it # anyway, but only with the misleading "not a registered action" text, # which hid the real defect: a sensitive action added to # SENSITIVE_ACTIONS without a matching entry in ACTION_SCOPE_REQUIREMENTS # is unreachable — policy never gets to route it to the human-approval # gate the sensitivity classification prepared for it. This branch makes # that state a loud, deliberate, audited denial rather than an accident, # so the incomplete vocabulary cannot hide again. if scope_required is None and canonical_action in SENSITIVE_ACTIONS: if self.audit_log: self.audit_log.record( AuditEventType.PERMISSION_CHECK, actor=agent.agent_id, agent_id=agent.agent_id, action=canonical_action, resource=resource or "", result="unscoped_sensitive_action", error=f"unscoped_sensitive_action:{canonical_action}", ) return PermissionResult( decision=PermissionDecision.DENY, reason=( f"Sensitive action '{canonical_action}' has no identity scope requirement " f"— denied by default for agent {agent.agent_id}" ), request=request, ) if scope_required is None and canonical_action not in KNOWN_SAFE_ACTIONS: if self.audit_log: self.audit_log.record( AuditEventType.PERMISSION_CHECK, actor=agent.agent_id, agent_id=agent.agent_id, action=canonical_action, resource=resource or "", result="unknown_action", error=f"unknown_action:{canonical_action or ''}", ) return PermissionResult( decision=PermissionDecision.DENY, reason=( f"Action '{canonical_action or ''}' is not a registered action " f"for agent {agent.agent_id} — denied by default" ), request=request, ) # Step 2: Check identity scope if scope_required is not None and not agent.has_scope(scope_required): if self.audit_log: self.audit_log.record( AuditEventType.PERMISSION_CHECK, actor=agent.agent_id, agent_id=agent.agent_id, action=canonical_action, resource=resource or "", result="scope_check_failed", error=f"missing_scope:{scope_required}", ) return PermissionResult( decision=PermissionDecision.DENY, reason=f"Agent {agent.agent_id} lacks required scope '{scope_required}' for action '{canonical_action}'", scope_required=scope_required, request=request, ) if self.audit_log: self.audit_log.record( AuditEventType.PERMISSION_CHECK, actor=agent.agent_id, agent_id=agent.agent_id, action=canonical_action, resource=resource or "", result="scope_check_passed", ) # Step 3: Evaluate policy policy_context = PolicyContext( action=canonical_action, resource=resource, environment=context or {}, risk_score=risk_score, mode=self.policy_engine.mode, ) policy_result = self.policy_engine.evaluate( agent, canonical_action, resource, policy_context ) # Step 4: Map policy decision to permission decision if policy_result.decision == PolicyDecision.DENY: if self.audit_log: self.audit_log.record( AuditEventType.PERMISSION_CHECK, actor=agent.agent_id, agent_id=agent.agent_id, action=canonical_action, resource=resource or "", policy_decision=policy_result.decision.value, result="denied", ) return PermissionResult( decision=PermissionDecision.DENY, reason=f"Policy denied action '{canonical_action}': {policy_result.reason}", policy_decision=policy_result.decision, scope_required=scope_required, request=request, ) if policy_result.decision == PolicyDecision.REQUIRE_APPROVAL: if self.audit_log: self.audit_log.record( AuditEventType.PERMISSION_CHECK, actor=agent.agent_id, agent_id=agent.agent_id, action=canonical_action, resource=resource or "", policy_decision=policy_result.decision.value, result="approval_required", ) return PermissionResult( decision=PermissionDecision.REQUIRE_APPROVAL, reason=f"Action '{canonical_action}' requires human approval: {policy_result.reason}", policy_decision=policy_result.decision, scope_required=scope_required, approval_required=True, request=request, ) # ALLOW if self.audit_log: self.audit_log.record( AuditEventType.PERMISSION_CHECK, actor=agent.agent_id, agent_id=agent.agent_id, action=canonical_action, resource=resource or "", policy_decision=policy_result.decision.value, result="allowed", ) return PermissionResult( decision=PermissionDecision.ALLOW, reason=f"Action '{canonical_action}' permitted for agent {agent.agent_id}", policy_decision=policy_result.decision, scope_required=scope_required, request=request, ) def require_scope(self, action: str) -> str | None: """Return the required scope for an action, or None if no scope mapping exists. The name is canonicalized first, so a cosmetic variant of an action resolves to the same scope instead of silently reporting "no mapping". """ return self.ACTION_SCOPE_MAP.get(normalize_action(action)) def has_scope_mapping(self, action: str) -> bool: """Check whether an action has a scope mapping.""" return normalize_action(action) in self.ACTION_SCOPE_MAP __all__ = [ "PermissionDecision", "PermissionGateway", "PermissionRequest", "PermissionResult", "SecurityError", ]