diff --git a/docs/SIDECAR_CONTEXT_AUTH.md b/docs/SIDECAR_CONTEXT_AUTH.md index 5f96ea63..8fce12fd 100644 --- a/docs/SIDECAR_CONTEXT_AUTH.md +++ b/docs/SIDECAR_CONTEXT_AUTH.md @@ -175,8 +175,10 @@ missing variables in strict prompt rendering produce clear errors. ## Access Plugin -Authorization is opt-in per node. A node marked `protected: true` is hidden from -the router unless the fixed access plugin allows it. +Authorization is opt-in per node. A child marked `protected: true` is hidden +from its parent unless the fixed access plugin allows it. A protected root +agent or orchestrator passes through the same policy before its executor is +entered, since it has no parent to filter it. ```yaml agents: @@ -202,10 +204,57 @@ Rules: - Protected nodes are checked before routing. - If `can_access` returns true, the node remains a routing candidate. - If `can_access` returns false or raises, the node is hidden. +- On a new run of a protected root, denial returns `Access denied.` with + an empty visited route and no tool usage. Prompt resolvers, models, and tools + are not invoked. `/invoke` keeps its normal response shape and HTTP 200; + `/stream` emits the fixed refusal as an `answer_delta`, then a `final` event + containing the same refusal and empty route. Delta-only clients such as the + CLI therefore display the refusal too. + No model-generated answer or route event is emitted for a denied root. - `protected: true` without `plugins/access.py` is a configuration error at - startup. + startup, including when a Python host builds the engine directly. - `plugins/access.py` with no protected nodes is harmless and may be ignored. +The access resolver receives the per-request, non-secret `run_context` +projection for roots and children alike: available run, conversation, user and +organization IDs, metadata, and `auth` facts such as roles and scopes. The +inbound access token is excluded. A Python host supplies trusted `RunContext` +values; `agent_manager` derives them from its verified principal. The thin +engine API accepts no caller-asserted roles or user identity, including forged +`X-User-ID` or `X-Roles` headers. + +No universal role requirement is imposed by the engine. If the policy requires +identity or metadata that is missing or malformed, it must deny or raise; both +outcomes fail closed. An explicitly permissive policy can allow an empty +context. A root's decision is evaluated per execution, never cached between +requests. This enforces node access without implementing API authentication or +replacing authorization inside client tools and data services. + +Approval replay uses the **current approval request's** trusted context. Python +hosts pass `context=RunContext(...)` to `resume` / `resume_stream`; +`agent_manager` builds it from the verified principal on each decision request, +including that principal's current roles, organization and credential. Supplied +run/user/session identities must match the approval's bound caller and run. +The root wrapper replaces its execution-local context projection before access +checks and child delegation. Saved roles, scopes, organization and metadata are +never merged into it, and credentials are never added to the graph projection. +Omitting fresh permission facts does not recover the old grant. A permissive +policy can still explicitly allow a context with no role or scope facts. + +The root policy is rechecked before its executor is entered. If that policy +denies access, the reviewed invocation is never reached. The existing +replay-binding guard raises `ApprovalInvocationMismatch`, rejects the pending +approval and fails the run; the decision is not reported as an executed action. +Removing a protected child can instead trigger the existing checkpoint replay +mismatch guard, which also prevents execution and fails the run. Neither case +re-generates the reviewed model/tool request. A denied resume retains the +existing error handling rather than a new run's normal refusal response. + +The optional Python approval ports require implementations to accept the new +`context` keyword; a host must not silently discard it for a legacy adapter. +The HTTP decision body, SSE event types, access-plugin signature and YAML +remain unchanged. See [ADR 0002](adr/0002-conversation-lifecycle-and-immutable-branches.md). + --- ## Why Not YAML Roles? diff --git a/docs/adr/0002-conversation-lifecycle-and-immutable-branches.md b/docs/adr/0002-conversation-lifecycle-and-immutable-branches.md index d9e2d29b..bf86f67b 100644 --- a/docs/adr/0002-conversation-lifecycle-and-immutable-branches.md +++ b/docs/adr/0002-conversation-lifecycle-and-immutable-branches.md @@ -78,6 +78,13 @@ One `conversation_id` owns an immutable message tree. turns and resumed turns share the engine's owned graph-task cancellation path; resume retains the original run and transitions `PENDING_APPROVAL → RESUMING → RUNNING` before emitting work. +- Approval replay uses current authorization facts supplied by the trusted + host on the decision request. The manager derives them from its freshly + verified Principal, never from the suspended run. Root entry replaces the + execution-local context projection for root and child access checks while + preserving the reviewed checkpoint/model/tool binding. Missing permission + facts cannot inherit checkpointed roles, scopes, organization or metadata. + Credentials stay in request-local context and are excluded from the projection. No separate `branch_id` is introduced. A branch is identified by its leaf/head message, which is already a stable domain identity. @@ -91,6 +98,16 @@ message, which is already a stable domain identity. not represented as an approval decision or as `DENY`. - Approval decisions add `/decision/stream`; its `resume_started` event carries the original `run_id`, and disconnecting it cancels the active graph task. +- The Python `ApprovalEngine.resume` and `ApprovalStreamingEngine.resume_stream` + ports accept optional `context: RunContext | None`. Embedding hosts supply a + current verified snapshot on every decision; the manager forwards current + roles, organization and credential. Supplied caller/run/session identities + must agree with the existing approval binding before it is claimed. External + implementations of these optional capabilities must accept the keyword; + silently dropping context to support an older signature is unsafe. Existing + direct calls remain valid, but without fresh facts a role/scope/metadata + policy cannot reuse the original grant. No HTTP/YAML/access-plugin schema + changes or new persistence revision are required for this correction. - Conversation sessions gain `head_message_id`; manager persistence includes durable run records. Because no database environment currently exists, no new migration revision was created; the existing fresh-database baseline revision diff --git a/src/agent_engine/engine/approval_engine.py b/src/agent_engine/engine/approval_engine.py index 9a3737e3..f0e99350 100644 --- a/src/agent_engine/engine/approval_engine.py +++ b/src/agent_engine/engine/approval_engine.py @@ -6,6 +6,7 @@ from agent_engine.approvals.decision import ApprovalDecision from agent_engine.engine.types import PendingApproval, RunResult +from agent_engine.runtime.hooks.models import RunContext @runtime_checkable @@ -30,4 +31,5 @@ async def resume( caller_user_id: str | None = None, caller_session_id: str | None = None, access_token: str | None = None, + context: RunContext | None = None, ) -> RunResult: ... diff --git a/src/agent_engine/engine/approval_streaming_engine.py b/src/agent_engine/engine/approval_streaming_engine.py index 40805bda..8b077b83 100644 --- a/src/agent_engine/engine/approval_streaming_engine.py +++ b/src/agent_engine/engine/approval_streaming_engine.py @@ -6,6 +6,7 @@ from typing import Protocol, runtime_checkable from agent_engine.approvals.decision import ApprovalDecision +from agent_engine.runtime.hooks.models import RunContext from agent_engine.runtime.streaming import RunStreamEvent @@ -20,4 +21,5 @@ def resume_stream( caller_user_id: str | None = None, caller_session_id: str | None = None, access_token: str | None = None, + context: RunContext | None = None, ) -> AsyncIterator[RunStreamEvent]: ... diff --git a/src/agent_engine/engine/langgraph/engine.py b/src/agent_engine/engine/langgraph/engine.py index 915b130d..945e0a47 100644 --- a/src/agent_engine/engine/langgraph/engine.py +++ b/src/agent_engine/engine/langgraph/engine.py @@ -4,6 +4,7 @@ import logging from collections.abc import AsyncGenerator, AsyncIterator, Awaitable, Callable, Iterator, Sequence from contextlib import contextmanager, suppress +from dataclasses import replace from pathlib import Path from typing import Any, cast @@ -16,7 +17,11 @@ from agent_engine.approvals.approval_manager import ApprovalManager from agent_engine.approvals.coordinator import ApprovalCoordinator from agent_engine.approvals.decision import ApprovalDecision, parse_decision -from agent_engine.approvals.errors import ApprovalInvocationMismatch, RunNotFound +from agent_engine.approvals.errors import ( + ApprovalInvocationMismatch, + RunNotFound, + UnauthorizedApprover, +) from agent_engine.approvals.in_memory_approval_repository import InMemoryApprovalRepository from agent_engine.approvals.in_memory_session_approval_repository import ( InMemorySessionApprovalRepository, @@ -74,6 +79,7 @@ RunContext, current_run_context, ) +from agent_engine.runtime.hooks.context import public_run_context from agent_engine.runtime.streaming import ( RunStreamEvent, StreamSinks, @@ -112,36 +118,10 @@ def _initial_state( ], } if expose_run_context: - state["run_context"] = _state_run_context(ctx) + state["run_context"] = public_run_context(ctx) return state -def _state_run_context(ctx: RunContext) -> dict[str, Any]: - """Return the generic, non-secret context exposed to graph runtime filters.""" - data: dict[str, Any] = {} - for key in ("run_id", "conversation_id", "user_id", "organization_id"): - value = getattr(ctx, key) - if value is not None: - data[key] = value - if ctx.metadata: - data["metadata"] = dict(ctx.metadata) - if ctx.auth_context is not None: - auth: dict[str, Any] = {} - for key in ("user_id", "organization_id"): - value = getattr(ctx.auth_context, key) - if value is not None: - auth[key] = value - if ctx.auth_context.scopes: - auth["scopes"] = tuple(ctx.auth_context.scopes) - if ctx.auth_context.roles: - auth["roles"] = tuple(ctx.auth_context.roles) - if ctx.auth_context.metadata: - auth["metadata"] = dict(ctx.auth_context.metadata) - if auth: - data["auth"] = auth - return data - - def _trace_metadata(ctx: RunContext) -> dict[str, Any]: """Map per-run identity onto the metadata keys Langfuse reads to group traces. @@ -694,6 +674,7 @@ async def resume( caller_user_id: str | None = None, caller_session_id: str | None = None, access_token: str | None = None, + context: RunContext | None = None, ) -> RunResult: """Apply a human decision to a pending tool call and resume the same run. @@ -703,6 +684,10 @@ async def resume( either executes or denies the original tool call. ``ALLOW_FOR_SESSION`` additionally records a session permission so the tool is not re-prompted for the rest of the conversation. + + ``context`` must come from the current trusted host request. Permission + facts are replaced on replay, never recovered from the checkpoint. + With no context, role/scope/metadata-based policies see no saved grants. """ app, lifecycle = self._require_built("resuming") kind = parse_decision(decision) @@ -713,6 +698,7 @@ async def resume( caller_user_id=caller_user_id, caller_session_id=caller_session_id, access_token=access_token, + context=context, ) approved = kind != ApprovalDecision.DENY log( @@ -775,8 +761,9 @@ async def resume_stream( caller_user_id: str | None = None, caller_session_id: str | None = None, access_token: str | None = None, + context: RunContext | None = None, ) -> AsyncIterator[RunStreamEvent]: - """Resume one approval through the same owned stream used by new runs.""" + """Resume with current trusted context through the owned run stream.""" app, lifecycle = self._require_built("streaming an approval resume") kind = parse_decision(decision) ctx, approval = await self._activate_approval_resume( @@ -786,6 +773,7 @@ async def resume_stream( caller_user_id=caller_user_id, caller_session_id=caller_session_id, access_token=access_token, + context=context, ) approved = kind != ApprovalDecision.DENY @@ -834,6 +822,7 @@ async def _activate_approval_resume( caller_user_id: str | None, caller_session_id: str | None, access_token: str | None = None, + context: RunContext | None = None, ) -> tuple[RunContext, ApprovalRecord]: """Claim and activate one resume without an interruptible ownership gap. @@ -844,24 +833,38 @@ async def _activate_approval_resume( """ async def activate() -> tuple[RunContext, ApprovalRecord]: + request = context or RunContext() + auth = request.auth_context or AuthContext() + users = { + value + for value in (caller_user_id, request.user_id, auth.user_id) + if value is not None + } + sessions = { + value for value in (caller_session_id, request.conversation_id) if value is not None + } + if request.run_id not in (None, run_id) or len(users) > 1 or len(sessions) > 1: + raise UnauthorizedApprover(approval_id) approval = await self._approval_manager.claim( run_id=run_id, approval_id=approval_id, - caller_user_id=caller_user_id, - caller_auth_ref=caller_session_id, + caller_user_id=next(iter(users), None), + caller_auth_ref=next(iter(sessions), None), ) - ctx = RunContext( + ctx = request.replace( run_id=run_id, conversation_id=approval.auth_ref, user_id=approval.authorized_user_id, - organization_id=approval.organization_id, - metadata={"approval_id": approval.approval_id}, - # The approver's credential as of now, not one captured when the - # run started and possibly expired while it waited for a human. - auth_context=AuthContext( + organization_id=request.organization_id or auth.organization_id, + metadata={**request.metadata, "approval_id": approval.approval_id}, + # Authorization facts come only from this request. Saved facts + # describe the original run and cannot prove current access. + auth_context=replace( + auth, user_id=approval.authorized_user_id, - organization_id=approval.organization_id, - inbound_access_token=access_token, + inbound_access_token=access_token + if access_token is not None + else auth.inbound_access_token, ), ) await lifecycle.activate_resume(ctx) @@ -1043,9 +1046,7 @@ async def _stream_graph( def _setup_filters(self, spec: SystemSpec) -> list[RouteFilter]: filters: list[RouteFilter] = [] if has_protected_nodes(spec.graph): - access_plugin = self._base_dir / "plugins" / "access.py" - if access_plugin.is_file(): - filters.append(AccessFilter(self._base_dir)) + filters.append(AccessFilter(self._base_dir)) return filters def _build_graph(self, spec: SystemSpec) -> RunGraph: diff --git a/src/agent_engine/engine/langgraph/graph/graph_builder.py b/src/agent_engine/engine/langgraph/graph/graph_builder.py index 78bef498..63d2d123 100644 --- a/src/agent_engine/engine/langgraph/graph/graph_builder.py +++ b/src/agent_engine/engine/langgraph/graph/graph_builder.py @@ -3,6 +3,7 @@ from __future__ import annotations import inspect +import logging from collections.abc import Callable, Mapping, Sequence from pathlib import Path @@ -17,6 +18,8 @@ from agent_engine.core.spec import AgentSpec, BaseModelConfig, GraphNode, OrchestratorSpec from agent_engine.core.spec import ModelConfig as NodeModelConfig from agent_engine.engine.langgraph.checkpointing import CheckpointerHandle +from agent_engine.engine.langgraph.execution.approval_resume import current_approval_resume +from agent_engine.engine.langgraph.execution.node_executor import NodeExecutor from agent_engine.engine.langgraph.filters import RouteFilter from agent_engine.engine.langgraph.graph.traversal import node_id from agent_engine.engine.langgraph.nodes.agent_node import AgentNode @@ -27,12 +30,15 @@ from agent_engine.loaders.resolver_loader import ResolverLoader from agent_engine.loaders.tool_loader import ToolLoader from agent_engine.runtime.hooks import HookManager +from agent_engine.runtime.hooks.context import current_run_context, public_run_context from agent_engine.runtime.state import GraphState +from agent_engine.runtime.streaming import current_streams from agent_engine.tool_usage.context_provider import ToolUsageContextProvider from agent_engine.tool_usage.tracker import ToolUsageTracker ModelFactory = Callable[..., BaseChatModel] _MODEL_FACTORY_OPTIONAL_KWARGS = ("region", "max_tokens", "top_p") +logger = logging.getLogger(__name__) # noinspection PyTypeChecker RunGraphBuilder = StateGraph[GraphState, None, GraphState, GraphState] @@ -141,13 +147,37 @@ def _wire_node( parent_path: str | None, ) -> None: path = node_id(node, parent_path) + executor: NodeExecutor if isinstance(node.node, OrchestratorSpec): - orchestrator_node = self._build_orchestrator_node(node, parent_path) - builder.add_node(path, orchestrator_node.execute) + executor = self._build_orchestrator_node(node, parent_path) else: assert isinstance(node.node, AgentSpec) - agent_node = self._build_agent_node(node.node, path) - builder.add_node(path, agent_node.execute) + executor = self._build_agent_node(node.node, path) + filters = tuple(self._filters) + + async def execute_root(state: GraphState) -> GraphState: + if current_approval_resume.get() is not None: + ctx = current_run_context.get() + # Checkpoints retain the reviewed work, not current authority. + # Replace facts locally for the root and all delegated children. + state = {**state, "run_context": public_run_context(ctx) if ctx else {}} + if not node.node.protected: + return await executor.execute(state) + # The root has no parent to filter it before delegation. Apply the + # same policy before entering either kind of protected executor. + candidates = [node.node] + for route_filter in filters: + candidates = route_filter.filter(state.get("run_context", {}), candidates) + if not candidates: + logger.warning("access denied for protected root node '%s'", path) + answer = "Access denied." + sink = current_streams.get().answer + if sink is not None: + sink(answer) + return {"visited": [], "answer": answer} + return await executor.execute(state) + + builder.add_node(path, execute_root) builder.add_edge(path, END) def _build_orchestrator_node( diff --git a/src/agent_engine/runtime/hooks/context.py b/src/agent_engine/runtime/hooks/context.py index 0bdf9727..58212ad0 100644 --- a/src/agent_engine/runtime/hooks/context.py +++ b/src/agent_engine/runtime/hooks/context.py @@ -10,7 +10,34 @@ from __future__ import annotations from contextvars import ContextVar +from typing import Any from agent_engine.runtime.hooks.models import RunContext current_run_context: ContextVar[RunContext | None] = ContextVar("current_run_context", default=None) + + +def public_run_context(ctx: RunContext) -> dict[str, Any]: + """Project request facts for access filters without exposing credentials.""" + data: dict[str, Any] = {} + for key in ("run_id", "conversation_id", "user_id", "organization_id"): + value = getattr(ctx, key) + if value is not None: + data[key] = value + if ctx.metadata: + data["metadata"] = dict(ctx.metadata) + if ctx.auth_context is not None: + auth: dict[str, Any] = {} + for key in ("user_id", "organization_id"): + value = getattr(ctx.auth_context, key) + if value is not None: + auth[key] = value + if ctx.auth_context.scopes: + auth["scopes"] = tuple(ctx.auth_context.scopes) + if ctx.auth_context.roles: + auth["roles"] = tuple(ctx.auth_context.roles) + if ctx.auth_context.metadata: + auth["metadata"] = dict(ctx.auth_context.metadata) + if auth: + data["auth"] = auth + return data diff --git a/src/agent_manager/application/conversation_service.py b/src/agent_manager/application/conversation_service.py index f49ae259..ce37f88b 100644 --- a/src/agent_manager/application/conversation_service.py +++ b/src/agent_manager/application/conversation_service.py @@ -67,18 +67,20 @@ def _assistant_message_id(session_id: str, run_id: str) -> str: return uuid.uuid5(uuid.NAMESPACE_URL, f"agent-manager:{session_id}:{run_id}:assistant").hex -def _run_context(turn: PreparedConversationTurn) -> RunContext: +def _run_context( + run_id: str, session_id: str, user_id: str | None, principal: Principal +) -> RunContext: """The per-run context one turn executes under, carrying the caller's own host credential so tools act as this user rather than share a privileged key.""" return RunContext( - run_id=turn.run_id, - conversation_id=turn.session_id, - user_id=turn.user_id, + run_id=run_id, + conversation_id=session_id, + user_id=user_id, auth_context=AuthContext( - user_id=turn.user_id, - organization_id=turn.principal.organization_id, - roles=turn.principal.roles, - inbound_access_token=turn.principal.access_token, + user_id=user_id, + organization_id=principal.organization_id, + roles=principal.roles, + inbound_access_token=principal.access_token, ), ) @@ -224,7 +226,9 @@ async def send( result = await self._engine.run( turn.message, history=turn.history, - context=_run_context(turn), + context=_run_context( + turn.run_id, turn.session_id, turn.user_id, turn.principal + ), ) except asyncio.CancelledError: await self.cancel_turn(turn) @@ -458,6 +462,7 @@ async def decide_approval( caller_user_id=session.user_id, caller_session_id=session.session_id, access_token=principal.access_token, + context=_run_context(run_id, session.session_id, session.user_id, principal), ) except ApprovalAlreadyProcessed: recovered = await engine.get_processed_result( @@ -527,6 +532,7 @@ async def events() -> AsyncIterator[RunStreamEvent]: caller_user_id=session.user_id, caller_session_id=session.session_id, access_token=principal.access_token, + context=_run_context(run_id, session.session_id, session.user_id, principal), ) async with self._retaining_run_usage(conversation_id, run_id): try: @@ -607,7 +613,7 @@ async def stream_turn(self, turn: PreparedConversationTurn) -> AsyncIterator[Run engine_stream = self._engine.stream( turn.message, history=turn.history, - context=_run_context(turn), + context=_run_context(turn.run_id, turn.session_id, turn.user_id, turn.principal), ) async with self._retaining_run_usage(turn.session_id, turn.run_id): try: diff --git a/src/agentctl/chat.py b/src/agentctl/chat.py index d38a8602..01798ca7 100644 --- a/src/agentctl/chat.py +++ b/src/agentctl/chat.py @@ -246,6 +246,7 @@ async def _resolve_local_approvals( decision, caller_user_id=context.user_id if context is not None else None, caller_session_id=context.conversation_id if context is not None else None, + context=context, ) return result diff --git a/tasks/README.md b/tasks/README.md index bf34eb4e..c0a0ec07 100644 --- a/tasks/README.md +++ b/tasks/README.md @@ -33,5 +33,6 @@ finishing, Expected final report.** | 0021 | [WhatsApp pilot packaging](0021-whatsapp-pilot-packaging.md) | 0020; WhatsApp WA-06 | | PR 148 | [Atomic run leg accounting contract](pr-0148-run-leg-accounting-contract.md) | issue 140 review follow-up | | PR 148 | [Controls while sending is paused](pr-0148-paused-channel-controls.md) | WhatsApp composition review follow-up | +| Issue 151 | [Protected root access](issue-0151-protected-root-access.md) | 0006 security follow-up | Before starting any task, read `AGENTS.md` and the skill(s) the task lists. diff --git a/tasks/issue-0151-protected-root-access.md b/tasks/issue-0151-protected-root-access.md new file mode 100644 index 00000000..0977834a --- /dev/null +++ b/tasks/issue-0151-protected-root-access.md @@ -0,0 +1,316 @@ +# Issue 151 — Enforce access before protected root execution + +## Goal / authorization + +Fix [issue 151](https://github.com/extra-org/extra/issues/151), as requested by +the user. Apply the existing access resolver policy to a protected root agent +or orchestrator before any prompt resolution, model invocation, or tool work. + +## Context / read first + +AGENTS.md, .ai/README.md; project-architecture, senior-python-engineering, +testing, runtime-engine, sidecar-auth-context, architecture-review, +documentation and git-workflow; task 0006 and docs/SIDECAR_CONTEXT_AUTH.md. +Use systematic debugging, test-driven development and verification before +completion. The issue's HTTP reproduction distinguishes root bypass from +already-working protected-child filtering. + +## Scope / files allowed to change + +- src/agent_engine/engine/langgraph/graph/graph_builder.py: gate the root using + the existing request-scoped route filters before its executor is entered. +- src/agent_engine/engine/langgraph/engine.py: require the configured access + plugin when building a graph with protected nodes, including Python callers. +- tests/api/test_protected_access_e2e.py: real application regressions using a + fake model, covering root/child allow/deny and forged identity headers. +- tests/engine/test_protected_root.py: run/stream, agent/orchestrator roots, + missing/invalid policy context, resolver failures, missing access plugin, + streaming-console visibility and pending tool-approval replay. +- docs/SIDECAR_CONTEXT_AUTH.md: document root denial and identity semantics. +- This task and tasks/README.md. + +## Requirements / acceptance criteria + +- [x] Deny-all policy prevents protected root agent and orchestrator execution. +- [x] Permitted callers can execute protected roots using the same resolver + and non-secret context projection as protected children. +- [x] Missing/invalid required context and resolver errors fail closed. +- [x] Denied roots perform no prompt, model or tool work and expose no route. +- [x] Denial is a fixed refusal using existing result/SSE shapes; child + filtering and unprotected roots retain their existing behavior. +- [x] Missing access plugin is a clear startup error for protected graphs. +- [x] HTTP and engine regressions fail before the fix and pass afterwards. +- [x] Documentation reflects the implemented flow; make check passes. + +## Out of scope + +API authentication redesign, new identity claims or role rules, WhatsApp +changes, new HTTP/SSE contracts, configuration rejection of protected roots, +new persistence or dependencies, unrelated documentation cleanup, and merging. + +## Commands / expected final report + +Run the new regressions before implementation, the relevant engine/API tests, +make check, and git diff --check. Report using AGENTS.md section 9, including +test counts, the denial response, and the limits of the existing identity +boundary. Preserve the user's original checkout and work. + +## Implementation / validation result + +The compiled root entry applies the existing route filters before calling its +agent or orchestrator executor. Denial returns `Access denied.` and an empty +route using the existing result/SSE shapes. The wrapper captures startup +collaborators and receives request state as an argument; no request decision is +cached. Access plugin loading now fails at startup for a protected graph even +when a Python host bypasses the API's configuration validator. + +The initial regression run had 17 expected failures reproducing root execution +and missing-plugin acceptance, with nine existing allow/child/public behaviors +passing. Four additional prompt-plugin and per-request decision cases failed +when the root gate was temporarily bypassed, then passed with it restored. +The initial fix added 30 cases in the two test files. Its make check gate passed: +format/lint, mypy (293 files), 1,085 tests / 9 dependency warnings, and generated +stub checks. The first gate attempt caught import ordering and an unescaped +test regex; both were corrected before the passing run. All external models +are faked; storage uses temporary local paths. The final make check rerun also +passed after the wrapper's captured filters were made immutable; make validate +passed for both configured examples. git diff --check and new-file whitespace +checks passed. + +The fix preserves YAML, plugin signatures, HTTP schemas and SSE event types; +it corrects enforcement of the existing protected-node contract, so no new ADR +or migration is required. General API authentication and older project-status +documentation remain separate concerns. No commit, PR, merge or deployment +was performed during the initial implementation. + +## Post-implementation code review + +The user requested this review after the initial fix. Review date: 2026-10-22. + +### 1. Summary + +Reviewed root/child policy semantics, request isolation, startup failure, +stream consumers, lifecycle, and checkpointed approval replay. One display +regression was confirmed and corrected; no open blocking finding remains in +this scope. + +### 2. Blocking issues + +**[P2, fixed] Streaming clients could display a blank refusal.** Returning +only a final event skipped the answer sink. `agentctl run --stream` and the +local/remote streaming chat consumers render answer deltas, so the denial was +invisible. A regression using the real local chat consumer reproduced only +`Agent > ` without the refusal; both HTTP streaming cases also lacked the +expected delta. All three cases failed before correction. The denied-root +branch now emits its fixed, non-model-generated answer through the existing +request-scoped answer sink, followed by the usual final event. + +### 3. Non-blocking issues + +No additional actionable defect was confirmed in the scoped changes. Broader +API authentication and legacy project-status prose remain outside this task. + +### 4. Architecture concerns + +The guard runs before either protected root executor. It reuses the existing +access filter and keeps request decisions in invocation state; only immutable +startup filter references are captured by the compiled wrapper. The correction +uses an existing stream sink and event type. No new YAML/plugin/HTTP schema, +engine request state, persistence, dependency or ADR is introduced. + +### 5. Security concerns + +Denied new runs expose no visited path and execute no prompt plugin, model or +tool. Missing/invalid policy context and resolver failures remain fail closed. +Permitted and denied requests on one engine do not cache grants between calls. +Eight additional approval-replay cases cover root agents/orchestrators and +normal/streamed resumes: unchanged policy permits one real local tool effect; +policy revocation prevents that effect and the existing invocation-binding +guard rejects the decision, fails the run and leaves no pending approval or +recoverable success. This tests resolver policy rechecks, not a redesign of +identity propagation or the host's authentication boundary. + +### 6. Testing gaps + +The review added nine cases, bringing the two new files to 39 cases. The +streaming console regression uses the real consumer and engine; replay tests +observe a real temporary-file tool effect. Models are deterministic fakes. +Disabling the root gate made the replay-revocation regressions fail, confirming +they exercise enforcement. No confirmed coverage gap remains for this scope. + +The final make check passed: format/lint, mypy (293 files), **1,094 tests / +9 dependency warnings**, and generated-stub checks. make validate passed for +both configured examples; git diff --check passed. No warnings were suppressed. +One intervening full gate caught a stale source snapshot restored by the +mutation probe: the two HTTP streaming tests and console regression failed. +The stream correction was restored; the passing final run also verified its +source hash stayed unchanged throughout the gate. + +### 7. Suggested improvements + +No further change is required before reviewing this scoped fix for a PR. +Keep general authentication and outdated overview status documentation in +separate work. + +### 8. Final recommendation + +**Approve** the scoped fix with the streaming correction and added replay +coverage. This is a local self-review recommendation, not a GitHub approval. +No commit, push, PR, merge or deployment was performed during that review. + +## PR preparation + +On 2026-10-22, the user requested a pull request for the reviewed fix. This +authorizes committing and pushing the scoped change and opening a PR against +`main`. The PR includes the implementation, 39 regression cases, access-policy +documentation and this task record. Merging and deployment remain separate. + +Fresh pre-commit validation passed: make check (1,094 tests, 9 dependency +warnings, format/lint, mypy and generated-stub checks), make validate (both +configured examples), and git diff --check. The reviewed source hash was +unchanged throughout the gate. + +## PR 154 review follow-up — Fresh approval authorization + +The user's 2026-10-23 request to fix PR 154 authorizes addressing +https://github.com/extra-org/extra/pull/154#discussion_r4360627342 and updating +the existing branch. The root guard currently reads checkpointed roles; the +manager forwards only the new credential on approval. A policy that reads +roles or scopes can therefore accept a revoked grant. + +Scope additions required for this correction: + +- engine/approval_engine.py and engine/approval_streaming_engine.py: an optional + typed RunContext on the existing Python approval ports, documented in ADR 0002. +- engine/langgraph/engine.py: use fresh context while preserving approval-owner + binding; missing fresh permission facts must never fall back to checkpoints. +- engine/langgraph/graph/graph_builder.py and runtime/hooks/context.py: project + the active request's non-secret context into root/child execution on replay. +- agent_manager/application/conversation_service.py: supply the current verified + Principal for both approval paths using the existing RunContext model. +- tests/api/test_protected_access_e2e.py and tests/engine/test_protected_root.py: + real HTTP and engine revocation/allow/omitted-context regressions. +- tests/agent_manager/test_api.py: update approval doubles to the additive port. +- tests/api/test_run_context.py: follow the shared non-secret projection helper + when it moves out of the engine module; keep the credential assertion intact. +- docs/SIDECAR_CONTEXT_AUTH.md, ADR 0002 and this task: describe semantics, + compatibility, scope and verification. No HTTP/YAML/plugin schema change. + +Acceptance criteria: + +- [x] Both approval endpoints use the current verified principal's roles. +- [x] Protected agents/orchestrators and children reject revoked permissions + before the pending tool effect, including streaming replay. +- [x] Fresh scope/metadata facts replace old authorization facts; missing facts + do not inherit a grant from the original run. +- [x] Still-authorized approvals execute the reviewed tool exactly once. +- [x] Owner binding, credential forwarding, checkpoint replay and cancellation + remain intact; request context is never stored on the shared engine. +- [x] The reviewed symptom fails before the fix; make check and examples pass. + +Merged current main (6b9ada3) after resolving the task-index conflict by +retaining both sets of entries. The combined baseline passed make check: +1,498 tests / 27 dependency warnings, format/lint, mypy and generated stubs. + +### Follow-up result and validation + +The manager now sends a current RunContext through both typed approval ports. +The engine validates caller/run/session consistency before claiming an +approval and binds the active request context to the original approved owner. +Only current role/scope/organization/metadata facts enter access checks. The +shared non-secret projection helper is reused on initial execution and replay; +root entry copies state locally so child filters receive the same fresh facts. +No checkpoint edit, model regeneration, global identity state or persistence +migration is required. ADR 0002 records the additive Python keyword and adapter +upgrade requirement; HTTP/YAML/access-plugin shapes are unchanged. + +The valid initial regression run reproduced six HTTP revoked-role cases that +executed the tool, plus two omitted-context resumes that reused old roles; +the six still-authorized HTTP cases already passed. An earlier HTTP test setup +used the engine's free-text decision alias instead of the HTTP enum; that was +corrected before observing the security failures. The final suite adds 40 cases +in total, including scope, metadata, organization, ownership and concurrent +approval isolation; the two protected-access files now contain 79 cases. + +Final make check passed: **1,538 tests / 27 dependency warnings**, 363 files +formatted, lint, mypy (359 files) and generated-stub checks. Both make validate +examples and the WhatsApp pilot configuration passed; git diff --check passed. +An intermediate gate caught stale helper imports and test typing after adding +the keyword; these were corrected before the passing gate. No warnings were +suppressed, and all models/providers are fakes with temporary local storage. + +### Follow-up code review + +1. **Summary:** The checkpointed-authority bypass is corrected for root and + child access checks on normal and streamed approval replay. +2. **Blocking issues:** The reviewed revoked-role tool effect is fixed; no + additional blocking finding was confirmed in the scoped changes. +3. **Non-blocking issues:** Removing a protected child can reach the existing + checkpoint mismatch error path (sanitized HTTP 500 or SSE error), while root + denial uses the existing approval mismatch response. Both stop the effect + and fail the run; changing those public error policies is separate work. +4. **Architecture concerns:** The manager supplies verified identity through + typed ports; the engine remains generic, startup-scoped and free of request + state. Replayed work and current authority remain separate. +5. **Security concerns:** Missing fresh facts never inherit old grants. Caller + identity conflicts fail before claiming the approval. Token exclusion, + original invocation binding and independent concurrent contexts are covered. +6. **Testing gaps:** Real manager HTTP verification and engine replay use fake + models and observable local tool effects. The full gate covers existing + cancellation, duplicate decisions, credential forwarding and nested replay. + No live identity/provider deployment was exercised. +7. **Suggested improvements:** External approval adapters must accept the + new context keyword; hosts must supply a trusted current snapshot. +8. **Final recommendation:** Approve the scoped correction after independent + review. This is a local self-review, not a GitHub approval or merge. + +## PR 154 local chat follow-up (2026-10-24) + +The user requested fixing the current PR review, including +https://github.com/extra-org/extra/pull/154#discussion_r4361338882. +Local chat receives a host-supplied RunContext but drops it at the approval +port. This makes an authorized protected root fail on local HITL replay. + +Scope: src/agentctl/chat.py forwards its existing context on every local +approval; tests/cli/test_chat_command.py adds real-engine regressions for both +local stream modes and updates its approval double to the existing port; +this task records scope, verification and self-review. No new contract or +identity source, checkpoint fallback, permission relaxation or migration. + +Acceptance criteria: + +- [x] Authorized local chat approvals retain current role/scope/metadata facts + and execute the pending tool once, with stream enabled or disabled. +- [x] Protected root agents and orchestrators are covered. +- [x] Revoking a metadata permission at the approval prompt still prevents the + tool effect, using current context rather than checkpointed facts. +- [x] Credentials remain absent from policy projections and console output. +- [x] Existing role-revocation HTTP/engine regressions and make check pass. + +After validation, commit and push this scoped correction to the existing PR +branch and verify the remote head in the final report. + +Validation: all eight new cases failed before the production change. Four +authorized cases failed to resume; four revoked cases showed that the actual +current metadata was omitted. All eight pass after forwarding context. The +focused local-chat/API/engine set passed, and make check passed **1,546 tests / +27 dependency warnings**, 363 formatting checks, lint, mypy (359 files) and +generated-stub checks. make validate and the WhatsApp pilot example passed; +git diff --check passed. Tests use fake models and observable temporary-file +tool effects. No live provider or development database was used. + +Local self-review: + +1. **Summary:** The local approval adapter now carries its host-supplied context + through the existing typed port in both console modes. +2. **Blocking issues:** The reported local HITL regression is fixed. +3. **Non-blocking issues:** No new actionable finding in this scoped change. +4. **Architecture concerns:** No new contract, shared request state or runtime + policy is introduced; the engine still enforces access. +5. **Security concerns:** Changed permission facts are observed before the tool + effect; credentials stay outside policy projections and console output. +6. **Testing gaps:** Both root kinds and console modes cover allow/revoke with + a real engine and fake models; existing revocation regressions also pass. +7. **Suggested improvements:** None required for this correction. +8. **Final recommendation:** Approve the scoped fix after independent review; + this is not a GitHub approval, merge or deployment. diff --git a/tests/agent_manager/test_api.py b/tests/agent_manager/test_api.py index 3f3c0195..2fd0e59a 100644 --- a/tests/agent_manager/test_api.py +++ b/tests/agent_manager/test_api.py @@ -130,6 +130,7 @@ async def resume( caller_user_id: str | None = None, caller_session_id: str | None = None, access_token: str | None = None, + context: RunContext | None = None, ) -> RunResult: self.resume_calls.append((run_id, approval_id, decision, caller_user_id, caller_session_id)) self.resume_tokens.append(access_token) @@ -153,6 +154,7 @@ async def resume_stream( caller_user_id: str | None = None, caller_session_id: str | None = None, access_token: str | None = None, + context: RunContext | None = None, ) -> AsyncIterator[RunStreamEvent]: self.resume_calls.append((run_id, approval_id, decision, caller_user_id, caller_session_id)) self.resume_tokens.append(access_token) @@ -855,6 +857,7 @@ async def resume_stream( caller_user_id: str | None = None, caller_session_id: str | None = None, access_token: str | None = None, + context: RunContext | None = None, ) -> AsyncIterator[RunStreamEvent]: del approval_id, decision, caller_user_id, caller_session_id, access_token await runs.transition_if_allowed(run_id, RunStatus.RESUMING) @@ -1175,6 +1178,7 @@ async def resume( caller_user_id: str | None = None, caller_session_id: str | None = None, access_token: str | None = None, + context: RunContext | None = None, ) -> RunResult: del run_id, approval_id, decision, caller_user_id, caller_session_id, access_token raise RuntimeError("private approval failure") @@ -1211,6 +1215,7 @@ async def resume( caller_user_id: str | None = None, caller_session_id: str | None = None, access_token: str | None = None, + context: RunContext | None = None, ) -> RunResult: del run_id, decision, caller_user_id, caller_session_id, access_token raise ApprovalNotFound(f"private-{approval_id}") @@ -1246,6 +1251,7 @@ async def resume( caller_user_id: str | None = None, caller_session_id: str | None = None, access_token: str | None = None, + context: RunContext | None = None, ) -> RunResult: if run_id in self.completed_results: raise ApprovalAlreadyProcessed(approval_id, "approved") @@ -1256,6 +1262,7 @@ async def resume( caller_user_id=caller_user_id, caller_session_id=caller_session_id, access_token=access_token, + context=context, ) class FailFirstAssistantWrite(MemoryRepository): diff --git a/tests/api/test_protected_access_e2e.py b/tests/api/test_protected_access_e2e.py new file mode 100644 index 00000000..215a3032 --- /dev/null +++ b/tests/api/test_protected_access_e2e.py @@ -0,0 +1,216 @@ +"""Issue 151: exercise protected root and child access through the real HTTP app.""" + +from __future__ import annotations + +import json +from dataclasses import replace +from pathlib import Path +from typing import Any + +import pytest +from fastapi.testclient import TestClient +from httpx import ASGITransport, AsyncClient +from tests.agent_manager.conftest import bearer, build_test_app +from tests.approvals.test_engine_hitl import _multi_tool_spec +from tests.approvals.test_resume_binding import write_tool +from tests.engine.test_engine_flow import orchestrator, system +from tests.fixtures.utils import FakeChatModel, fake_model_factory + +from agent_engine.api.app import create_app +from agent_engine.approvals.decision import ApprovalDecision +from agent_engine.approvals.models import RunStatus +from agent_engine.engine.langgraph.engine import LangGraphEngine +from agent_manager.application import ConversationService +from agent_manager.infrastructure.persistence.memory_repository import MemoryRepository + +_FORGED_HEADERS = { + "X-User-ID": "admin", + "X-Roles": "admin", + "Authorization": "Bearer forged-admin-token", +} +_MODEL = " model: {provider: openai, name: test-model}\n" + + +def _client( + base_dir: Path, + monkeypatch: pytest.MonkeyPatch, + *, + root_kind: str = "agent", + protected_root: bool = True, + allow: bool = False, +) -> TestClient: + plugins = base_dir / "plugins" + plugins.mkdir() + (plugins / "access.py").write_text( + "class AccessResolver:\n" + " def can_access(self, ctx, node_id):\n" + f" return {allow!r}\n", + encoding="utf-8", + ) + if root_kind == "agent": + declarations = ( + "agents:\n admin:\n description: Administrative agent.\n" + f" protected: {str(protected_root).lower()}\n{_MODEL}" + ) + graph = "graph:\n admin:\n" + else: + (base_dir / "route.md").write_text("Delegate to admin.", encoding="utf-8") + declarations = ( + "orchestrators:\n root:\n description: Routes requests.\n" + f" protected: {str(protected_root).lower()}\n" + f" prompts: {{orchestrator: route.md}}\n{_MODEL}" + f"agents:\n admin:\n description: Administrative agent.\n{_MODEL}" + " protected: true\n" + ) + graph = "graph:\n root:\n admin:\n" + config = base_dir / "agents.yaml" + config.write_text(f"system: {{name: Access Probe}}\n{declarations}{graph}", encoding="utf-8") + # Keep startup, validation, graph compilation and request execution real; + # replace only model construction at the engine's existing injection seam. + monkeypatch.setattr( + "agent_engine.api.app.LangGraphEngine", + lambda *args, **kwargs: LangGraphEngine(*args, model_factory=fake_model_factory, **kwargs), + ) + return TestClient(create_app(str(config))) + + +@pytest.fixture +def model_calls(monkeypatch: pytest.MonkeyPatch) -> list[list[Any]]: + calls: list[list[Any]] = [] + respond = FakeChatModel._respond + + def record(model: FakeChatModel, messages: list[Any]): + calls.append(messages) + return respond(model, messages) + + monkeypatch.setattr(FakeChatModel, "_respond", record) + return calls + + +@pytest.mark.parametrize("root_kind", ["agent", "orchestrator"]) +@pytest.mark.parametrize("endpoint", ["/invoke", "/stream"]) +def test_denied_protected_root_never_executes_over_http( + tmp_path, monkeypatch, model_calls, root_kind, endpoint +): + with _client(tmp_path, monkeypatch, root_kind=root_kind) as client: + response = client.post(endpoint, json={"message": "admin"}, headers=_FORGED_HEADERS) + + assert response.status_code == 200 + if endpoint == "/invoke": + body = response.json() + assert body["visited"] == [] + assert body["answer"] == "Access denied." + assert body["used_tools"] == [] + else: + events = [ + json.loads(line[6:]) for line in response.text.splitlines() if line.startswith("data: ") + ] + assert [event["type"] for event in events] == ["answer_delta", "final"] + assert events[0]["content"] == "Access denied." + assert events[1]["route"] == [] + assert events[1]["content"] == "Access denied." + assert model_calls == [] + + +@pytest.mark.parametrize("root_kind", ["agent", "orchestrator"]) +def test_protected_root_executes_when_policy_allows(tmp_path, monkeypatch, root_kind): + with _client(tmp_path, monkeypatch, root_kind=root_kind, allow=True) as client: + response = client.post("/invoke", json={"message": "admin"}) + + assert response.status_code == 200 + assert response.json()["visited"] == ( + ["admin"] if root_kind == "agent" else ["root", "root/admin"] + ) + + +@pytest.mark.parametrize("allow", [False, True]) +def test_protected_child_still_obeys_policy_with_forged_headers(tmp_path, monkeypatch, allow): + with _client( + tmp_path, monkeypatch, root_kind="orchestrator", protected_root=False, allow=allow + ) as client: + response = client.post("/invoke", json={"message": "admin"}, headers=_FORGED_HEADERS) + + assert response.status_code == 200 + assert response.json()["visited"] == (["root", "root/admin"] if allow else ["root"]) + + +def test_unprotected_root_still_executes_with_deny_all_policy(tmp_path, monkeypatch): + with _client(tmp_path, monkeypatch, protected_root=False) as client: + response = client.post("/invoke", json={"message": "hello"}) + + assert response.status_code == 200 + assert response.json()["visited"] == ["admin"] + assert response.json()["answer"] == "ok" + + +@pytest.mark.parametrize("location", ["agent", "orchestrator", "child"]) +@pytest.mark.parametrize("streaming", [False, True]) +@pytest.mark.parametrize("permitted", [False, True]) +async def test_manager_approval_uses_current_verified_roles( + tmp_path, location, streaming, permitted +): + output = tmp_path / "executed.log" + write_tool(tmp_path, "send_email", output) + (tmp_path / "plugins" / "access.py").write_text( + "class AccessResolver:\n" + " def can_access(self, ctx, node_id):\n" + " return 'admin' in ctx.get('auth', {}).get('roles', ())\n", + encoding="utf-8", + ) + spec = _multi_tool_spec("send_email") + if location in {"agent", "child"}: + spec = replace( + spec, graph=replace(spec.graph, node=replace(spec.graph.node, protected=True)) + ) + if location != "agent": + spec = system(orchestrator("root", [spec.graph])) + if location == "orchestrator": + spec = replace( + spec, graph=replace(spec.graph, node=replace(spec.graph.node, protected=True)) + ) + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(spec) + service = ConversationService(engine, MemoryRepository()) + app = build_test_app(service) + initial = bearer("owner", roles=["admin"], org_id="org") + current = bearer("owner", roles=["admin"] if permitted else [], org_id="org") + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + response = await client.post( + "/conversations", json={"session_id": "session"}, headers=initial + ) + assert response.status_code == 200 + response = await client.post( + "/conversations/session/messages", json={"message": "hello"}, headers=initial + ) + assert response.status_code == 200 + pending = response.json()["pending_approval"] + assert pending is not None and not output.exists() + suffix = "/stream" if streaming else "" + response = await client.post( + f"/conversations/session/runs/{pending['run_id']}/approvals/" + f"{pending['approval_id']}/decision{suffix}", + json={"decision": ApprovalDecision.ALLOW_ONCE.value}, + headers={**current, "X-Roles": "admin", "X-User-ID": "owner"}, + ) + + if permitted: + assert response.status_code == 200 + assert len(output.read_text().splitlines()) == 1 + assert await engine.get_run_status(pending["run_id"]) == RunStatus.COMPLETED + else: + assert not output.exists() + if streaming: + assert response.status_code == 200 + events = [ + json.loads(line[6:]) + for line in response.text.splitlines() + if line.startswith("data: ") and line != "data: [DONE]" + ] + assert any(event["type"] == "error" for event in events) + assert not any(event["type"] == "final" for event in events) + else: + # Removing a child changes the replayed model tool set, which + # the existing checkpoint guard rejects before tool execution. + assert response.status_code == (500 if location == "child" else 409) + assert await engine.get_run_status(pending["run_id"]) == RunStatus.FAILED + assert await engine.get_pending_approval(pending["run_id"]) is None diff --git a/tests/api/test_run_context.py b/tests/api/test_run_context.py index 0759f872..824587f5 100644 --- a/tests/api/test_run_context.py +++ b/tests/api/test_run_context.py @@ -4,7 +4,7 @@ from __future__ import annotations from agent_engine.api.app import _run_context -from agent_engine.engine.langgraph.engine import _state_run_context +from agent_engine.runtime.hooks.context import public_run_context TOKEN = "eyJhbGciOiJIUzI1NiJ9.host-token.signature" @@ -35,4 +35,4 @@ def test_the_credential_never_reaches_the_state_the_model_sees() -> None: """Graph state reaches prompts and callers; a credential there is a leak.""" ctx = _run_context(None, run_id="r1", authorization=f"Bearer {TOKEN}") - assert TOKEN not in repr(_state_run_context(ctx)) + assert TOKEN not in repr(public_run_context(ctx)) diff --git a/tests/cli/test_chat_command.py b/tests/cli/test_chat_command.py index 593bd7cf..e582ac72 100644 --- a/tests/cli/test_chat_command.py +++ b/tests/cli/test_chat_command.py @@ -1,13 +1,15 @@ """Tests for the interactive ``agentctl chat`` console (local + remote modes). -No real LLM, MCP, or network is touched: the local engine is a fake stand-in and -the remote server is an httpx ``MockTransport``. +No real LLM, MCP, or network is touched: local tests use fake engines or the +real engine with fake models, and remote tests use httpx ``MockTransport``. """ from __future__ import annotations import json from collections.abc import AsyncIterator, Callable, Sequence +from dataclasses import replace +from pathlib import Path import httpx import pytest @@ -16,12 +18,17 @@ from agent_engine.approvals.decision import ApprovalDecision from agent_engine.approvals.models import RunStatus from agent_engine.engine.engine import Engine +from agent_engine.engine.langgraph.engine import LangGraphEngine from agent_engine.engine.types import ChatMessage, PendingApproval, RunResult -from agent_engine.runtime.hooks.models import RunContext +from agent_engine.runtime.hooks.models import AuthContext, RunContext from agent_engine.runtime.streaming import RunStreamEvent from agentctl import chat as chat_mod from agentctl.chat import run_chat_loop, run_remote_chat from agentctl.main import cli +from tests.approvals.test_engine_hitl import _multi_tool_spec +from tests.approvals.test_resume_binding import write_tool +from tests.engine.test_engine_flow import orchestrator, system +from tests.fixtures.utils import fake_model_factory # --------------------------------------------------------------------------- # fakes / helpers @@ -157,8 +164,9 @@ async def resume( caller_user_id: str | None = None, caller_session_id: str | None = None, access_token: str | None = None, + context: RunContext | None = None, ) -> RunResult: - del caller_session_id, access_token + del caller_session_id, access_token, context assert isinstance(decision, ApprovalDecision) self.resume_calls.append((run_id, approval_id, decision, caller_user_id)) return next(self.results) @@ -540,6 +548,78 @@ async def test_local_stream_prompts_and_resumes_pending_approval() -> None: assert "Agent > stream resumed" in echo.lines +@pytest.mark.parametrize("stream", [False, True]) +@pytest.mark.parametrize("root_kind", ["agent", "orchestrator"]) +@pytest.mark.parametrize("revoke", [False, True]) +async def test_local_approval_retains_current_authorization( + tmp_path: Path, capsys: pytest.CaptureFixture[str], stream: bool, root_kind: str, revoke: bool +) -> None: + output = tmp_path / "executed.log" + policy = tmp_path / "policy.jsonl" + write_tool(tmp_path, "send_email", output) + (tmp_path / "plugins" / "access.py").write_text( + "import json\n" + "class AccessResolver:\n" + " def can_access(self, ctx, node_id):\n" + f" with open({str(policy)!r}, 'a') as output:\n" + " output.write(json.dumps(ctx) + '\\n')\n" + " auth = ctx.get('auth', {})\n" + " return ('admin' in auth.get('roles', ())\n" + " and 'write' in auth.get('scopes', ())\n" + " and ctx.get('metadata', {}).get('permission') is True)\n", + encoding="utf-8", + ) + spec = _multi_tool_spec("send_email") + if root_kind == "orchestrator": + spec = system(orchestrator("root", [spec.graph])) + spec = replace(spec, graph=replace(spec.graph, node=replace(spec.graph.node, protected=True))) + context = RunContext( + run_id="local-run", + user_id="owner", + conversation_id="local-session", + metadata={"permission": True}, + auth_context=AuthContext( + user_id="owner", + roles=("admin",), + scopes=("write",), + inbound_access_token="local-test-credential", + ), + ) + inputs = scripted_reader(["hello", "o"]) + + def read_line(prompt: str) -> str: + if prompt.startswith("Allow?") and revoke: + # Metadata is the explicitly mutable host/hook enrichment field. + context.metadata["permission"] = False + return inputs(prompt) + + echo = CollectingEcho() + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(spec) + await run_chat_loop( + engine, stream=stream, context=context, read_line=read_line, echo=echo.write + ) + if revoke: + assert not output.exists() + assert await engine.get_run_status("local-run") == RunStatus.FAILED + assert await engine.get_pending_approval("local-run") is None + else: + assert not any(line.startswith("✗") for line in echo.lines), echo.lines + assert len(output.read_text().splitlines()) == 1 + assert await engine.get_run_status("local-run") == RunStatus.COMPLETED + assert any(line.startswith("Agent > ") for line in echo.lines) + decisions = policy.read_text() + latest = json.loads(decisions.splitlines()[-1]) + assert latest["metadata"]["permission"] is (not revoke) + assert latest["auth"]["roles"] == ["admin"] + assert latest["auth"]["scopes"] == ["write"] + assert latest["user_id"] == "owner" + assert latest["conversation_id"] == "local-session" + assert "local-test-credential" not in decisions + captured = capsys.readouterr() + assert "local-test-credential" not in "\n".join(echo.lines) + captured.out + captured.err + + # --------------------------------------------------------------------------- # remote mode: endpoint, error handling, stream flag # --------------------------------------------------------------------------- diff --git a/tests/engine/test_protected_root.py b/tests/engine/test_protected_root.py new file mode 100644 index 00000000..8114c50b --- /dev/null +++ b/tests/engine/test_protected_root.py @@ -0,0 +1,424 @@ +"""Protected roots use the existing access policy before entering their executor.""" + +from __future__ import annotations + +import asyncio +import json +from dataclasses import replace +from pathlib import Path + +import pytest + +from agent_engine.approvals.errors import ApprovalInvocationMismatch, UnauthorizedApprover +from agent_engine.approvals.models import RunStatus +from agent_engine.core.spec import ResolverSpec +from agent_engine.engine.langgraph.engine import LangGraphEngine +from agent_engine.runtime.hooks import AuthContext, RunContext +from tests.approvals.test_engine_hitl import _multi_tool_spec +from tests.approvals.test_resume_binding import write_tool +from tests.engine.test_engine_flow import ( + agent, + orchestrator, + system, + write_access, + write_context_access, + write_raising_access, +) +from tests.fixtures.utils import fake_model_factory + + +def _protected_root(kind): + root = agent("admin", protected=True) + if kind == "orchestrator": + root = orchestrator("admin", [agent("public")]) + root = replace(root, node=replace(root.node, protected=True)) + return system(root) + + +@pytest.mark.parametrize("kind", ["agent", "orchestrator"]) +@pytest.mark.parametrize("streaming", [False, True]) +@pytest.mark.parametrize("context_kind", ["missing", "invalid", "allowed"]) +async def test_root_policy_uses_request_context(tmp_path, kind, streaming, context_kind): + write_context_access(tmp_path) + context = None + if context_kind != "missing": + context = RunContext( + metadata={"allowed_nodes": ("admin",) if context_kind == "allowed" else None}, + auth_context=AuthContext( + inbound_access_token="test-only-secret", metadata={"groups": ("docs",)} + ), + ) + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(_protected_root(kind)) + if streaming: + events = [event async for event in engine.stream("hello", context=context)] + result = events[-1] + visited, answer = list(result.route or ()), result.content + else: + response = await engine.run("hello", context=context) + visited, answer = response.visited, response.answer + + if context_kind == "allowed": + assert visited == (["admin"] if kind == "agent" else ["admin", "admin/public"]) + assert answer == "ok" + else: + assert visited == [] + assert answer == "Access denied." + + +@pytest.mark.parametrize("kind", ["agent", "orchestrator"]) +@pytest.mark.parametrize("exc", ["RuntimeError('private-policy-error')", "NotImplementedError"]) +async def test_root_denied_when_access_resolver_raises(tmp_path: Path, kind, exc): + write_raising_access(tmp_path, exc=exc) + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(_protected_root(kind)) + result = await engine.run("hello") + + assert result.visited == [] + assert result.answer == "Access denied." + assert result.used_tools == () + + +async def test_protected_root_requires_access_plugin_for_python_callers(tmp_path: Path): + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + with pytest.raises(FileNotFoundError, match=r"access\.py"): + await engine.build(_protected_root("agent")) + + +@pytest.mark.parametrize("kind", ["agent", "orchestrator"]) +async def test_denied_root_does_not_resolve_prompt_plugins(tmp_path: Path, kind): + write_access(tmp_path, allow=False) + accessed = tmp_path / "prompt-accessed" + resolvers = tmp_path / "plugins" / "resolvers" + resolvers.mkdir() + (resolvers / "admin.py").write_text( + "from pathlib import Path\n" + "class Resolver:\n" + " def private_data(self, ctx):\n" + f" Path({str(accessed)!r}).write_text('accessed')\n" + " return 'private prompt data'\n", + encoding="utf-8", + ) + spec = _protected_root(kind) + spec = replace( + spec, + graph=replace( + spec.graph, + node=replace(spec.graph.node, resolvers=(ResolverSpec("private_data", "agent"),)), + ), + ) + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(spec) + result = await engine.run("hello") + + assert not accessed.exists() + assert result.visited == [] + assert result.answer == "Access denied." + + +@pytest.mark.parametrize("kind", ["agent", "orchestrator"]) +async def test_root_access_decision_is_evaluated_for_each_request(tmp_path: Path, kind): + write_context_access(tmp_path) + permitted = RunContext( + metadata={"allowed_nodes": ("admin",)}, + auth_context=AuthContext(metadata={"groups": ("docs",)}), + ) + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(_protected_root(kind)) + first = await engine.run("hello", context=permitted) + denied = await engine.run("hello") + last = await engine.run("hello", context=permitted) + + expected = ["admin"] if kind == "agent" else ["admin", "admin/public"] + assert first.visited == last.visited == expected + assert denied.visited == [] + assert denied.answer == "Access denied." + + +async def test_denial_reaches_the_streaming_chat_console(tmp_path: Path, capsys): + from agentctl.chat import _answer_local_stream + + write_access(tmp_path, allow=False) + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(_protected_root("agent")) + await _answer_local_stream( + engine, "hello", context=None, read_line=lambda _: "", echo=print + ) + + assert "Access denied." in capsys.readouterr().out + + +@pytest.mark.parametrize("kind", ["agent", "orchestrator"]) +@pytest.mark.parametrize("streaming", [False, True]) +@pytest.mark.parametrize("permitted", [False, True]) +async def test_root_policy_is_rechecked_before_approval_replay( + tmp_path: Path, kind, streaming, permitted +): + allowed = tmp_path / "allowed" + allowed.touch() + write_access(tmp_path, allow=True) + (tmp_path / "plugins" / "access.py").write_text( + "from pathlib import Path\n" + "class AccessResolver:\n" + " def can_access(self, ctx, node_id):\n" + f" return Path({str(allowed)!r}).exists()\n", + encoding="utf-8", + ) + output = tmp_path / "executed.log" + write_tool(tmp_path, "send_email", output) + spec = _multi_tool_spec("send_email") + if kind == "orchestrator": + spec = system(orchestrator("root", [spec.graph])) + spec = replace(spec, graph=replace(spec.graph, node=replace(spec.graph.node, protected=True))) + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(spec) + pending = await engine.run( + "hello", context=RunContext(run_id="protected-run", user_id="user", conversation_id="s") + ) + assert pending.pending_approval is not None + assert not output.exists() + approval_id = pending.pending_approval.approval_id + if not permitted: + allowed.unlink() + + async def resume(): + if streaming: + return [ + event + async for event in engine.resume_stream( + "protected-run", + approval_id, + "allow once", + caller_user_id="user", + caller_session_id="s", + ) + ] + return await engine.resume( + "protected-run", + approval_id, + "allow once", + caller_user_id="user", + caller_session_id="s", + ) + + if permitted: + await resume() + assert len(output.read_text().splitlines()) == 1 + assert await engine.get_run_status("protected-run") == RunStatus.COMPLETED + else: + # Root denial prevents reaching the reviewed tool. The existing + # replay-binding guard rejects that decision rather than falsely + # reporting that the approved invocation was executed. + with pytest.raises(ApprovalInvocationMismatch): + await resume() + assert not output.exists() + assert await engine.get_run_status("protected-run") == RunStatus.FAILED + assert await engine.get_pending_approval("protected-run") is None + assert ( + await engine.get_processed_result( + "protected-run", approval_id, caller_user_id="user", caller_session_id="s" + ) + is None + ) + + +@pytest.mark.parametrize("streaming", [False, True]) +async def test_approval_without_fresh_context_cannot_reuse_checkpoint_roles(tmp_path, streaming): + output = tmp_path / "executed.log" + write_tool(tmp_path, "send_email", output) + (tmp_path / "plugins" / "access.py").write_text( + "class AccessResolver:\n" + " def can_access(self, ctx, node_id):\n" + " return 'admin' in ctx.get('auth', {}).get('roles', ())\n", + encoding="utf-8", + ) + spec = _multi_tool_spec("send_email") + spec = replace(spec, graph=replace(spec.graph, node=replace(spec.graph.node, protected=True))) + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(spec) + pending = await engine.run( + "hello", context=RunContext(run_id="r", auth_context=AuthContext(roles=("admin",))) + ) + assert pending.pending_approval is not None + with pytest.raises(ApprovalInvocationMismatch): + if streaming: + _ = [ + event + async for event in engine.resume_stream( + "r", pending.pending_approval.approval_id, "allow once" + ) + ] + else: + await engine.resume("r", pending.pending_approval.approval_id, "allow once") + assert not output.exists() + + +@pytest.mark.parametrize("fact", ["scopes", "metadata", "auth_metadata", "organization"]) +@pytest.mark.parametrize("streaming", [False, True]) +@pytest.mark.parametrize("permitted", [False, True]) +async def test_approval_authorization_facts_are_fresh(tmp_path, fact, streaming, permitted): + output = tmp_path / "executed.log" + policy_calls = tmp_path / "policy.jsonl" + write_tool(tmp_path, "send_email", output) + conditions = { + "scopes": "'write' in ctx.get('auth', {}).get('scopes', ())", + "metadata": "ctx.get('metadata', {}).get('permission') == 'write'", + "auth_metadata": "ctx.get('auth', {}).get('metadata', {}).get('permission') == 'write'", + "organization": "ctx.get('auth', {}).get('organization_id') == 'org'", + } + (tmp_path / "plugins" / "access.py").write_text( + "import json\n" + "class AccessResolver:\n" + " def can_access(self, ctx, node_id):\n" + f" with open({str(policy_calls)!r}, 'a') as output:\n" + " output.write(json.dumps(ctx) + '\\n')\n" + f" return {conditions[fact]}\n", + encoding="utf-8", + ) + spec = _multi_tool_spec("send_email") + spec = replace(spec, graph=replace(spec.graph, node=replace(spec.graph.node, protected=True))) + initial = RunContext( + run_id="r", + user_id="user", + conversation_id="s", + metadata={"permission": "write"}, + auth_context=AuthContext( + scopes=("write",), + organization_id="org", + metadata={"permission": "write"}, + inbound_access_token="old-secret", + ), + ) + # The resume host supplies a complete current snapshot; omitted facts must + # not be filled from the original run, even when the checkpoint has them. + current = initial.replace( + metadata={"permission": "write"} if permitted else {}, + auth_context=AuthContext( + scopes=("write",) if permitted else (), + organization_id="org" if permitted else "other", + metadata={"permission": "write"} if permitted else {}, + inbound_access_token="new-secret", + ), + ) + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(spec) + pending = await engine.run("hello", context=initial) + assert pending.pending_approval is not None + approval_id = pending.pending_approval.approval_id + + async def resume(): + if streaming: + return [ + event + async for event in engine.resume_stream( + "r", approval_id, "allow once", context=current + ) + ] + return await engine.resume("r", approval_id, "allow once", context=current) + + if permitted: + await resume() + assert len(output.read_text().splitlines()) == 1 + else: + with pytest.raises(ApprovalInvocationMismatch): + await resume() + assert not output.exists() + calls = policy_calls.read_text() + assert "old-secret" not in calls and "new-secret" not in calls + latest = json.loads(calls.splitlines()[-1]) + assert latest["auth"].get("scopes", []) == (["write"] if permitted else []) + assert latest["auth"]["organization_id"] == ("org" if permitted else "other") + assert latest["metadata"].get("permission") == ("write" if permitted else None) + + +@pytest.mark.parametrize("identity", ["run", "session", "user", "auth_user"]) +@pytest.mark.parametrize("streaming", [False, True]) +async def test_resume_context_cannot_replace_approval_ownership(tmp_path, identity, streaming): + output = tmp_path / "executed.log" + write_tool(tmp_path, "send_email", output) + context = RunContext(run_id="r", user_id="owner", conversation_id="s") + contexts = { + "run": context.replace(run_id="another-run"), + "session": context.replace(conversation_id="another-session"), + "user": context.replace(user_id="another-user"), + "auth_user": context.replace(auth_context=AuthContext(user_id="another-user")), + } + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(_multi_tool_spec("send_email")) + pending = await engine.run("hello", context=context) + assert pending.pending_approval is not None + with pytest.raises(UnauthorizedApprover): + if streaming: + _ = [ + event + async for event in engine.resume_stream( + "r", + pending.pending_approval.approval_id, + "allow once", + caller_user_id="owner", + caller_session_id="s", + context=contexts[identity], + ) + ] + else: + await engine.resume( + "r", + pending.pending_approval.approval_id, + "allow once", + caller_user_id="owner", + caller_session_id="s", + context=contexts[identity], + ) + assert not output.exists() + assert await engine.get_run_status("r") == RunStatus.PENDING_APPROVAL + assert await engine.get_pending_approval("r") is not None + + +@pytest.mark.parametrize("streaming", [False, True]) +async def test_concurrent_approval_requests_do_not_share_authorization(tmp_path, streaming): + output = tmp_path / "executed.log" + write_tool(tmp_path, "send_email", output) + (tmp_path / "plugins" / "access.py").write_text( + "class AccessResolver:\n" + " def can_access(self, ctx, node_id):\n" + " return 'admin' in ctx.get('auth', {}).get('roles', ())\n", + encoding="utf-8", + ) + spec = _multi_tool_spec("send_email") + spec = replace(spec, graph=replace(spec.graph, node=replace(spec.graph.node, protected=True))) + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(spec) + approvals: dict[str, str] = {} + for run_id in ("allowed", "revoked"): + pending = await engine.run( + "hello", + context=RunContext(run_id=run_id, auth_context=AuthContext(roles=("admin",))), + ) + assert pending.pending_approval is not None + approvals[run_id] = pending.pending_approval.approval_id + + async def resume_one(run_id): + current = RunContext( + auth_context=AuthContext(roles=("admin",) if run_id == "allowed" else ()) + ) + + async def invoke(): + if streaming: + return [ + event + async for event in engine.resume_stream( + run_id, approvals[run_id], "allow once", context=current + ) + ] + return await engine.resume(run_id, approvals[run_id], "allow once", context=current) + + if run_id == "allowed": + await invoke() + else: + with pytest.raises(ApprovalInvocationMismatch): + await invoke() + + await asyncio.gather(resume_one("allowed"), resume_one("revoked")) + assert len(output.read_text().splitlines()) == 1 + assert await engine.get_run_status("allowed") == RunStatus.COMPLETED + assert await engine.get_run_status("revoked") == RunStatus.FAILED