mjalmoudy94
a0c8473eb8
The execution boundary for v2. Adds the local docker/podman sandbox adapter and the hardened host subprocess runner every other delivery adapter will use. - adapters/process.py: the one place the platform shells out. argv-only (no shell), start_new_session + killpg on timeout, concurrent stdout/stderr drains with a byte cap, explicit minimal env (never os.environ), and a missing binary reported as spawn_failed rather than an exception. - adapters/docker_sandbox.py: DockerSandboxAdapter over SandboxProviderPort. --user 1000:1000, --cap-drop ALL, no-new-privileges, pids/cpu/memory limits, --read-only with tmpfs scratch, one bind mount (the work tree, never the git dir), labels, no published ports. Path-safe file access; secrets refused in `docker exec -e`. Extras for later phases: start_demo_container (loopback publish only), list_managed_containers, remove_container. - registry: SANDBOX_PROVIDER=docker, cached singleton so the workspace registry survives create_workspace -> start_session. - sandbox_gateway worker: reap_orphans() removes labelled containers no live session points at, audited as sandbox.reap; run_pass() does both passes. - sandbox router: every mutating route now requires the platform owner (PLAN-v2 §7.7); read-only status routes stay open to org members. - sandbox-panel: canManageInfra prop (fails closed) hides the create form, the session panel and the action buttons from non-owners. Tests: 346 API (+37), 67 web (+3); ruff, mypy, tsc and lint clean. Three existing tests now drive the sandbox as the platform owner, since that is who may touch it. Report: docs/plan/reports/agent-3-phase-1.md
386 lines
14 KiB
Python
386 lines
14 KiB
Python
"""Repositories for the Sandbox Gateway.
|
|
|
|
Every tenant-scoped read joins back to ``workspaces.organization_id`` so a
|
|
workspace or session id from another tenant resolves to ``None``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.domain.base import utcnow
|
|
from app.domain.sandbox.models import (
|
|
CommandRun,
|
|
NetworkPolicy,
|
|
PreviewUrl,
|
|
SandboxProviderKind,
|
|
SandboxResourceLimit,
|
|
SandboxSession,
|
|
SandboxTemplate,
|
|
SessionStatus,
|
|
Snapshot,
|
|
Workspace,
|
|
WorkspaceStatus,
|
|
)
|
|
from app.infrastructure.db.models import (
|
|
CommandRunRow,
|
|
PreviewUrlRow,
|
|
SandboxResourceLimitRow,
|
|
SandboxSessionRow,
|
|
SandboxTemplateRow,
|
|
SnapshotRow,
|
|
WorkspaceRow,
|
|
)
|
|
|
|
|
|
def _template(row: SandboxTemplateRow) -> SandboxTemplate:
|
|
return SandboxTemplate(
|
|
id=row.id, name=row.name, provider=row.provider, image=row.image,
|
|
description=row.description, default_cpu=row.default_cpu,
|
|
default_memory_mb=row.default_memory_mb, default_disk_mb=row.default_disk_mb,
|
|
is_system=row.is_system, created_at=row.created_at,
|
|
)
|
|
|
|
|
|
def _workspace(row: WorkspaceRow) -> Workspace:
|
|
return Workspace(
|
|
id=row.id, organization_id=row.organization_id, project_id=row.project_id,
|
|
template_id=row.template_id, provider=row.provider,
|
|
external_workspace_id=row.external_workspace_id,
|
|
status=WorkspaceStatus(row.status), created_at=row.created_at,
|
|
)
|
|
|
|
|
|
def _session(row: SandboxSessionRow) -> SandboxSession:
|
|
return SandboxSession(
|
|
id=row.id, workspace_id=row.workspace_id,
|
|
external_session_id=row.external_session_id,
|
|
status=SessionStatus(row.status), started_at=row.started_at,
|
|
stopped_at=row.stopped_at, last_active_at=row.last_active_at,
|
|
)
|
|
|
|
|
|
class SandboxTemplateRepository:
|
|
def __init__(self, session: AsyncSession) -> None:
|
|
self.session = session
|
|
|
|
async def list(self) -> list[SandboxTemplate]:
|
|
stmt = select(SandboxTemplateRow).order_by(SandboxTemplateRow.name)
|
|
return [_template(r) for r in (await self.session.execute(stmt)).scalars()]
|
|
|
|
async def get(self, template_id: str) -> SandboxTemplate | None:
|
|
row = await self.session.get(SandboxTemplateRow, template_id)
|
|
return _template(row) if row else None
|
|
|
|
async def get_by_name(self, name: str) -> SandboxTemplate | None:
|
|
stmt = select(SandboxTemplateRow).where(SandboxTemplateRow.name == name)
|
|
row = (await self.session.execute(stmt)).scalar_one_or_none()
|
|
return _template(row) if row else None
|
|
|
|
async def create(self, template: SandboxTemplate) -> SandboxTemplate:
|
|
row = SandboxTemplateRow(
|
|
id=template.id, name=template.name, provider=str(template.provider),
|
|
image=template.image, description=template.description,
|
|
default_cpu=template.default_cpu,
|
|
default_memory_mb=template.default_memory_mb,
|
|
default_disk_mb=template.default_disk_mb, is_system=template.is_system,
|
|
created_at=template.created_at,
|
|
)
|
|
self.session.add(row)
|
|
await self.session.flush()
|
|
return _template(row)
|
|
|
|
|
|
class WorkspaceRepository:
|
|
def __init__(self, session: AsyncSession) -> None:
|
|
self.session = session
|
|
|
|
async def get(self, workspace_id: str) -> Workspace | None:
|
|
"""Unscoped lookup. Platform components only — tenant paths use
|
|
:meth:`get_for_org` so a foreign id resolves to ``None``."""
|
|
row = await self.session.get(WorkspaceRow, workspace_id)
|
|
return _workspace(row) if row else None
|
|
|
|
async def get_for_org(self, organization_id: str, workspace_id: str) -> Workspace | None:
|
|
stmt = select(WorkspaceRow).where(
|
|
WorkspaceRow.id == workspace_id,
|
|
WorkspaceRow.organization_id == organization_id,
|
|
)
|
|
row = (await self.session.execute(stmt)).scalar_one_or_none()
|
|
return _workspace(row) if row else None
|
|
|
|
async def list_for_project(
|
|
self, project_id: str, *, limit: int = 100, offset: int = 0
|
|
) -> list[Workspace]:
|
|
stmt = (
|
|
select(WorkspaceRow)
|
|
.where(WorkspaceRow.project_id == project_id)
|
|
.order_by(WorkspaceRow.created_at.desc())
|
|
.limit(min(limit, 500))
|
|
.offset(offset)
|
|
)
|
|
return [_workspace(r) for r in (await self.session.execute(stmt)).scalars()]
|
|
|
|
async def create(self, workspace: Workspace) -> Workspace:
|
|
row = WorkspaceRow(
|
|
id=workspace.id, organization_id=workspace.organization_id,
|
|
project_id=workspace.project_id, template_id=workspace.template_id,
|
|
provider=str(workspace.provider),
|
|
external_workspace_id=workspace.external_workspace_id,
|
|
status=str(workspace.status), created_at=workspace.created_at,
|
|
)
|
|
self.session.add(row)
|
|
await self.session.flush()
|
|
return _workspace(row)
|
|
|
|
async def set_status(self, workspace_id: str, status: WorkspaceStatus) -> None:
|
|
row = await self.session.get(WorkspaceRow, workspace_id)
|
|
if row:
|
|
row.status = str(status)
|
|
await self.session.flush()
|
|
|
|
|
|
class SandboxSessionRepository:
|
|
def __init__(self, session: AsyncSession) -> None:
|
|
self.session = session
|
|
|
|
async def get(self, session_id: str) -> SandboxSession | None:
|
|
row = await self.session.get(SandboxSessionRow, session_id)
|
|
return _session(row) if row else None
|
|
|
|
async def get_for_org(self, organization_id: str, session_id: str) -> SandboxSession | None:
|
|
"""Resolve a session through its workspace, which carries the tenant."""
|
|
stmt = (
|
|
select(SandboxSessionRow)
|
|
.join(WorkspaceRow, WorkspaceRow.id == SandboxSessionRow.workspace_id)
|
|
.where(
|
|
SandboxSessionRow.id == session_id,
|
|
WorkspaceRow.organization_id == organization_id,
|
|
)
|
|
)
|
|
row = (await self.session.execute(stmt)).scalar_one_or_none()
|
|
return _session(row) if row else None
|
|
|
|
async def list_for_workspace(self, workspace_id: str) -> list[SandboxSession]:
|
|
stmt = (
|
|
select(SandboxSessionRow)
|
|
.where(SandboxSessionRow.workspace_id == workspace_id)
|
|
.order_by(SandboxSessionRow.started_at.desc())
|
|
)
|
|
return [_session(r) for r in (await self.session.execute(stmt)).scalars()]
|
|
|
|
async def list_stale_active(
|
|
self, *, idle_before: datetime, limit: int = 200
|
|
) -> list[SandboxSession]:
|
|
"""Active sessions untouched since ``idle_before``.
|
|
|
|
The gateway reaper's query. Only ``starting``/``running``/``paused``
|
|
count as active — a stopped session is already gone and must not be
|
|
stopped twice.
|
|
"""
|
|
stmt = (
|
|
select(SandboxSessionRow)
|
|
.where(
|
|
SandboxSessionRow.status.in_(
|
|
[
|
|
str(SessionStatus.STARTING),
|
|
str(SessionStatus.RUNNING),
|
|
str(SessionStatus.PAUSED),
|
|
]
|
|
),
|
|
SandboxSessionRow.last_active_at < idle_before,
|
|
)
|
|
.order_by(SandboxSessionRow.last_active_at.asc())
|
|
.limit(min(limit, 500))
|
|
)
|
|
return [_session(r) for r in (await self.session.execute(stmt)).scalars()]
|
|
|
|
async def list_active_external_ids(self, *, limit: int = 5_000) -> set[str]:
|
|
"""External ids of every session that is not stopped.
|
|
|
|
The orphan reaper's query: a container whose id is not in this set has
|
|
no live row pointing at it, so it is holding resources for nobody.
|
|
"""
|
|
stmt = (
|
|
select(SandboxSessionRow.external_session_id)
|
|
.where(
|
|
SandboxSessionRow.status.in_(
|
|
[
|
|
str(SessionStatus.STARTING),
|
|
str(SessionStatus.RUNNING),
|
|
str(SessionStatus.PAUSED),
|
|
]
|
|
),
|
|
SandboxSessionRow.external_session_id.is_not(None),
|
|
)
|
|
.limit(min(limit, 20_000))
|
|
)
|
|
rows = (await self.session.execute(stmt)).scalars().all()
|
|
return {str(value) for value in rows if value}
|
|
|
|
async def create(self, sandbox_session: SandboxSession) -> SandboxSession:
|
|
row = SandboxSessionRow(
|
|
id=sandbox_session.id, workspace_id=sandbox_session.workspace_id,
|
|
external_session_id=sandbox_session.external_session_id,
|
|
status=str(sandbox_session.status), started_at=sandbox_session.started_at,
|
|
stopped_at=sandbox_session.stopped_at,
|
|
last_active_at=sandbox_session.last_active_at,
|
|
)
|
|
self.session.add(row)
|
|
await self.session.flush()
|
|
return _session(row)
|
|
|
|
async def set_status(
|
|
self, session_id: str, status: SessionStatus, *, stopped: bool = False
|
|
) -> None:
|
|
row = await self.session.get(SandboxSessionRow, session_id)
|
|
if row:
|
|
row.status = str(status)
|
|
row.last_active_at = utcnow()
|
|
if stopped:
|
|
row.stopped_at = utcnow()
|
|
await self.session.flush()
|
|
|
|
|
|
class CommandRunRepository:
|
|
def __init__(self, session: AsyncSession) -> None:
|
|
self.session = session
|
|
|
|
async def list_for_session(self, session_id: str) -> list[CommandRun]:
|
|
stmt = (
|
|
select(CommandRunRow)
|
|
.where(CommandRunRow.session_id == session_id)
|
|
.order_by(CommandRunRow.created_at)
|
|
)
|
|
return [
|
|
CommandRun(
|
|
id=r.id, session_id=r.session_id, command=r.command,
|
|
exit_code=r.exit_code, stdout_ref=r.stdout_ref, stderr_ref=r.stderr_ref,
|
|
duration_ms=r.duration_ms, truncated=r.truncated, created_at=r.created_at,
|
|
)
|
|
for r in (await self.session.execute(stmt)).scalars()
|
|
]
|
|
|
|
async def create(self, run: CommandRun) -> CommandRun:
|
|
row = CommandRunRow(
|
|
id=run.id, session_id=run.session_id, command=run.command,
|
|
exit_code=run.exit_code, stdout_ref=run.stdout_ref, stderr_ref=run.stderr_ref,
|
|
duration_ms=run.duration_ms, truncated=run.truncated, created_at=run.created_at,
|
|
)
|
|
self.session.add(row)
|
|
await self.session.flush()
|
|
return run
|
|
|
|
|
|
class SnapshotRepository:
|
|
def __init__(self, session: AsyncSession) -> None:
|
|
self.session = session
|
|
|
|
async def list_for_workspace(self, workspace_id: str) -> list[Snapshot]:
|
|
stmt = (
|
|
select(SnapshotRow)
|
|
.where(SnapshotRow.workspace_id == workspace_id)
|
|
.order_by(SnapshotRow.created_at.desc())
|
|
)
|
|
return [
|
|
Snapshot(
|
|
id=r.id, workspace_id=r.workspace_id, name=r.name,
|
|
external_snapshot_id=r.external_snapshot_id, created_at=r.created_at,
|
|
)
|
|
for r in (await self.session.execute(stmt)).scalars()
|
|
]
|
|
|
|
async def create(self, snapshot: Snapshot) -> Snapshot:
|
|
row = SnapshotRow(
|
|
id=snapshot.id, workspace_id=snapshot.workspace_id, name=snapshot.name,
|
|
external_snapshot_id=snapshot.external_snapshot_id,
|
|
created_at=snapshot.created_at,
|
|
)
|
|
self.session.add(row)
|
|
await self.session.flush()
|
|
return snapshot
|
|
|
|
|
|
class PreviewUrlRepository:
|
|
def __init__(self, session: AsyncSession) -> None:
|
|
self.session = session
|
|
|
|
async def list_for_workspace(self, workspace_id: str) -> list[PreviewUrl]:
|
|
stmt = select(PreviewUrlRow).where(PreviewUrlRow.workspace_id == workspace_id)
|
|
return [
|
|
PreviewUrl(
|
|
id=r.id, workspace_id=r.workspace_id, port=r.port, url=r.url,
|
|
created_at=r.created_at,
|
|
)
|
|
for r in (await self.session.execute(stmt)).scalars()
|
|
]
|
|
|
|
async def create(self, preview: PreviewUrl) -> PreviewUrl:
|
|
row = PreviewUrlRow(
|
|
id=preview.id, workspace_id=preview.workspace_id, port=preview.port,
|
|
url=preview.url, created_at=preview.created_at,
|
|
)
|
|
self.session.add(row)
|
|
await self.session.flush()
|
|
return preview
|
|
|
|
|
|
class ResourceLimitRepository:
|
|
def __init__(self, session: AsyncSession) -> None:
|
|
self.session = session
|
|
|
|
async def get_for_scope(self, scope: str, scope_id: str | None) -> SandboxResourceLimit | None:
|
|
stmt = select(SandboxResourceLimitRow).where(
|
|
SandboxResourceLimitRow.scope == scope,
|
|
SandboxResourceLimitRow.scope_id == scope_id,
|
|
)
|
|
row = (await self.session.execute(stmt)).scalar_one_or_none()
|
|
if row is None:
|
|
return None
|
|
return SandboxResourceLimit(
|
|
id=row.id, scope=row.scope, scope_id=row.scope_id, cpu=row.cpu,
|
|
memory_mb=row.memory_mb, disk_mb=row.disk_mb,
|
|
timeout_seconds=row.timeout_seconds, max_output_bytes=row.max_output_bytes,
|
|
network_policy=NetworkPolicy(row.network_policy),
|
|
)
|
|
|
|
async def upsert(self, limit: SandboxResourceLimit) -> SandboxResourceLimit:
|
|
stmt = select(SandboxResourceLimitRow).where(
|
|
SandboxResourceLimitRow.scope == limit.scope,
|
|
SandboxResourceLimitRow.scope_id == limit.scope_id,
|
|
)
|
|
row = (await self.session.execute(stmt)).scalar_one_or_none()
|
|
if row is None:
|
|
row = SandboxResourceLimitRow(
|
|
id=limit.id, scope=limit.scope, scope_id=limit.scope_id, cpu=limit.cpu,
|
|
memory_mb=limit.memory_mb, disk_mb=limit.disk_mb,
|
|
timeout_seconds=limit.timeout_seconds,
|
|
max_output_bytes=limit.max_output_bytes,
|
|
network_policy=str(limit.network_policy),
|
|
)
|
|
self.session.add(row)
|
|
else:
|
|
row.cpu = limit.cpu
|
|
row.memory_mb = limit.memory_mb
|
|
row.disk_mb = limit.disk_mb
|
|
row.timeout_seconds = limit.timeout_seconds
|
|
row.max_output_bytes = limit.max_output_bytes
|
|
row.network_policy = str(limit.network_policy)
|
|
await self.session.flush()
|
|
return limit
|
|
|
|
|
|
__all__ = [
|
|
"CommandRunRepository",
|
|
"PreviewUrlRepository",
|
|
"ResourceLimitRepository",
|
|
"SandboxProviderKind",
|
|
"SandboxSessionRepository",
|
|
"SandboxTemplateRepository",
|
|
"SnapshotRepository",
|
|
"WorkspaceRepository",
|
|
]
|