"""End-to-end checks for the agent security and approval path.""" from __future__ import annotations import asyncio import shutil import pytest from cyberbrain26.orchestration.ptah_bridge import PtahBridge, RoadmapStep from cyberbrain26.security_plane.approval import ApprovalManager from cyberbrain26.security_plane.audit import AuditEventType, AuditQuery from cyberbrain26.security_plane.identity import PermissionScope requires_bubblewrap = pytest.mark.skipif( shutil.which("bwrap") is None, reason="Bubblewrap is unavailable" ) @pytest.mark.asyncio async def test_ptah_dispatch_requires_approval_and_audits_the_full_task() -> None: approvals = ApprovalManager() bridge = PtahBridge( approval_manager=approvals, approval_timeout_seconds=1.0, ) bridge.coder.identity.scopes.add(PermissionScope.SHELL.value) async def approve_when_requested() -> None: while not approvals.list_pending(): await asyncio.sleep(0.005) request = approvals.list_pending()[0] approvals.approve(request.request_id, "reviewer", "approved in integration test") approval_task = asyncio.create_task(approve_when_requested()) result = await bridge.execute_step( "roadmap-e2e", RoadmapStep( id="task-e2e-1", role="coder", description="exercise the guarded agent task", payload={"action": "execute_shell"}, ), ) await approval_task assert result.success is True assert result.step_id == "task-e2e-1" required = bridge.audit_log.query( AuditQuery(event_type=AuditEventType.APPROVAL_REQUIRED, task_id="task-e2e-1") ) granted = bridge.audit_log.query( AuditQuery(event_type=AuditEventType.APPROVAL_GRANTED, task_id="task-e2e-1") ) completed = bridge.audit_log.query( AuditQuery(event_type=AuditEventType.AGENT_FINISHED, task_id="task-e2e-1") ) assert len(required) == 1 assert len(granted) == 1 assert len(completed) == 1 assert required[0].agent_id == "ptah-coder" assert granted[0].actor == "ptah-coder" assert completed[0].result == "succeeded" @pytest.mark.asyncio async def test_ptah_task_is_denied_when_required_approval_is_missing() -> None: bridge = PtahBridge(approval_timeout_seconds=0.05) bridge.coder.identity.scopes.add(PermissionScope.SHELL.value) result = await bridge.execute_step( "roadmap-deny", RoadmapStep( id="task-denied-1", role="coder", description="must not execute without approval", payload={"action": "execute_shell"}, ), ) assert result.success is False assert "approval unavailable" in (result.error or "").lower() assert bridge.coder.get_implemented() == [] denied = bridge.audit_log.query( AuditQuery(event_type=AuditEventType.APPROVAL_DENIED, task_id="task-denied-1") ) assert len(denied) == 1 @pytest.mark.asyncio async def test_ptah_task_stays_blocked_when_reviewer_denies() -> None: approvals = ApprovalManager() bridge = PtahBridge(approval_manager=approvals, approval_timeout_seconds=1.0) bridge.coder.identity.scopes.add(PermissionScope.SHELL.value) async def deny_when_requested() -> None: while not approvals.list_pending(): await asyncio.sleep(0.005) request = approvals.list_pending()[0] approvals.deny(request.request_id, "reviewer", "not approved") denial_task = asyncio.create_task(deny_when_requested()) result = await bridge.execute_step( "roadmap-denied", RoadmapStep( id="task-denied-2", role="coder", description="must not execute after denial", payload={"action": "execute_shell"}, ), ) await denial_task assert result.success is False assert bridge.coder.get_implemented() == [] denied = bridge.audit_log.query( AuditQuery(event_type=AuditEventType.APPROVAL_DENIED, task_id="task-denied-2") ) assert len(denied) == 1 @pytest.mark.asyncio @requires_bubblewrap async def test_coder_executes_code_only_after_permission_and_approval() -> None: approvals = ApprovalManager() bridge = PtahBridge(approval_manager=approvals, approval_timeout_seconds=1.0) bridge.coder.identity.scopes.add(PermissionScope.PYTHON.value) async def approve_when_requested() -> None: while not approvals.list_pending(): await asyncio.sleep(0.005) request = approvals.list_pending()[0] approvals.approve(request.request_id, "reviewer", "sandbox run approved") approval_task = asyncio.create_task(approve_when_requested()) result = await bridge.execute_step( "roadmap-code", RoadmapStep( id="task-code-1", role="coder", description="run a small isolated Python task", payload={ "code": "print('from-coder')", "language": "python", "action": "code", }, ), ) await approval_task assert result.success is True assert result.data["execution"]["stdout"].strip() == "from-coder" approvals_audited = bridge.audit_log.query( AuditQuery(event_type=AuditEventType.APPROVAL_GRANTED, task_id="task-code-1") ) results_audited = bridge.audit_log.query( AuditQuery(event_type=AuditEventType.TOOL_RESULT, task_id="task-code-1") ) assert len(approvals_audited) == 1 assert len(results_audited) == 1 assert results_audited[0].result == "succeeded" @pytest.mark.asyncio @requires_bubblewrap async def test_coder_does_not_run_code_without_execution_scope() -> None: bridge = PtahBridge() result = await bridge.execute_step( "roadmap-no-scope", RoadmapStep( id="task-code-denied", role="coder", description="must be denied before entering sandbox", payload={ "code": "print('must-not-run')", "language": "python", "action": "code", }, ), ) assert result.success is False assert "lacks required scope" in (result.error or "") assert bridge.coder.get_implemented() == [] denied = bridge.audit_log.query( AuditQuery(event_type=AuditEventType.TOOL_CALLED, task_id="task-code-denied") ) assert len(denied) == 1 assert denied[0].result == "denied" # ---------------------------------------------------------------------- # One step, one approval: the inner gate re-verifies instead of re-asking # ---------------------------------------------------------------------- @pytest.mark.asyncio @requires_bubblewrap async def test_coder_step_raises_one_approval_that_the_inner_gate_reuses() -> None: """A code step needs exactly one human decision, and both gates still run. The operator approves the first request and nothing else. The step declares the action it really performs (``execute_python``), so the dispatcher's approval covers the execution gateway's re-check and the gateway records a reuse instead of raising a second request. Before this was fixed the run hung here until the approval timeout: the second request was never answered. """ approvals = ApprovalManager() bridge = PtahBridge(approval_manager=approvals, approval_timeout_seconds=5.0) bridge.coder.identity.scopes.add(PermissionScope.PYTHON.value) async def approve_the_first_request() -> None: while not approvals.list_pending(): await asyncio.sleep(0.005) # Exactly one approval, and the test never answers anything else. approvals.approve(approvals.list_pending()[0].request_id, "reviewer", "approved once") approval_task = asyncio.create_task(approve_the_first_request()) result = await bridge.execute_step( "roadmap-single-approval", RoadmapStep( id="task-single-approval", role="coder", description="declare the action it really performs", payload={ "action": "execute_python", "code": "print('single approval')", "language": "python", }, ), ) await approval_task assert result.success is True, result.error assert result.data["execution"]["stdout"].strip() == "single approval" assert len(approvals.list_records()) == 1, "one step must not need two decisions" # The audit trail states the gating decision for a reviewer: one request, one # human grant, and one explicit reuse by the inner gate. required = bridge.audit_log.query( AuditQuery(event_type=AuditEventType.APPROVAL_REQUIRED, task_id="task-single-approval") ) granted = bridge.audit_log.query( AuditQuery(event_type=AuditEventType.APPROVAL_GRANTED, task_id="task-single-approval") ) assert [event.result for event in required] == ["pending"] assert sorted(event.result for event in granted) == ["approved", "reused"] assert {event.action for event in required} == {"execute_python"} @pytest.mark.asyncio @requires_bubblewrap async def test_declared_label_still_asks_for_the_real_execution_action() -> None: """A non-mutating label is not an approval for the code behind it. ``code`` is a known safe label, so the outer chain allows the step without a human. Because the action the gateway is about to perform is a different one (``execute_python``), the grant cannot apply and the gateway asks for the real action on its own - which is what stops a label from authorizing executable content. """ approvals = ApprovalManager() bridge = PtahBridge(approval_manager=approvals, approval_timeout_seconds=5.0) bridge.coder.identity.scopes.add(PermissionScope.PYTHON.value) async def approve_when_requested() -> None: while not approvals.list_pending(): await asyncio.sleep(0.005) approvals.approve(approvals.list_pending()[0].request_id, "reviewer", "approved") approval_task = asyncio.create_task(approve_when_requested()) result = await bridge.execute_step( "roadmap-label", RoadmapStep( id="task-label", role="coder", description="declared as the non-mutating label", payload={ "action": "code", "code": "print('labelled')", "language": "python", }, ), ) await approval_task assert result.success is True, result.error records = approvals.list_records() assert [record.action for record in records] == ["execute_python"] required = bridge.audit_log.query( AuditQuery(event_type=AuditEventType.APPROVAL_REQUIRED, task_id="task-label") ) assert [event.action for event in required] == ["execute_python"]