"""Authorized entry point for invoking execution backends.""" from __future__ import annotations import asyncio import hashlib from typing import Any from cyberbrain26.extension_layer.execution_provider import ExecutionProvider from cyberbrain26.security_plane.actions import ( DEFAULT_EXECUTION_TIMEOUT_SECONDS, ExecutionAuthorization, current_execution_authorization, execution_payload_digest, ) from cyberbrain26.security_plane.approval import ApprovalManager, ApprovalStatus from cyberbrain26.security_plane.audit import AuditEventType, AuditLog from cyberbrain26.security_plane.egress import ( DataClassification, EgressGate, EgressPolicy, EgressRequest, ) from cyberbrain26.security_plane.extension_points import ( EgressPermissionGate, LocalClassificationGate, RemoteClassificationGate, ) from cyberbrain26.security_plane.identity import AgentIdentity from cyberbrain26.security_plane.permission import ( PermissionDecision, PermissionGateway, SecurityError, ) # An engagement profile is carried on the authenticated identity's metadata, so # the grant travels with the identity that the gates above already validated # rather than with the step payload, which is attacker-controlled. ENGAGEMENT_PROFILE_KEY: str = "engagement_profile" NETWORK_ALLOWED_KEY: str = "network_allowed" def engagement_profile_allows_network(identity: AgentIdentity) -> bool: """Report whether this identity's engagement profile authorised egress. The grant has to be *explicit*, so the check is a strict identity test against the singleton ``True``: the string ``"true"``, the integer ``1``, a truthy value of any other type, and a missing profile all resolve to ``False``. An identity that carries no engagement profile therefore behaves exactly as it did before the profile existed - the network namespace stays unshared - and no other role changes behaviour by accident. This is the only place the sandbox is allowed to be told that sharing the host's network namespace is authorised, and it says nothing about whether the code itself is trusted; see :class:`~cyberbrain26.execution_plane.bubblewrap_provider.BubblewrapExecutionProvider`. """ metadata = getattr(identity, "metadata", None) if not isinstance(metadata, dict): return False profile = metadata.get(ENGAGEMENT_PROFILE_KEY) if not isinstance(profile, dict): return False return profile.get(NETWORK_ALLOWED_KEY) is True class ExecutionRefused(SecurityError): """Raised when a request is refused before any code reached a provider. This type is the structural marker for "refused, nothing ran". Only the three gates standing in front of ``provider.execute`` raise it: identity validation, permission denial, and a missing/denied/expired approval. Every failure at or after the provider call raises a plain ``SecurityError`` (or propagates the provider's own exception), because there the sandbox may already have executed the code and produced side effects. Consumers must tell the two cases apart structurally, by type, never by matching the error message: only a refusal is safe to re-dispatch. """ class AuthorizedExecutionService: """Enforce identity, permission, approval, and audit before execution. The underlying execution provider is a sandbox backend, not an authorization boundary. Application and agent code should invoke this service so unauthenticated calls cannot reach the backend. This is the inner of the two gates in front of the provider, and it is not a formality: scope, policy, approval and audit all run again here, on the values this call actually carries. What it no longer does is raise a *second* independent approval request for a decision a human already made. When the dispatcher's chain has already obtained an approval for the same agent, the same canonical action and a byte-identical execution payload, the grant it minted is re-verified here (identity, action, digest, and that the approval record is still ``approved``) and recorded as a reuse instead of a new request. Anything else - no grant, a different action, a changed payload, a different identity, an approval that is no longer approved - leaves this service on its own full gate, and a stale grant is refused outright rather than executed. """ def __init__( self, provider: ExecutionProvider, permission_gateway: PermissionGateway, *, approval_manager: ApprovalManager | None = None, audit_log: AuditLog | None = None, approval_timeout_seconds: float = 30.0, approval_poll_interval: float = 0.05, ) -> None: if approval_timeout_seconds <= 0 or approval_poll_interval <= 0: raise ValueError("approval timeout and poll interval must be positive") self.provider = provider self.permission_gateway = permission_gateway self.approval_manager = approval_manager self.audit_log = audit_log or permission_gateway.audit_log or AuditLog() # Built from the audit log resolved above, so the gate and the service # never disagree about where a decision is recorded. # The execution provider is a local provider by definition, so add it to # the local allowlist so that local egress is permitted. provider_name = getattr(provider, "provider_name", "unknown") egress_policy = EgressPolicy(local_provider_ids=frozenset([provider_name])) self.egress_gate = EgressGate(policy=egress_policy, audit_log=self.audit_log, base_decision=True) self.permission_gateway.audit_log = self.audit_log self.permission_gateway.policy_engine.audit_log = self.audit_log self.approval_timeout_seconds = approval_timeout_seconds self.approval_poll_interval = approval_poll_interval # New extension points for classification and permission gating self.remote_classification_gate = RemoteClassificationGate() self.local_classification_gate = LocalClassificationGate() self.egress_permission_gate = EgressPermissionGate() async def execute( self, identity: AgentIdentity, code: str, language: str, *, task_id: str, timeout_seconds: int = DEFAULT_EXECUTION_TIMEOUT_SECONDS, workspace: str | None = None, ) -> dict[str, Any]: """Authorize and run a Python or shell task through the sandbox. The approval gate below reuses the dispatcher's grant when - and only when - that grant provably covers this exact call; see :meth:`_grant_covers_this_execution`. """ if not isinstance(identity, AgentIdentity): # Gate 1 of 3 in front of the provider: nothing has executed yet. raise ExecutionRefused("a valid agent identity is required for execution") lang = language.strip().lower() if lang in {"python", "py"}: action = "execute_python" elif lang in {"shell", "sh"}: action = "execute_shell" else: # Payload validation, not an authorization refusal: keep ValueError # so callers matching on it are unaffected. raise ValueError(f"unsupported execution language: {language}") decision = self.permission_gateway.evaluate(identity, action, workspace) if decision.decision == PermissionDecision.DENY: self._audit( AuditEventType.TOOL_CALLED, task_id, identity, action, "denied", decision.reason, ) raise ExecutionRefused(decision.reason) if decision.decision == PermissionDecision.REQUIRE_APPROVAL: if not self._grant_covers_this_execution( identity, action, code, lang, workspace, timeout_seconds, task_id ): await self._wait_for_approval( identity, action, workspace, task_id, decision.reason ) # Past this point the request is authorized and the provider will run. # However, we first check the egress gate to ensure remote execution # is authorized and recorded in the audit log. allow_network = engagement_profile_allows_network(identity) # Build an EgressRequest from the available context. # provider_id comes from self.provider, destination can be inferred # from the execution context; for local execution, use loopback. provider_name = self.provider.provider_name if self.provider.provider_name else "unknown" dest = "http://localhost" # Local execution uses loopback destination payload_sha256 = hashlib.sha256(code.encode("utf-8")).hexdigest() request = EgressRequest( agent_id=identity.agent_id if identity else "unknown", provider_id=provider_name, payload_sha256=payload_sha256, # SECURITY MARKER: this classification is a hardcoded constant, not # derived from the payload. Nothing inspects `code` above, so the # audit record's classifications=["public"] proves nothing about the # sensitivity of what is about to run. Redesign tracked in plan F3. classifications=frozenset({DataClassification.PUBLIC}), destination=dest, ) try: self.egress_gate.authorize(request) except SecurityError as err: # If authorization fails, we deny the execution. raise ExecutionRefused("egress denied by policy") from err self._audit( AuditEventType.TOOL_CALLED, task_id, identity, action, "started", self.provider.provider_name, network=allow_network, ) try: result = await self.provider.execute( code, lang, timeout_seconds=timeout_seconds, workspace=workspace, allow_network=allow_network, ) except Exception as exc: self._audit( AuditEventType.TOOL_RESULT, task_id, identity, action, "failed", type(exc).__name__, ) raise successful = result.get("return_code") == 0 and not result.get("timed_out", False) self._audit( AuditEventType.TOOL_RESULT, task_id, identity, action, "succeeded" if successful else "failed", self.provider.provider_name, ) return result def _grant_covers_this_execution( self, identity: AgentIdentity, action: str, code: str, language: str, workspace: str | None, timeout_seconds: int, task_id: str, ) -> bool: """Report whether a human already approved this exact execution. The caller has *not* skipped any gate: it has already re-run the scope check and the policy engine for this identity, action and workspace, and it arrives here only because those gates still say "a human has to approve". This method answers one question - did one - by re-verifying the dispatcher's grant against what is about to be executed. The grant is honoured only when all of the following hold: * it names the same identity, so one agent's approval is not spendable by another; * it was minted for the same canonical action, so an approval for action A can never authorize action B - a grant minted for a different action is ignored here, and the caller then raises its own full approval request; * it was minted for a payload whose digest equals the digest of the code, language, workspace and timeout in this call, so content that changed after the operator approved it is refused instead of run; * the approval record it names still reads ``approved``, checked live rather than trusted from the grant. When the action matches but the payload does not, the caller approved code that is no longer the code about to run. That is a tampering signal inside a single dispatch rather than a normal condition, so it is refused here and audited, instead of quietly raising a fresh request that would make the change invisible in the audit trail. """ grant = current_execution_authorization() if grant is None or grant.approval_request_id is None: # No grant, or an outer "allow" that carries no human decision at # all. Neither may stand in for an approval. return False if grant.action != action: return False digest = execution_payload_digest( code=code, language=language, workspace=workspace, timeout_seconds=timeout_seconds, ) if not grant.covers(agent_id=identity.agent_id, action=action, payload_digest=digest): reason = ( "the approval covers a different payload than the one about to be " f"executed (approved digest {grant.payload_digest[:12]}, " f"executed digest {digest[:12]})" ) self._audit( AuditEventType.APPROVAL_DENIED, task_id, identity, action, "payload_changed_after_approval", reason, ) raise ExecutionRefused( f"Approval {grant.approval_request_id} does not cover this execution: {reason}" ) if not self._grant_approval_still_approved(grant, task_id, identity, action): return False self._audit( AuditEventType.APPROVAL_GRANTED, task_id, identity, action, "reused", f"approval {grant.approval_request_id} for task {grant.task_id} covers " f"this exact payload (digest {digest[:12]})", ) return True def _grant_approval_still_approved( self, grant: ExecutionAuthorization, task_id: str, identity: AgentIdentity, action: str, ) -> bool: """Re-read the approval record a grant names, and report whether it holds. The grant is a statement made at dispatch time; this is the check made now. A request that has since been denied, cancelled, expired or removed is not honoured, and when the approval cannot be read at all the grant is not honoured either - the caller then raises its own request, which is strictly stronger than trusting an unverifiable claim. """ manager = self.approval_manager if manager is None: return False current = manager.get_request(grant.approval_request_id or "") if current is not None and current.status == ApprovalStatus.APPROVED: return True self._audit( AuditEventType.APPROVAL_DENIED, task_id, identity, action, "grant_not_approved", f"approval {grant.approval_request_id} is " f"{current.status.value if current is not None else 'unreadable'}", ) return False async def _wait_for_approval( self, identity: AgentIdentity, action: str, resource: str | None, task_id: str, reason: str, ) -> None: manager = self.approval_manager if manager is None: self._audit( AuditEventType.APPROVAL_DENIED, task_id, identity, action, "unavailable", "approval manager is not configured", ) raise ExecutionRefused(f"Approval unavailable for action '{action}'") request = manager.create_request( identity, action, resource, context={"task_id": task_id}, reason=reason, ) self._audit( AuditEventType.APPROVAL_REQUIRED, task_id, identity, action, "pending", request.request_id, ) wait_seconds = min(self.approval_timeout_seconds, manager.timeout_seconds) deadline = asyncio.get_running_loop().time() + wait_seconds try: while asyncio.get_running_loop().time() < deadline: current = manager.get_request(request.request_id) if current is None: break if current.status == ApprovalStatus.APPROVED: self._audit( AuditEventType.APPROVAL_GRANTED, task_id, identity, action, "approved", current.reviewer or "approval granted", ) return if current.status != ApprovalStatus.PENDING: break await asyncio.sleep(min(self.approval_poll_interval, wait_seconds)) except asyncio.CancelledError: manager.cancel(request.request_id) self._audit( AuditEventType.APPROVAL_DENIED, task_id, identity, action, "cancelled", request.request_id, ) raise manager.check_expired() current = manager.get_request(request.request_id) if current is not None and current.status == ApprovalStatus.PENDING: manager.cancel(request.request_id) status = current.status.value if current is not None else "missing" self._audit( AuditEventType.APPROVAL_DENIED, task_id, identity, action, status, request.request_id, ) raise ExecutionRefused(f"Approval {status} for action '{action}'") def _audit( self, event_type: AuditEventType, task_id: str, identity: AgentIdentity, action: str, result: str, detail: str, *, network: bool | None = None, ) -> None: if self.audit_log is not None: metadata: dict[str, Any] = { "detail": detail, "provider": self.provider.provider_name, } # Recorded only where it is actually decided - the event that # starts an execution - so every other event keeps the metadata # shape it has always had and nothing has to be re-derived later. if network is not None: metadata["network"] = network self.audit_log.record( event_type, actor=identity.agent_id, agent_id=identity.agent_id, task_id=task_id, action=action, result=result, metadata=metadata, ) __all__ = [ "ENGAGEMENT_PROFILE_KEY", "NETWORK_ALLOWED_KEY", "AuthorizedExecutionService", "ExecutionRefused", "engagement_profile_allows_network", ]