wehub-resource-sync
1528 lines
50 KiB
Python
1528 lines
50 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
|
|
class _SessionManager:
|
|
def __init__(self):
|
|
self.node = SimpleNamespace(
|
|
session_key="agent:main:webchat:abc",
|
|
agent_id="main",
|
|
origin=None,
|
|
)
|
|
self.sessions = {self.node.session_key: self.node}
|
|
self.created: list[str] = []
|
|
|
|
async def get_session(self, session_key: str):
|
|
return self.sessions.get(session_key)
|
|
|
|
async def get_or_create(self, session_key: str, agent_id: str = "main", **kwargs):
|
|
existing = self.sessions.get(session_key)
|
|
if existing is not None:
|
|
return existing, False
|
|
node = SimpleNamespace(
|
|
session_key=session_key,
|
|
agent_id=agent_id,
|
|
origin=None,
|
|
**kwargs,
|
|
)
|
|
self.sessions[session_key] = node
|
|
self.created.append(session_key)
|
|
return node, True
|
|
|
|
async def update(self, session_key: str, **fields):
|
|
node = self.sessions[session_key]
|
|
for key, value in fields.items():
|
|
setattr(node, key, value)
|
|
return node
|
|
|
|
|
|
def _ctx(
|
|
manager: _SessionManager,
|
|
*,
|
|
is_owner: bool = True,
|
|
run_mode: str = "standard",
|
|
sandbox: bool = True,
|
|
security_grading: bool = True,
|
|
permissions_default_mode: str = "off",
|
|
scopes: frozenset[str] | None = None,
|
|
):
|
|
from opensquilla.gateway.auth import Principal
|
|
from opensquilla.gateway.rpc import RpcContext
|
|
|
|
config = SimpleNamespace(
|
|
workspace_dir="/tmp/ws",
|
|
agents=[],
|
|
sandbox=SimpleNamespace(
|
|
run_mode=run_mode,
|
|
sandbox=sandbox,
|
|
security_grading=security_grading,
|
|
backend="noop",
|
|
network_default="proxy_allowlist",
|
|
),
|
|
permissions=SimpleNamespace(default_mode=permissions_default_mode),
|
|
)
|
|
return RpcContext(
|
|
conn_id="c",
|
|
principal=Principal(
|
|
role="operator",
|
|
scopes=scopes or frozenset(["operator.read", "operator.write"]),
|
|
is_owner=is_owner,
|
|
authenticated=True,
|
|
),
|
|
session_manager=manager,
|
|
config=config,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_resolved_overlays() -> None:
|
|
from opensquilla.sandbox.escalation import reset_resolved_run_context_overlays
|
|
|
|
reset_resolved_run_context_overlays()
|
|
yield
|
|
reset_resolved_run_context_overlays()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_add_domain_returns_updated_context() -> None:
|
|
from opensquilla.gateway.rpc_sandbox import _handle_sandbox_domain_add
|
|
|
|
manager = _SessionManager()
|
|
|
|
result = await _handle_sandbox_domain_add(
|
|
{
|
|
"sessionKey": manager.node.session_key,
|
|
"domain": "https://pypi.org/simple",
|
|
"scope": "workspace",
|
|
},
|
|
_ctx(manager),
|
|
)
|
|
|
|
assert result["domains"] == [
|
|
{"domain": "pypi.org", "scope": "workspace", "source": "manual"}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_add_mount_rejects_non_owner() -> None:
|
|
from opensquilla.gateway.rpc import RpcHandlerError
|
|
from opensquilla.gateway.rpc_sandbox import _handle_sandbox_mount_add
|
|
|
|
manager = _SessionManager()
|
|
|
|
with pytest.raises(RpcHandlerError, match="requires owner principal"):
|
|
await _handle_sandbox_mount_add(
|
|
{"sessionKey": manager.node.session_key, "path": "/tmp/ws/extras"},
|
|
_ctx(manager, is_owner=False),
|
|
)
|
|
|
|
assert manager.node.origin is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("handler_name", "params"),
|
|
[
|
|
("_handle_sandbox_mount_add", {}),
|
|
("_handle_sandbox_mount_add", {"path": ""}),
|
|
("_handle_sandbox_mount_add", {"path": " "}),
|
|
("_handle_sandbox_mount_remove", {}),
|
|
("_handle_sandbox_mount_remove", {"path": ""}),
|
|
("_handle_sandbox_mount_remove", {"path": " "}),
|
|
],
|
|
)
|
|
async def test_rpc_mount_mutations_require_path_without_mutating_origin(
|
|
handler_name: str,
|
|
params: dict[str, str],
|
|
) -> None:
|
|
import opensquilla.gateway.rpc_sandbox as rpc_sandbox
|
|
|
|
manager = _SessionManager()
|
|
handler = getattr(rpc_sandbox, handler_name)
|
|
|
|
with pytest.raises(ValueError, match="params.path is required"):
|
|
await handler(
|
|
{"sessionKey": manager.node.session_key, **params},
|
|
_ctx(manager),
|
|
)
|
|
|
|
assert manager.node.origin is None
|
|
assert manager.created == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_mutation_rejects_whitespace_session_key_without_creating_session() -> None:
|
|
from opensquilla.gateway.rpc_sandbox import _handle_sandbox_mount_add
|
|
|
|
manager = _SessionManager()
|
|
|
|
with pytest.raises(ValueError, match="params.sessionKey is required"):
|
|
await _handle_sandbox_mount_add(
|
|
{"sessionKey": " ", "path": "/tmp/ws/extras"},
|
|
_ctx(manager),
|
|
)
|
|
|
|
assert " " not in manager.sessions
|
|
assert manager.created == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_run_context_get_includes_bundles_and_temporary_grants() -> None:
|
|
from opensquilla.gateway.rpc_sandbox import _handle_sandbox_run_context_get
|
|
from opensquilla.sandbox.run_context import (
|
|
PublicNetworkGrant,
|
|
RunContext,
|
|
TemporaryGrant,
|
|
persist_run_context,
|
|
)
|
|
from opensquilla.sandbox.run_mode import RunMode
|
|
from opensquilla.sandbox.user_grants import upsert_bundle_grant
|
|
|
|
manager = _SessionManager()
|
|
upsert_bundle_grant(
|
|
{
|
|
"bundle_id": "python-package-install",
|
|
"scope": "workspace",
|
|
"source": "manual",
|
|
}
|
|
)
|
|
await persist_run_context(
|
|
manager,
|
|
manager.node.session_key,
|
|
RunContext(
|
|
run_mode=RunMode.STANDARD,
|
|
workspace="/tmp/ws",
|
|
public_network=(PublicNetworkGrant(scope="chat", source="manual"),),
|
|
temporary_grants=(
|
|
TemporaryGrant(
|
|
kind="domain",
|
|
value="pypi.org",
|
|
fingerprint="abc123",
|
|
),
|
|
),
|
|
source="saved",
|
|
),
|
|
)
|
|
|
|
result = await _handle_sandbox_run_context_get(
|
|
{"sessionKey": manager.node.session_key},
|
|
_ctx(manager),
|
|
)
|
|
|
|
assert result["bundles"] == [
|
|
{
|
|
"bundle_id": "python-package-install",
|
|
"scope": "workspace",
|
|
"source": "manual",
|
|
}
|
|
]
|
|
assert result["publicNetwork"] == [
|
|
{
|
|
"scope": "chat",
|
|
"source": "manual",
|
|
}
|
|
]
|
|
assert result["temporaryGrants"] == [
|
|
{
|
|
"kind": "domain",
|
|
"value": "pypi.org",
|
|
"fingerprint": "abc123",
|
|
"expires_after": "once",
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exec_approval_resolve_allows_non_owner_chat_scoped_sandbox_grant() -> None:
|
|
from opensquilla.gateway.approval_queue import get_approval_queue, reset_approval_queue
|
|
from opensquilla.gateway.rpc_approvals import _handle_exec_approval_resolve
|
|
from opensquilla.sandbox.escalation import build_network_approval_params
|
|
from opensquilla.sandbox.network_guard import NetworkDecision
|
|
from opensquilla.sandbox.run_context import get_run_context
|
|
|
|
reset_approval_queue()
|
|
manager = _SessionManager()
|
|
params = build_network_approval_params(
|
|
NetworkDecision(
|
|
status="ask",
|
|
normalized_host="example.com",
|
|
reason="unknown_domain",
|
|
source=None,
|
|
),
|
|
session_key=manager.node.session_key,
|
|
workspace="/tmp/ws",
|
|
fingerprint="fp123",
|
|
)
|
|
assert params is not None
|
|
queue = get_approval_queue()
|
|
approval_id = queue.request(namespace="exec", params=params)
|
|
|
|
result = await _handle_exec_approval_resolve(
|
|
{"id": approval_id, "approved": True, "choice": "allow_same_type"},
|
|
_ctx(
|
|
manager,
|
|
is_owner=False,
|
|
scopes=frozenset(["operator.approvals"]),
|
|
),
|
|
)
|
|
|
|
assert result["resolved"] is True
|
|
assert result["approved"] is True
|
|
context = await get_run_context(
|
|
manager,
|
|
manager.node.session_key,
|
|
config=_ctx(manager).config,
|
|
workspace="/tmp/ws",
|
|
)
|
|
assert ("example.com", "chat") in [(grant.domain, grant.scope) for grant in context.domains]
|
|
|
|
reset_approval_queue()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exec_approval_resolve_rejects_legacy_intent_flags() -> None:
|
|
# "Allow always" / rememberIntent were a removed no-op; a truthy value must
|
|
# now be rejected loudly rather than silently accepted, and must not resolve
|
|
# the approval or apply any grant.
|
|
from opensquilla.gateway.approval_queue import get_approval_queue, reset_approval_queue
|
|
from opensquilla.gateway.rpc import RpcHandlerError
|
|
from opensquilla.gateway.rpc_approvals import _handle_exec_approval_resolve
|
|
from opensquilla.sandbox.escalation import build_network_approval_params
|
|
from opensquilla.sandbox.network_guard import NetworkDecision
|
|
|
|
reset_approval_queue()
|
|
manager = _SessionManager()
|
|
params = build_network_approval_params(
|
|
NetworkDecision(
|
|
status="ask",
|
|
normalized_host="example.com",
|
|
reason="unknown_domain",
|
|
source=None,
|
|
),
|
|
session_key=manager.node.session_key,
|
|
workspace="/tmp/ws",
|
|
fingerprint="fp123",
|
|
)
|
|
assert params is not None
|
|
queue = get_approval_queue()
|
|
approval_id = queue.request(namespace="exec", params=params)
|
|
|
|
with pytest.raises(RpcHandlerError) as excinfo:
|
|
await _handle_exec_approval_resolve(
|
|
{
|
|
"id": approval_id,
|
|
"approved": True,
|
|
"choice": "allow_same_type",
|
|
"allowAlways": True,
|
|
"rememberIntent": True,
|
|
},
|
|
_ctx(
|
|
manager,
|
|
is_owner=False,
|
|
scopes=frozenset(["operator.approvals"]),
|
|
),
|
|
)
|
|
|
|
assert excinfo.value.code == "UNSUPPORTED_PARAM"
|
|
# The approval is untouched — rejection happens before any resolution.
|
|
assert queue.get(approval_id).resolved is False
|
|
|
|
reset_approval_queue()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exec_approval_resolve_rejects_non_owner_missing_sandbox_choice() -> None:
|
|
from opensquilla.gateway.approval_queue import get_approval_queue, reset_approval_queue
|
|
from opensquilla.gateway.rpc import RpcHandlerError
|
|
from opensquilla.gateway.rpc_approvals import _handle_exec_approval_resolve
|
|
from opensquilla.sandbox.escalation import build_network_approval_params
|
|
from opensquilla.sandbox.network_guard import NetworkDecision
|
|
|
|
reset_approval_queue()
|
|
manager = _SessionManager()
|
|
params = build_network_approval_params(
|
|
NetworkDecision(
|
|
status="ask",
|
|
normalized_host="example.com",
|
|
reason="unknown_domain",
|
|
source=None,
|
|
),
|
|
session_key=manager.node.session_key,
|
|
workspace="/tmp/ws",
|
|
fingerprint="fp123",
|
|
)
|
|
assert params is not None
|
|
queue = get_approval_queue()
|
|
approval_id = queue.request(namespace="exec", params=params)
|
|
|
|
with pytest.raises(RpcHandlerError, match="requires owner principal"):
|
|
await _handle_exec_approval_resolve(
|
|
{"id": approval_id, "approved": True},
|
|
_ctx(
|
|
manager,
|
|
is_owner=False,
|
|
scopes=frozenset(["operator.approvals"]),
|
|
),
|
|
)
|
|
|
|
pending = queue.get(approval_id)
|
|
assert pending.resolved is False
|
|
|
|
reset_approval_queue()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exec_approval_resolve_allows_non_owner_chat_scoped_path_mount(tmp_path) -> None:
|
|
from opensquilla.gateway.approval_queue import get_approval_queue, reset_approval_queue
|
|
from opensquilla.gateway.rpc_approvals import _handle_exec_approval_resolve
|
|
from opensquilla.sandbox.escalation import build_path_approval_params
|
|
from opensquilla.sandbox.path_validation import MountDecision
|
|
from opensquilla.sandbox.run_context import get_run_context
|
|
|
|
reset_approval_queue()
|
|
manager = _SessionManager()
|
|
workspace = tmp_path / "workspace"
|
|
outside = tmp_path / "outside"
|
|
workspace.mkdir()
|
|
outside.mkdir()
|
|
params = build_path_approval_params(
|
|
MountDecision(
|
|
status="request",
|
|
normalized_path=str(outside.resolve(strict=False)),
|
|
access="ro",
|
|
reason="outside_sandbox_mounts",
|
|
),
|
|
session_key=manager.node.session_key,
|
|
workspace=str(workspace),
|
|
)
|
|
assert params is not None
|
|
queue = get_approval_queue()
|
|
approval_id = queue.request(namespace="exec", params=params)
|
|
|
|
result = await _handle_exec_approval_resolve(
|
|
{"id": approval_id, "approved": True, "choice": "allow_same_type"},
|
|
_ctx(
|
|
manager,
|
|
is_owner=False,
|
|
scopes=frozenset(["operator.approvals"]),
|
|
),
|
|
)
|
|
|
|
assert result["resolved"] is True
|
|
assert result["approved"] is True
|
|
context = await get_run_context(
|
|
manager,
|
|
manager.node.session_key,
|
|
config=_ctx(manager).config,
|
|
workspace=str(workspace),
|
|
)
|
|
assert (str(outside.resolve(strict=False)), "ro", "chat") in [
|
|
(grant.path, grant.access, grant.scope) for grant in context.mounts
|
|
]
|
|
|
|
reset_approval_queue()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exec_approval_resolve_allows_non_owner_sandbox_grant_denial() -> None:
|
|
from opensquilla.gateway.approval_queue import get_approval_queue, reset_approval_queue
|
|
from opensquilla.gateway.rpc_approvals import _handle_exec_approval_resolve
|
|
from opensquilla.sandbox.escalation import build_network_approval_params
|
|
from opensquilla.sandbox.network_guard import NetworkDecision
|
|
from opensquilla.sandbox.run_context import get_run_context
|
|
|
|
reset_approval_queue()
|
|
manager = _SessionManager()
|
|
params = build_network_approval_params(
|
|
NetworkDecision(
|
|
status="ask",
|
|
normalized_host="example.com",
|
|
reason="unknown_domain",
|
|
source=None,
|
|
),
|
|
session_key=manager.node.session_key,
|
|
workspace="/tmp/ws",
|
|
fingerprint="fp123",
|
|
)
|
|
assert params is not None
|
|
queue = get_approval_queue()
|
|
approval_id = queue.request(namespace="exec", params=params)
|
|
|
|
result = await _handle_exec_approval_resolve(
|
|
{"id": approval_id, "approved": False, "choice": "deny"},
|
|
_ctx(
|
|
manager,
|
|
is_owner=False,
|
|
scopes=frozenset(["operator.approvals"]),
|
|
),
|
|
)
|
|
|
|
assert result["resolved"] is True
|
|
assert result["approved"] is False
|
|
context = await get_run_context(
|
|
manager,
|
|
manager.node.session_key,
|
|
config=_ctx(manager).config,
|
|
workspace="/tmp/ws",
|
|
)
|
|
assert context.domains == ()
|
|
|
|
reset_approval_queue()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exec_approval_resolve_leaves_sandbox_approval_pending_when_mutation_fails(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from opensquilla.gateway.approval_queue import get_approval_queue, reset_approval_queue
|
|
from opensquilla.gateway.rpc import get_dispatcher
|
|
from opensquilla.sandbox.escalation import build_network_approval_params
|
|
from opensquilla.sandbox.network_guard import NetworkDecision
|
|
from opensquilla.sandbox.run_context import get_run_context
|
|
|
|
reset_approval_queue()
|
|
manager = _SessionManager()
|
|
params = build_network_approval_params(
|
|
NetworkDecision(
|
|
status="ask",
|
|
normalized_host="example.com",
|
|
reason="unknown_domain",
|
|
source=None,
|
|
),
|
|
session_key=manager.node.session_key,
|
|
workspace="/tmp/ws",
|
|
fingerprint="fp123",
|
|
)
|
|
assert params is not None
|
|
queue = get_approval_queue()
|
|
approval_id = queue.request(namespace="exec", params=params)
|
|
|
|
async def fail_apply(*args, **kwargs) -> None:
|
|
raise RuntimeError("mutation failed")
|
|
|
|
monkeypatch.setattr(
|
|
"opensquilla.gateway.rpc_approvals.apply_sandbox_approval_choice",
|
|
fail_apply,
|
|
)
|
|
|
|
result = await get_dispatcher().dispatch(
|
|
"r1",
|
|
"exec.approval.resolve",
|
|
{"id": approval_id, "approved": True, "choice": "allow_same_type"},
|
|
_ctx(manager, scopes=frozenset(["operator.approvals"])),
|
|
)
|
|
|
|
assert result.error is not None
|
|
assert "mutation failed" in result.error.message
|
|
pending = queue.get(approval_id)
|
|
assert pending.resolved is False
|
|
assert pending.approved is False
|
|
context = await get_run_context(
|
|
manager,
|
|
manager.node.session_key,
|
|
config=_ctx(manager).config,
|
|
workspace="/tmp/ws",
|
|
)
|
|
assert context.domains == ()
|
|
|
|
reset_approval_queue()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exec_approval_resolve_claim_prevents_deny_race_from_landing_grant(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from opensquilla.gateway.approval_queue import get_approval_queue, reset_approval_queue
|
|
from opensquilla.gateway.rpc import get_dispatcher
|
|
from opensquilla.sandbox import escalation as escalation_mod
|
|
from opensquilla.sandbox.escalation import build_network_approval_params
|
|
from opensquilla.sandbox.network_guard import NetworkDecision
|
|
from opensquilla.sandbox.run_context import get_run_context
|
|
|
|
reset_approval_queue()
|
|
manager = _SessionManager()
|
|
ctx = _ctx(manager, scopes=frozenset(["operator.approvals"]))
|
|
params = build_network_approval_params(
|
|
NetworkDecision(
|
|
status="ask",
|
|
normalized_host="example.com",
|
|
reason="unknown_domain",
|
|
source=None,
|
|
),
|
|
session_key=manager.node.session_key,
|
|
workspace="/tmp/ws",
|
|
fingerprint="fp123",
|
|
)
|
|
assert params is not None
|
|
queue = get_approval_queue()
|
|
approval_id = queue.request(namespace="exec", params=params)
|
|
|
|
mutation_started = asyncio.Event()
|
|
release_mutation = asyncio.Event()
|
|
|
|
async def delayed_apply(*args, **kwargs) -> None:
|
|
mutation_started.set()
|
|
await release_mutation.wait()
|
|
await escalation_mod.apply_sandbox_approval_choice(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(
|
|
"opensquilla.gateway.rpc_approvals.apply_sandbox_approval_choice",
|
|
delayed_apply,
|
|
)
|
|
|
|
approve_task = asyncio.create_task(
|
|
get_dispatcher().dispatch(
|
|
"approve",
|
|
"exec.approval.resolve",
|
|
{"id": approval_id, "approved": True, "choice": "allow_same_type"},
|
|
ctx,
|
|
)
|
|
)
|
|
await asyncio.wait_for(mutation_started.wait(), timeout=1)
|
|
|
|
deny_result = await get_dispatcher().dispatch(
|
|
"deny",
|
|
"exec.approval.resolve",
|
|
{"id": approval_id, "approved": False, "choice": "deny"},
|
|
ctx,
|
|
)
|
|
release_mutation.set()
|
|
approve_result = await approve_task
|
|
|
|
assert deny_result.error is not None
|
|
assert approve_result.error is None, approve_result.error
|
|
resolved = queue.get(approval_id)
|
|
assert resolved.resolved is True
|
|
assert resolved.approved is True
|
|
context = await get_run_context(
|
|
manager,
|
|
manager.node.session_key,
|
|
config=ctx.config,
|
|
workspace="/tmp/ws",
|
|
)
|
|
assert ("example.com", "chat") in [(grant.domain, grant.scope) for grant in context.domains]
|
|
|
|
reset_approval_queue()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exec_approval_wait_and_consume_wait_for_sandbox_grant_apply(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from opensquilla.gateway.approval_queue import get_approval_queue, reset_approval_queue
|
|
from opensquilla.gateway.rpc import get_dispatcher
|
|
from opensquilla.sandbox import escalation as escalation_mod
|
|
from opensquilla.sandbox.escalation import build_network_approval_params
|
|
from opensquilla.sandbox.network_guard import NetworkDecision
|
|
|
|
reset_approval_queue()
|
|
manager = _SessionManager()
|
|
ctx = _ctx(manager, scopes=frozenset(["operator.approvals"]))
|
|
params = build_network_approval_params(
|
|
NetworkDecision(
|
|
status="ask",
|
|
normalized_host="example.com",
|
|
reason="unknown_domain",
|
|
source=None,
|
|
),
|
|
session_key=manager.node.session_key,
|
|
workspace="/tmp/ws",
|
|
fingerprint="fp123",
|
|
)
|
|
assert params is not None
|
|
queue = get_approval_queue()
|
|
approval_id = queue.request(namespace="exec", params=params)
|
|
|
|
mutation_started = asyncio.Event()
|
|
release_mutation = asyncio.Event()
|
|
|
|
async def delayed_apply(*args, **kwargs) -> None:
|
|
mutation_started.set()
|
|
await release_mutation.wait()
|
|
await escalation_mod.apply_sandbox_approval_choice(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(
|
|
"opensquilla.gateway.rpc_approvals.apply_sandbox_approval_choice",
|
|
delayed_apply,
|
|
)
|
|
|
|
approve_task = asyncio.create_task(
|
|
get_dispatcher().dispatch(
|
|
"approve",
|
|
"exec.approval.resolve",
|
|
{"id": approval_id, "approved": True, "choice": "allow_same_type"},
|
|
ctx,
|
|
)
|
|
)
|
|
await asyncio.wait_for(mutation_started.wait(), timeout=1)
|
|
|
|
wait_task = asyncio.create_task(queue.wait(approval_id, timeout=1.0))
|
|
wait_decision_task = asyncio.create_task(
|
|
get_dispatcher().dispatch(
|
|
"wait",
|
|
"exec.approval.waitDecision",
|
|
{"id": approval_id},
|
|
ctx,
|
|
)
|
|
)
|
|
await asyncio.sleep(0.05)
|
|
|
|
assert wait_task.done() is False
|
|
assert wait_decision_task.done() is False
|
|
with pytest.raises(ValueError, match="in progress|not approved"):
|
|
queue.consume(approval_id)
|
|
|
|
release_mutation.set()
|
|
approve_result = await approve_task
|
|
assert approve_result.error is None, approve_result.error
|
|
assert await wait_task is True
|
|
wait_decision_result = await wait_decision_task
|
|
assert wait_decision_result.error is None, wait_decision_result.error
|
|
assert wait_decision_result.payload["approved"] is True
|
|
assert wait_decision_result.payload["resolved"] is True
|
|
|
|
queue.consume(approval_id)
|
|
assert queue.status(approval_id)["consumed"] is True
|
|
|
|
reset_approval_queue()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exec_approval_resolve_recovers_complete_failure_after_grant_apply(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from opensquilla.gateway.approval_queue import get_approval_queue, reset_approval_queue
|
|
from opensquilla.gateway.rpc import get_dispatcher
|
|
from opensquilla.sandbox.escalation import build_network_approval_params
|
|
from opensquilla.sandbox.network_guard import NetworkDecision
|
|
from opensquilla.sandbox.run_context import get_run_context
|
|
|
|
reset_approval_queue()
|
|
manager = _SessionManager()
|
|
ctx = _ctx(manager, scopes=frozenset(["operator.approvals"]))
|
|
params = build_network_approval_params(
|
|
NetworkDecision(
|
|
status="ask",
|
|
normalized_host="example.com",
|
|
reason="unknown_domain",
|
|
source=None,
|
|
),
|
|
session_key=manager.node.session_key,
|
|
workspace="/tmp/ws",
|
|
fingerprint="fp123",
|
|
)
|
|
assert params is not None
|
|
queue = get_approval_queue()
|
|
approval_id = queue.request(namespace="exec", params=params)
|
|
original_complete = queue.complete_claimed_resolution
|
|
attempts = 0
|
|
|
|
def fail_once_then_complete(*args, **kwargs) -> None:
|
|
nonlocal attempts
|
|
attempts += 1
|
|
if attempts == 1:
|
|
raise RuntimeError("transient complete failed")
|
|
original_complete(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(queue, "complete_claimed_resolution", fail_once_then_complete)
|
|
|
|
result = await get_dispatcher().dispatch(
|
|
"approve",
|
|
"exec.approval.resolve",
|
|
{"id": approval_id, "approved": True, "choice": "allow_same_type"},
|
|
ctx,
|
|
)
|
|
|
|
assert result.error is None, result.error
|
|
assert attempts == 2
|
|
assert queue.status(approval_id)["resolved"] is True
|
|
assert queue.status(approval_id)["approved"] is True
|
|
context = await get_run_context(
|
|
manager,
|
|
manager.node.session_key,
|
|
config=ctx.config,
|
|
workspace="/tmp/ws",
|
|
)
|
|
assert ("example.com", "chat") in [(grant.domain, grant.scope) for grant in context.domains]
|
|
|
|
reset_approval_queue()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exec_approval_resolve_finalize_failure_does_not_land_grant(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from opensquilla.gateway.approval_queue import get_approval_queue, reset_approval_queue
|
|
from opensquilla.gateway.rpc import get_dispatcher
|
|
from opensquilla.sandbox.escalation import build_network_approval_params
|
|
from opensquilla.sandbox.network_guard import NetworkDecision
|
|
from opensquilla.sandbox.run_context import get_run_context
|
|
|
|
reset_approval_queue()
|
|
manager = _SessionManager()
|
|
ctx = _ctx(manager, scopes=frozenset(["operator.approvals"]))
|
|
params = build_network_approval_params(
|
|
NetworkDecision(
|
|
status="ask",
|
|
normalized_host="example.com",
|
|
reason="unknown_domain",
|
|
source=None,
|
|
),
|
|
session_key=manager.node.session_key,
|
|
workspace="/tmp/ws",
|
|
fingerprint="fp123",
|
|
)
|
|
assert params is not None
|
|
queue = get_approval_queue()
|
|
approval_id = queue.request(namespace="exec", params=params)
|
|
|
|
def fail_finalize(*args, **kwargs) -> None:
|
|
raise RuntimeError("finalize failed")
|
|
|
|
monkeypatch.setattr(queue, "finalize_claimed_resolution", fail_finalize)
|
|
|
|
result = await get_dispatcher().dispatch(
|
|
"r1",
|
|
"exec.approval.resolve",
|
|
{"id": approval_id, "approved": True, "choice": "allow_same_type"},
|
|
ctx,
|
|
)
|
|
|
|
assert result.error is not None
|
|
assert "finalize failed" in result.error.message
|
|
pending = queue.get(approval_id)
|
|
assert pending.resolved is False
|
|
assert queue.list_pending("exec")[0]["id"] == approval_id
|
|
context = await get_run_context(
|
|
manager,
|
|
manager.node.session_key,
|
|
config=ctx.config,
|
|
workspace="/tmp/ws",
|
|
)
|
|
assert context.domains == ()
|
|
|
|
reset_approval_queue()
|
|
|
|
|
|
def test_claimed_approval_reappears_after_claim_lease_expires(tmp_path) -> None:
|
|
from opensquilla.application.approval_queue import ApprovalQueue
|
|
|
|
db_path = tmp_path / "approval_queue.sqlite"
|
|
queue = ApprovalQueue(db_path=str(db_path), claim_ttl_seconds=0)
|
|
approval_id = queue.request(namespace="exec", params={"command": "echo ok"})
|
|
queue.claim_resolution(approval_id)
|
|
queue.close()
|
|
|
|
reloaded = ApprovalQueue(db_path=str(db_path), claim_ttl_seconds=0)
|
|
|
|
assert [item["id"] for item in reloaded.list_pending("exec")] == [approval_id]
|
|
reloaded.resolve(approval_id, False)
|
|
assert reloaded.status(approval_id)["resolved"] is True
|
|
assert reloaded.status(approval_id)["approved"] is False
|
|
reloaded.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_mount_remove_updates_resolved_overlay_for_current_tool_mounts() -> None:
|
|
from opensquilla.gateway.rpc_sandbox import _handle_sandbox_mount_remove
|
|
from opensquilla.sandbox.escalation import current_tool_mounts, remember_resolved_run_context
|
|
from opensquilla.sandbox.run_context import MountGrant, RunContext
|
|
from opensquilla.sandbox.run_mode import RunMode
|
|
from opensquilla.tools.types import CallerKind, ToolContext, current_tool_context
|
|
|
|
manager = _SessionManager()
|
|
ctx = _ctx(manager)
|
|
|
|
remembered = RunContext(
|
|
run_mode=RunMode.STANDARD,
|
|
workspace="/tmp/ws",
|
|
mounts=(MountGrant(path="/tmp/ws/extras", access="ro", scope="chat"),),
|
|
source="saved",
|
|
)
|
|
remember_resolved_run_context(
|
|
manager.node.session_key,
|
|
"/tmp/ws",
|
|
remembered,
|
|
session_manager=manager,
|
|
config=ctx.config,
|
|
)
|
|
manager.node.origin = remembered.to_origin_payload() and {
|
|
"sandbox_run_context": remembered.to_origin_payload()
|
|
}
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir="/tmp/ws",
|
|
session_key=manager.node.session_key,
|
|
sandbox_mounts=[{"path": "/tmp/ws/extras", "access": "ro"}],
|
|
sandbox_run_context=remembered,
|
|
)
|
|
)
|
|
try:
|
|
assert current_tool_mounts() == [{"path": "/tmp/ws/extras", "access": "ro"}]
|
|
|
|
result = await _handle_sandbox_mount_remove(
|
|
{"sessionKey": manager.node.session_key, "path": "/tmp/ws/extras"},
|
|
ctx,
|
|
)
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert result["mounts"] == []
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir="/tmp/ws",
|
|
session_key=manager.node.session_key,
|
|
sandbox_mounts=[{"path": "/tmp/ws/extras", "access": "ro"}],
|
|
sandbox_run_context=remembered,
|
|
)
|
|
)
|
|
try:
|
|
assert current_tool_mounts() == []
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_mount_remove_chat_scope_leaves_user_scope_mount_visible() -> None:
|
|
from opensquilla.gateway.rpc_sandbox import _handle_sandbox_mount_remove
|
|
from opensquilla.sandbox.path_validation import normalize_path
|
|
from opensquilla.sandbox.run_context import MountGrant, RunContext, get_run_context
|
|
from opensquilla.sandbox.run_mode import RunMode
|
|
from opensquilla.sandbox.user_grants import upsert_mount_grant
|
|
|
|
manager = _SessionManager()
|
|
ctx = _ctx(manager)
|
|
path = "/tmp/shared-mount"
|
|
normalized_path = str(normalize_path(path))
|
|
upsert_mount_grant({"path": path, "access": "ro", "scope": "workspace"})
|
|
manager.node.origin = {
|
|
"sandbox_run_context": RunContext(
|
|
run_mode=RunMode.STANDARD,
|
|
workspace="/tmp/ws",
|
|
mounts=(MountGrant(path=path, access="rw", scope="chat"),),
|
|
source="saved",
|
|
).to_origin_payload()
|
|
}
|
|
|
|
result = await _handle_sandbox_mount_remove(
|
|
{"sessionKey": manager.node.session_key, "path": path, "scope": "chat"},
|
|
ctx,
|
|
)
|
|
|
|
assert result["mounts"] == [
|
|
{"path": normalized_path, "access": "ro", "scope": "workspace"}
|
|
]
|
|
context = await get_run_context(
|
|
manager,
|
|
manager.node.session_key,
|
|
config=ctx.config,
|
|
workspace="/tmp/ws",
|
|
)
|
|
assert [(grant.path, grant.access, grant.scope) for grant in context.mounts] == [
|
|
(normalized_path, "ro", "workspace")
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_domain_remove_updates_resolved_overlay_for_current_tool_context() -> None:
|
|
from opensquilla.gateway.rpc_sandbox import _handle_sandbox_domain_remove
|
|
from opensquilla.sandbox.escalation import (
|
|
current_tool_run_context,
|
|
remember_resolved_run_context,
|
|
)
|
|
from opensquilla.sandbox.network_guard import decide_network_access
|
|
from opensquilla.sandbox.run_context import DomainGrant, RunContext
|
|
from opensquilla.sandbox.run_mode import RunMode
|
|
from opensquilla.tools.types import CallerKind, ToolContext, current_tool_context
|
|
|
|
manager = _SessionManager()
|
|
ctx = _ctx(manager)
|
|
remembered = RunContext(
|
|
run_mode=RunMode.STANDARD,
|
|
workspace="/tmp/ws",
|
|
domains=(DomainGrant(domain="example.com", scope="chat"),),
|
|
source="saved",
|
|
)
|
|
remember_resolved_run_context(
|
|
manager.node.session_key,
|
|
"/tmp/ws",
|
|
remembered,
|
|
session_manager=manager,
|
|
config=ctx.config,
|
|
)
|
|
manager.node.origin = {"sandbox_run_context": remembered.to_origin_payload()}
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir="/tmp/ws",
|
|
session_key=manager.node.session_key,
|
|
sandbox_run_context=remembered,
|
|
)
|
|
)
|
|
try:
|
|
merged = current_tool_run_context()
|
|
assert merged is not None
|
|
assert decide_network_access("example.com", merged).status == "allow"
|
|
|
|
result = await _handle_sandbox_domain_remove(
|
|
{"sessionKey": manager.node.session_key, "domain": "example.com"},
|
|
ctx,
|
|
)
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert result["domains"] == []
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir="/tmp/ws",
|
|
session_key=manager.node.session_key,
|
|
sandbox_run_context=remembered,
|
|
)
|
|
)
|
|
try:
|
|
merged = current_tool_run_context()
|
|
assert merged is not None
|
|
assert decide_network_access("example.com", merged).status == "ask"
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_domain_remove_chat_scope_leaves_user_scope_domain_visible() -> None:
|
|
from opensquilla.gateway.rpc_sandbox import _handle_sandbox_domain_remove
|
|
from opensquilla.sandbox.run_context import DomainGrant, RunContext, get_run_context
|
|
from opensquilla.sandbox.run_mode import RunMode
|
|
from opensquilla.sandbox.user_grants import upsert_domain_grant
|
|
|
|
manager = _SessionManager()
|
|
ctx = _ctx(manager)
|
|
upsert_domain_grant(
|
|
{"domain": "example.com", "scope": "workspace", "source": "manual"}
|
|
)
|
|
manager.node.origin = {
|
|
"sandbox_run_context": RunContext(
|
|
run_mode=RunMode.STANDARD,
|
|
workspace="/tmp/ws",
|
|
domains=(DomainGrant(domain="example.com", scope="chat", source="manual"),),
|
|
source="saved",
|
|
).to_origin_payload()
|
|
}
|
|
|
|
result = await _handle_sandbox_domain_remove(
|
|
{"sessionKey": manager.node.session_key, "domain": "example.com", "scope": "chat"},
|
|
ctx,
|
|
)
|
|
|
|
assert result["domains"] == [
|
|
{"domain": "example.com", "scope": "workspace", "source": "manual"}
|
|
]
|
|
context = await get_run_context(
|
|
manager,
|
|
manager.node.session_key,
|
|
config=ctx.config,
|
|
workspace="/tmp/ws",
|
|
)
|
|
assert [(grant.domain, grant.scope) for grant in context.domains] == [
|
|
("example.com", "workspace")
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_sandbox_status_reports_backend_managed_network_and_run_mode() -> None:
|
|
from opensquilla.gateway.rpc_sandbox import _handle_sandbox_status
|
|
|
|
manager = _SessionManager()
|
|
|
|
result = await _handle_sandbox_status({}, _ctx(manager))
|
|
|
|
assert result["run_mode"] == "standard"
|
|
assert result["run_mode_label"] == "Standard-Sandbox"
|
|
assert result["execution_target"] == "sandbox"
|
|
assert result["posture"] == "standard"
|
|
assert result["backend"] == "noop"
|
|
assert result["managed_network"] == "ready"
|
|
assert result["sandbox"] == {
|
|
"sandbox": True,
|
|
"security_grading": True,
|
|
"network_default": "proxy_allowlist",
|
|
}
|
|
catalog_by_id = {
|
|
bundle["bundle_id"]: set(bundle["domains"])
|
|
for bundle in result["bundle_catalog"]
|
|
}
|
|
expected_catalog_subsets = {
|
|
"python-package-install": {
|
|
"pypi.org",
|
|
"files.pythonhosted.org",
|
|
"pypi.python.org",
|
|
"bootstrap.pypa.io",
|
|
},
|
|
"node-package-install": {
|
|
"registry.npmjs.org",
|
|
"registry.yarnpkg.com",
|
|
"yarnpkg.com",
|
|
"nodejs.org",
|
|
},
|
|
"rust-package-install": {
|
|
"crates.io",
|
|
"static.crates.io",
|
|
"index.crates.io",
|
|
"github.com",
|
|
"objects.githubusercontent.com",
|
|
},
|
|
"go-package-install": {
|
|
"proxy.golang.org",
|
|
"sum.golang.org",
|
|
"go.dev",
|
|
"golang.org",
|
|
"storage.googleapis.com",
|
|
},
|
|
"github-default": {
|
|
"github.com",
|
|
"api.github.com",
|
|
"raw.githubusercontent.com",
|
|
"codeload.github.com",
|
|
"objects.githubusercontent.com",
|
|
},
|
|
}
|
|
for bundle_id, expected_domains in expected_catalog_subsets.items():
|
|
assert bundle_id in catalog_by_id
|
|
assert expected_domains.issubset(catalog_by_id[bundle_id])
|
|
assert result["permissions"] == {"default_mode": "off"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_sandbox_status_reports_full_host_access_without_managed_controls() -> None:
|
|
from opensquilla.gateway.rpc_sandbox import _handle_sandbox_status
|
|
|
|
result = await _handle_sandbox_status(
|
|
{},
|
|
_ctx(
|
|
_SessionManager(),
|
|
run_mode="full",
|
|
sandbox=False,
|
|
security_grading=False,
|
|
permissions_default_mode="full",
|
|
),
|
|
)
|
|
|
|
assert result["run_mode"] == "full"
|
|
assert result["run_mode_label"] == "Full Host Access"
|
|
assert result["execution_target"] == "host"
|
|
assert result["posture"] == "full"
|
|
assert result["managed_network"] == "inactive"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_sandbox_explain_returns_status_messages_and_optional_context() -> None:
|
|
from opensquilla.gateway.rpc_sandbox import _handle_sandbox_explain
|
|
from opensquilla.sandbox.run_context import RunContext, persist_run_context
|
|
from opensquilla.sandbox.run_mode import RunMode
|
|
|
|
manager = _SessionManager()
|
|
await persist_run_context(
|
|
manager,
|
|
manager.node.session_key,
|
|
RunContext(run_mode=RunMode.TRUSTED, workspace="/tmp/ws", source="saved"),
|
|
)
|
|
|
|
result = await _handle_sandbox_explain(
|
|
{"sessionKey": manager.node.session_key},
|
|
_ctx(manager),
|
|
)
|
|
|
|
assert result["status"]["run_mode"] == "standard"
|
|
assert result["runContext"]["runMode"] == "trusted"
|
|
assert result["messages"] == [
|
|
{"kind": "run_mode", "message": "Run mode is standard."},
|
|
{
|
|
"kind": "managed_network",
|
|
"message": "Managed network allowlist is ready.",
|
|
},
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("handler_name", "params"),
|
|
[
|
|
("_handle_sandbox_workspace_set", {"workspace": "/tmp/ws/project"}),
|
|
("_handle_sandbox_mount_remove", {"path": "/tmp/ws/extras"}),
|
|
("_handle_sandbox_domain_add", {"domain": "pypi.org"}),
|
|
("_handle_sandbox_domain_remove", {"domain": "pypi.org"}),
|
|
("_handle_sandbox_bundle_enable", {"bundleId": "python-package-install"}),
|
|
("_handle_sandbox_bundle_disable", {"bundleId": "python-package-install"}),
|
|
],
|
|
)
|
|
async def test_rpc_sandbox_mutations_reject_non_owner(
|
|
handler_name: str,
|
|
params: dict[str, str],
|
|
) -> None:
|
|
import opensquilla.gateway.rpc_sandbox as rpc_sandbox
|
|
from opensquilla.gateway.rpc import RpcHandlerError
|
|
|
|
manager = _SessionManager()
|
|
handler = getattr(rpc_sandbox, handler_name)
|
|
|
|
with pytest.raises(RpcHandlerError, match="requires owner principal"):
|
|
await handler(
|
|
{"sessionKey": manager.node.session_key, **params},
|
|
_ctx(manager, is_owner=False),
|
|
)
|
|
|
|
assert manager.node.origin is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("handler_name", "params", "message"),
|
|
[
|
|
("_handle_sandbox_domain_add", {}, "params.domain is required"),
|
|
("_handle_sandbox_domain_remove", {"domain": ""}, "params.domain is required"),
|
|
("_handle_sandbox_bundle_enable", {}, "params.bundleId is required"),
|
|
(
|
|
"_handle_sandbox_bundle_enable",
|
|
{"bundleId": "unknown-package-install"},
|
|
"unknown_package_bundle",
|
|
),
|
|
(
|
|
"_handle_sandbox_bundle_disable",
|
|
{"bundle_id": " "},
|
|
"params.bundleId is required",
|
|
),
|
|
("_handle_sandbox_workspace_set", {}, "params.workspace is required"),
|
|
],
|
|
)
|
|
async def test_rpc_sandbox_invalid_params_do_not_create_missing_session(
|
|
handler_name: str,
|
|
params: dict[str, str],
|
|
message: str,
|
|
) -> None:
|
|
import opensquilla.gateway.rpc_sandbox as rpc_sandbox
|
|
|
|
manager = _SessionManager()
|
|
missing_session_key = "agent:main:webchat:missing"
|
|
handler = getattr(rpc_sandbox, handler_name)
|
|
|
|
with pytest.raises(ValueError, match=message):
|
|
await handler(
|
|
{"sessionKey": missing_session_key, **params},
|
|
_ctx(manager),
|
|
)
|
|
|
|
assert missing_session_key not in manager.sessions
|
|
assert manager.created == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_sandbox_path_pick_validates_workspace_selection(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
import opensquilla.gateway.rpc_sandbox as rpc_sandbox
|
|
|
|
manager = _SessionManager()
|
|
monkeypatch.setattr(rpc_sandbox, "_pick_directory_path", lambda initial_dir=None: "/etc")
|
|
|
|
with pytest.raises(ValueError, match="sensitive_path"):
|
|
await rpc_sandbox._handle_sandbox_path_pick(
|
|
{
|
|
"sessionKey": manager.node.session_key,
|
|
"kind": "workspace",
|
|
},
|
|
_ctx(manager),
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_sandbox_path_pick_returns_valid_mount_selection(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path,
|
|
) -> None:
|
|
import opensquilla.gateway.rpc_sandbox as rpc_sandbox
|
|
|
|
manager = _SessionManager()
|
|
selected = tmp_path / "external"
|
|
selected.mkdir()
|
|
monkeypatch.setattr(
|
|
rpc_sandbox,
|
|
"_pick_directory_path",
|
|
lambda initial_dir=None: str(selected),
|
|
)
|
|
|
|
result = await rpc_sandbox._handle_sandbox_path_pick(
|
|
{
|
|
"sessionKey": manager.node.session_key,
|
|
"kind": "mount",
|
|
"access": "ro",
|
|
},
|
|
_ctx(manager),
|
|
)
|
|
|
|
assert result == {"path": str(selected), "kind": "mount"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_sandbox_path_list_returns_parent_directory_entries(
|
|
tmp_path,
|
|
) -> None:
|
|
import opensquilla.gateway.rpc_sandbox as rpc_sandbox
|
|
|
|
manager = _SessionManager()
|
|
parent = tmp_path / "parent"
|
|
child = parent / "child"
|
|
sibling = parent / "sibling"
|
|
file_entry = parent / "notes.txt"
|
|
child.mkdir(parents=True)
|
|
sibling.mkdir()
|
|
file_entry.write_text("not a directory", encoding="utf-8")
|
|
handler = getattr(rpc_sandbox, "_handle_sandbox_path_list", None)
|
|
|
|
assert callable(handler)
|
|
|
|
result = await handler(
|
|
{
|
|
"sessionKey": manager.node.session_key,
|
|
"path": str(child),
|
|
},
|
|
_ctx(manager),
|
|
)
|
|
|
|
assert result["path"] == str(child)
|
|
assert result["parentPath"] == str(parent)
|
|
entries_by_path = {entry["path"]: entry for entry in result["entries"]}
|
|
entries_by_name = {entry["name"]: entry for entry in result["entries"]}
|
|
assert entries_by_name[".."] == {
|
|
"name": "..",
|
|
"path": str(parent),
|
|
"kind": "directory",
|
|
"selectable": True,
|
|
}
|
|
assert {
|
|
str(child),
|
|
str(sibling),
|
|
str(file_entry),
|
|
}.issubset(entries_by_path)
|
|
|
|
assert entries_by_path[str(child)]["name"] == "child"
|
|
assert entries_by_path[str(child)]["kind"] == "directory"
|
|
assert entries_by_path[str(child)]["selectable"] is True
|
|
assert entries_by_path[str(sibling)]["name"] == "sibling"
|
|
assert entries_by_path[str(sibling)]["kind"] == "directory"
|
|
assert entries_by_path[str(sibling)]["selectable"] is True
|
|
assert entries_by_path[str(file_entry)]["name"] == "notes.txt"
|
|
assert entries_by_path[str(file_entry)]["kind"] == "file"
|
|
assert entries_by_path[str(file_entry)]["selectable"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_sandbox_path_list_requires_owner(tmp_path) -> None:
|
|
from opensquilla.gateway.rpc import RpcHandlerError
|
|
from opensquilla.gateway.rpc_sandbox import _handle_sandbox_path_list
|
|
|
|
manager = _SessionManager()
|
|
target = tmp_path / "target"
|
|
target.mkdir()
|
|
|
|
with pytest.raises(RpcHandlerError, match="requires owner principal"):
|
|
await _handle_sandbox_path_list(
|
|
{
|
|
"sessionKey": manager.node.session_key,
|
|
"path": str(target),
|
|
},
|
|
_ctx(manager, is_owner=False),
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_sandbox_path_list_supports_parent_row_and_child_drilldown(
|
|
tmp_path,
|
|
) -> None:
|
|
from opensquilla.gateway.rpc_sandbox import _handle_sandbox_path_list
|
|
|
|
manager = _SessionManager()
|
|
parent = tmp_path / "parent"
|
|
child = parent / "child"
|
|
grandchild = child / "grandchild"
|
|
child_file = child / "inside.txt"
|
|
child.mkdir(parents=True)
|
|
grandchild.mkdir()
|
|
child_file.write_text("inside", encoding="utf-8")
|
|
|
|
result = await _handle_sandbox_path_list(
|
|
{
|
|
"sessionKey": manager.node.session_key,
|
|
"path": str(child),
|
|
"browseChildren": True,
|
|
},
|
|
_ctx(manager),
|
|
)
|
|
|
|
assert result["path"] == str(child)
|
|
assert result["parentPath"] == str(child)
|
|
entries_by_name = {entry["name"]: entry for entry in result["entries"]}
|
|
assert entries_by_name[".."] == {
|
|
"name": "..",
|
|
"path": str(parent),
|
|
"kind": "directory",
|
|
"selectable": True,
|
|
}
|
|
assert entries_by_name["grandchild"]["path"] == str(grandchild)
|
|
assert entries_by_name["grandchild"]["kind"] == "directory"
|
|
assert entries_by_name["grandchild"]["selectable"] is True
|
|
assert entries_by_name["inside.txt"]["path"] == str(child_file)
|
|
assert entries_by_name["inside.txt"]["kind"] == "file"
|
|
assert entries_by_name["inside.txt"]["selectable"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_sandbox_path_browser_selection_is_validated_on_workspace_save(
|
|
tmp_path,
|
|
) -> None:
|
|
from opensquilla.gateway.rpc_sandbox import _handle_sandbox_workspace_set
|
|
|
|
manager = _SessionManager()
|
|
selected = tmp_path / ".aws" / "credentials"
|
|
selected.parent.mkdir()
|
|
selected.write_text("secret", encoding="utf-8")
|
|
|
|
with pytest.raises(ValueError, match="sensitive_path"):
|
|
await _handle_sandbox_workspace_set(
|
|
{
|
|
"sessionKey": manager.node.session_key,
|
|
"workspace": str(selected),
|
|
},
|
|
_ctx(manager),
|
|
)
|
|
|
|
assert manager.node.origin is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("handler_name", "params", "message"),
|
|
[
|
|
("_handle_sandbox_domain_add", {"domain": "127.0.0.1"}, "ip_literal"),
|
|
("_handle_sandbox_domain_remove", {"domain": "*.com"}, "broad_wildcard"),
|
|
("_handle_sandbox_workspace_set", {"workspace": "/"}, "sensitive_path"),
|
|
(
|
|
"_handle_sandbox_workspace_set",
|
|
{"workspacePath": "/tmp/ws/.aws/credentials"},
|
|
"sensitive_path",
|
|
),
|
|
("_handle_sandbox_mount_add", {"path": "/etc/shadow"}, "sensitive_path"),
|
|
("_handle_sandbox_mount_remove", {"path": "/"}, "sensitive_path"),
|
|
],
|
|
)
|
|
async def test_rpc_sandbox_semantic_validation_does_not_create_missing_session(
|
|
handler_name: str,
|
|
params: dict[str, str],
|
|
message: str,
|
|
) -> None:
|
|
import opensquilla.gateway.rpc_sandbox as rpc_sandbox
|
|
|
|
manager = _SessionManager()
|
|
missing_session_key = "agent:main:webchat:missing"
|
|
handler = getattr(rpc_sandbox, handler_name)
|
|
|
|
with pytest.raises(ValueError, match=message):
|
|
await handler(
|
|
{"sessionKey": missing_session_key, **params},
|
|
_ctx(manager),
|
|
)
|
|
|
|
assert missing_session_key not in manager.sessions
|
|
assert manager.created == []
|
|
|
|
|
|
def test_non_owner_route_full_hint_coerces_to_trusted() -> None:
|
|
from opensquilla.gateway.routing import build_web_route_envelope, tool_context_from_envelope
|
|
from opensquilla.sandbox.run_context import RunContext
|
|
from opensquilla.sandbox.run_mode import RunMode
|
|
|
|
envelope = build_web_route_envelope(
|
|
session_key="agent:main:webchat:abc",
|
|
principal_is_owner=False,
|
|
)
|
|
envelope.metadata["run_mode"] = "full"
|
|
envelope.metadata["sandbox_run_context"] = RunContext(
|
|
run_mode=RunMode.FULL,
|
|
source="route_metadata",
|
|
).to_origin_payload()
|
|
|
|
ctx = tool_context_from_envelope(envelope, is_owner=False)
|
|
|
|
assert ctx.run_mode == "trusted"
|
|
assert ctx.sandbox_run_context is not None
|
|
assert ctx.sandbox_run_context.run_mode is RunMode.TRUSTED
|
|
|
|
|
|
def test_non_owner_route_trusted_hint_is_preserved() -> None:
|
|
from opensquilla.gateway.routing import build_web_route_envelope, tool_context_from_envelope
|
|
from opensquilla.sandbox.run_context import RunContext
|
|
from opensquilla.sandbox.run_mode import RunMode
|
|
|
|
envelope = build_web_route_envelope(
|
|
session_key="agent:main:webchat:abc",
|
|
principal_is_owner=False,
|
|
)
|
|
envelope.metadata["run_mode"] = "trusted"
|
|
envelope.metadata["sandbox_run_context"] = RunContext(
|
|
run_mode=RunMode.TRUSTED,
|
|
source="route_metadata",
|
|
).to_origin_payload()
|
|
|
|
ctx = tool_context_from_envelope(envelope, is_owner=False)
|
|
|
|
assert ctx.run_mode == "trusted"
|
|
assert ctx.sandbox_run_context is not None
|
|
assert ctx.sandbox_run_context.run_mode is RunMode.TRUSTED
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_sandbox_resume_clears_denial_pause(monkeypatch) -> None:
|
|
# The owner-scoped resume RPC is the recovery surface for the sticky denial
|
|
# pause (issue #469): it must clear the pause so the run can continue.
|
|
from opensquilla.gateway.rpc_sandbox import _handle_sandbox_resume
|
|
from opensquilla.sandbox.governance import DenialLedger
|
|
|
|
ledger = DenialLedger(threshold=3)
|
|
manager = _SessionManager()
|
|
key = manager.node.session_key
|
|
await ledger.mark_paused(key)
|
|
monkeypatch.setattr(
|
|
"opensquilla.sandbox.integration.get_runtime",
|
|
lambda: SimpleNamespace(ledger=ledger),
|
|
)
|
|
|
|
result = await _handle_sandbox_resume({"sessionKey": key}, _ctx(manager))
|
|
|
|
assert result["resumed"] is True
|
|
assert result["autonomousPaused"] is False
|
|
assert await ledger.is_paused(key) is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_sandbox_resume_rejects_non_owner() -> None:
|
|
from opensquilla.gateway.rpc import RpcHandlerError
|
|
from opensquilla.gateway.rpc_sandbox import _handle_sandbox_resume
|
|
|
|
manager = _SessionManager()
|
|
with pytest.raises(RpcHandlerError, match="requires owner principal"):
|
|
await _handle_sandbox_resume(
|
|
{"sessionKey": manager.node.session_key},
|
|
_ctx(manager, is_owner=False),
|
|
)
|