feat(box): SaaS guard to force a single global sandbox scope

Add system.limitation.force_box_session_id_template: when non-empty it
overrides every pipeline's box-session-id-template at resolve time, pinning
all queries to one shared sandbox (e.g. {global}). This is the authoritative,
unbypassable guard — it runs on every exec call, so editing the pipeline
config via API cannot escape it. The web UI locks the Sandbox Scope selector
via a combined box_scope_editable flag (box available AND not forced).
This commit is contained in:
RockChinQ
2026-06-08 06:38:52 -04:00
parent c460bd7814
commit 3b5e89f17f
6 changed files with 138 additions and 20 deletions
+70 -1
View File
@@ -153,6 +153,7 @@ def make_app(
host_root: str = '',
workspace_quota_mb: int | None = None,
enabled: bool = True,
force_box_session_id_template: str = '',
):
box_config = {
'enabled': enabled,
@@ -171,7 +172,12 @@ def make_app(
return SimpleNamespace(
logger=logger,
instance_config=SimpleNamespace(data={'box': box_config}),
instance_config=SimpleNamespace(
data={
'box': box_config,
'system': {'limitation': {'force_box_session_id_template': force_box_session_id_template}},
}
),
)
@@ -362,6 +368,69 @@ async def test_box_service_session_id_falls_back_to_query_id_for_synthetic_queri
assert backend.start_calls == ['query_7']
@pytest.mark.asyncio
async def test_box_service_forced_global_scope_overrides_pipeline_template():
"""SaaS guard: a non-empty ``force_box_session_id_template`` pins every
query to one shared sandbox regardless of the pipeline's own scope."""
logger = Mock()
backend = FakeBackend(logger)
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
service = BoxService(
make_app(logger, force_box_session_id_template='{global}'),
client=_InProcessBoxRuntimeClient(logger, runtime),
)
await service.initialize()
# Two distinct callers that would otherwise get separate sandboxes.
q1 = pipeline_query.Query.model_construct(query_id=1, launcher_type='group', launcher_id='room-1')
q2 = pipeline_query.Query.model_construct(query_id=2, launcher_type='person', launcher_id='alice')
r1 = await service.execute_tool({'command': 'pwd'}, q1)
r2 = await service.execute_tool({'command': 'pwd'}, q2)
assert r1['session_id'] == 'global'
assert r2['session_id'] == 'global'
# Only one sandbox was ever started — the shared global one.
assert backend.start_calls == ['global']
def test_box_service_forced_template_ignores_pipeline_config():
"""The forced template wins even when the pipeline explicitly sets a
per-user scope — proving the override is not bypassable via pipeline config."""
logger = Mock()
service = BoxService(
make_app(logger, force_box_session_id_template='{global}'),
client=Mock(spec=BoxRuntimeClient),
)
query = pipeline_query.Query.model_construct(
query_id=7,
launcher_type='person',
launcher_id='test_user',
sender_id='test_user',
pipeline_config={'ai': {'local-agent': {'box-session-id-template': '{launcher_type}_{launcher_id}_{sender_id}'}}},
)
assert service.resolve_box_session_id(query) == 'global'
def test_box_service_empty_forced_template_respects_pipeline_config():
"""An empty/whitespace forced template is a no-op: the pipeline's own
scope template is honoured (default non-SaaS behaviour)."""
logger = Mock()
service = BoxService(
make_app(logger, force_box_session_id_template=' '),
client=Mock(spec=BoxRuntimeClient),
)
query = pipeline_query.Query.model_construct(
query_id=7,
launcher_type='group',
launcher_id='room-1',
pipeline_config={'ai': {'local-agent': {'box-session-id-template': '{launcher_type}_{launcher_id}'}}},
)
assert service.resolve_box_session_id(query) == 'group_room-1'
@pytest.mark.asyncio
async def test_box_service_fails_closed_when_backend_unavailable():
logger = Mock()