mjalmoudy94
f1e5df29ce
- Backend: ConnectionManager, Redis-backed bus with in-process fallback, channel authorization, and typed emit helpers wired into workers (ai_runtime, sandbox_gateway), workforce/sandbox/billing/audit/error services. - Frontend: single shared WS client with backoff + heartbeat, RealtimeProvider, useLiveChannel/useLiveEvent hooks, live indicators on task detail, manager chat, dashboard, traces, sandbox terminal, admin errors, and billing. - Contracts: WsEvent/WsServerFrame/WsClientFrame types mirroring the port. All 255 backend tests pass; web typecheck and build pass.
484 lines
19 KiB
Python
484 lines
19 KiB
Python
"""Sandbox Gateway use cases.
|
|
|
|
The security posture is the whole point of this module:
|
|
|
|
* **No host execution path.** The only way to run anything is through
|
|
``SandboxProviderPort``. There is no fallback that shells out locally.
|
|
* **Limits come from policy, never from the caller.** A request cannot ask for a
|
|
longer timeout, more memory, or a bigger output cap than its scope allows.
|
|
* **Egress is deny-by-default.** The stored ``network_policy`` defaults to
|
|
``deny_all``; nothing here opens it implicitly.
|
|
* **Every command is metered.** CPU seconds are written to the usage ledger, so
|
|
sandbox time is billable and auditable.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
from datetime import datetime
|
|
|
|
from app.application.ports import CommandResult, SandboxProviderPort, UsageRecord
|
|
from app.application.ports.realtime import (
|
|
emit_sandbox_command,
|
|
emit_sandbox_session,
|
|
)
|
|
from app.core.config import settings
|
|
from app.core.errors import ConflictError, NotFoundError, ProviderUnavailableError, ValidationError
|
|
from app.domain.ai_control.models import RateUnit
|
|
from app.domain.base import utcnow
|
|
from app.domain.sandbox.models import (
|
|
CommandRun,
|
|
NetworkPolicy,
|
|
PreviewUrl,
|
|
SandboxProviderKind,
|
|
SandboxResourceLimit,
|
|
SandboxSession,
|
|
SessionStatus,
|
|
Snapshot,
|
|
Workspace,
|
|
WorkspaceStatus,
|
|
)
|
|
from app.infrastructure.repositories.ai_control import RateCardRepository
|
|
from app.infrastructure.repositories.audit import UsageLedgerRepository
|
|
from app.infrastructure.repositories.projects import ProjectRepository
|
|
from app.infrastructure.repositories.sandbox import (
|
|
CommandRunRepository,
|
|
PreviewUrlRepository,
|
|
ResourceLimitRepository,
|
|
SandboxSessionRepository,
|
|
SandboxTemplateRepository,
|
|
SnapshotRepository,
|
|
WorkspaceRepository,
|
|
)
|
|
|
|
#: Hard ceiling, independent of policy. A policy may be lower, never higher.
|
|
ABSOLUTE_MAX_TIMEOUT_SECONDS = 900
|
|
ABSOLUTE_MAX_OUTPUT_BYTES = 2_000_000
|
|
MAX_COMMAND_LENGTH = 4_000
|
|
|
|
|
|
class SandboxService:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
templates: SandboxTemplateRepository,
|
|
workspaces: WorkspaceRepository,
|
|
sessions: SandboxSessionRepository,
|
|
commands: CommandRunRepository,
|
|
snapshots: SnapshotRepository,
|
|
previews: PreviewUrlRepository,
|
|
limits: ResourceLimitRepository,
|
|
rate_cards: RateCardRepository,
|
|
ledger: UsageLedgerRepository,
|
|
projects: ProjectRepository,
|
|
provider: SandboxProviderPort,
|
|
) -> None:
|
|
self.templates = templates
|
|
self.workspaces = workspaces
|
|
self.sessions = sessions
|
|
self.commands = commands
|
|
self.snapshots = snapshots
|
|
self.previews = previews
|
|
self.limits = limits
|
|
self.rate_cards = rate_cards
|
|
self.ledger = ledger
|
|
self.projects = projects
|
|
self.provider = provider
|
|
|
|
# -- limits --------------------------------------------------------------
|
|
|
|
async def effective_limits(
|
|
self, *, organization_id: str, project_id: str | None = None
|
|
) -> SandboxResourceLimit:
|
|
"""Resolve the strictest applicable policy.
|
|
|
|
Project scope wins over organization scope; the absolute ceiling always
|
|
wins over both.
|
|
"""
|
|
if project_id:
|
|
scoped = await self.limits.get_for_scope("project", project_id)
|
|
if scoped is not None:
|
|
return self._clamp(scoped)
|
|
|
|
org_scoped = await self.limits.get_for_scope("organization", organization_id)
|
|
if org_scoped is not None:
|
|
return self._clamp(org_scoped)
|
|
|
|
return self._clamp(
|
|
SandboxResourceLimit(
|
|
scope="platform",
|
|
scope_id=None,
|
|
timeout_seconds=settings.sandbox_command_timeout_seconds,
|
|
max_output_bytes=settings.sandbox_max_output_bytes,
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def _clamp(limit: SandboxResourceLimit) -> SandboxResourceLimit:
|
|
limit.timeout_seconds = max(
|
|
1, min(int(limit.timeout_seconds), ABSOLUTE_MAX_TIMEOUT_SECONDS)
|
|
)
|
|
limit.max_output_bytes = max(
|
|
1024, min(int(limit.max_output_bytes), ABSOLUTE_MAX_OUTPUT_BYTES)
|
|
)
|
|
limit.memory_mb = max(128, int(limit.memory_mb))
|
|
limit.disk_mb = max(256, int(limit.disk_mb))
|
|
return limit
|
|
|
|
# -- templates -----------------------------------------------------------
|
|
|
|
async def list_templates(self):
|
|
return await self.templates.list()
|
|
|
|
# -- workspaces ----------------------------------------------------------
|
|
|
|
async def create_workspace(
|
|
self, *, organization_id: str, project_id: str, template_id: str
|
|
) -> Workspace:
|
|
template = await self.templates.get(template_id)
|
|
if template is None:
|
|
raise NotFoundError("قالب سندباکس یافت نشد.")
|
|
|
|
provider_kind = str(template.provider) or SandboxProviderKind.MOCK
|
|
try:
|
|
info = await self.provider.create_workspace(
|
|
organization_id=organization_id,
|
|
project_id=project_id,
|
|
template_id=template_id,
|
|
image=template.image,
|
|
)
|
|
except Exception as exc: # noqa: BLE001 - surface as a provider outage
|
|
raise ProviderUnavailableError(
|
|
"ساخت محیط اجرای ایزوله ناموفق بود."
|
|
) from exc
|
|
|
|
workspace = await self.workspaces.create(
|
|
Workspace(
|
|
organization_id=organization_id,
|
|
project_id=project_id,
|
|
template_id=template_id,
|
|
provider=provider_kind,
|
|
external_workspace_id=info.external_id,
|
|
status=WorkspaceStatus.READY,
|
|
)
|
|
)
|
|
return workspace
|
|
|
|
async def list_workspaces(
|
|
self,
|
|
*,
|
|
organization_id: str,
|
|
project_id: str,
|
|
limit: int = 100,
|
|
offset: int = 0,
|
|
) -> list[Workspace]:
|
|
"""Workspaces for a project the caller's tenant actually owns."""
|
|
project = await self.projects.get(project_id)
|
|
# 404 rather than an empty list, so a foreign project id is
|
|
# indistinguishable from a non-existent one.
|
|
if project is None or project.organization_id != organization_id:
|
|
raise NotFoundError("پروژه یافت نشد.")
|
|
return await self.workspaces.list_for_project(
|
|
project_id, limit=limit, offset=offset
|
|
)
|
|
|
|
async def get_workspace_or_404(
|
|
self, *, organization_id: str, workspace_id: str
|
|
) -> Workspace:
|
|
workspace = await self.workspaces.get_for_org(organization_id, workspace_id)
|
|
if workspace is None:
|
|
raise NotFoundError("محیط اجرا یافت نشد.")
|
|
return workspace
|
|
|
|
# -- sessions ------------------------------------------------------------
|
|
|
|
async def start_session(
|
|
self, *, organization_id: str, workspace_id: str
|
|
) -> SandboxSession:
|
|
workspace = await self.get_workspace_or_404(
|
|
organization_id=organization_id, workspace_id=workspace_id
|
|
)
|
|
if workspace.status == WorkspaceStatus.DESTROYED:
|
|
raise ConflictError("این محیط اجرا از بین رفته است.")
|
|
if not workspace.external_workspace_id:
|
|
raise ProviderUnavailableError("محیط اجرا در ارائهدهنده ثبت نشده است.")
|
|
|
|
info = await self.provider.start_session(
|
|
workspace_external_id=workspace.external_workspace_id
|
|
)
|
|
session = await self.sessions.create(
|
|
SandboxSession(
|
|
workspace_id=workspace.id,
|
|
external_session_id=info.external_id,
|
|
status=SessionStatus.RUNNING,
|
|
)
|
|
)
|
|
await self.workspaces.set_status(workspace.id, WorkspaceStatus.READY)
|
|
await emit_sandbox_session(
|
|
session_id=session.id,
|
|
organization_id=organization_id,
|
|
status=SessionStatus.RUNNING.value,
|
|
workspace_id=workspace.id,
|
|
)
|
|
return session
|
|
|
|
async def get_session_or_404(
|
|
self, *, organization_id: str, session_id: str
|
|
) -> SandboxSession:
|
|
session = await self.sessions.get_for_org(organization_id, session_id)
|
|
if session is None:
|
|
raise NotFoundError("نشست اجرا یافت نشد.")
|
|
return session
|
|
|
|
async def stop_session(self, *, organization_id: str, session_id: str) -> SandboxSession:
|
|
session = await self.get_session_or_404(
|
|
organization_id=organization_id, session_id=session_id
|
|
)
|
|
if not session.external_session_id:
|
|
raise ProviderUnavailableError("نشست در ارائهدهنده ثبت نشده است.")
|
|
await self.provider.destroy(session_external_id=session.external_session_id)
|
|
await self.sessions.set_status(session.id, SessionStatus.STOPPED, stopped=True)
|
|
refreshed = await self.sessions.get(session.id)
|
|
await emit_sandbox_session(
|
|
session_id=session.id,
|
|
organization_id=organization_id,
|
|
status=SessionStatus.STOPPED.value,
|
|
)
|
|
return refreshed or session
|
|
|
|
# -- commands ------------------------------------------------------------
|
|
|
|
async def run_command(
|
|
self,
|
|
*,
|
|
organization_id: str,
|
|
session_id: str,
|
|
command: str,
|
|
user_id: str | None = None,
|
|
) -> tuple[CommandRun, CommandResult]:
|
|
session = await self.get_session_or_404(
|
|
organization_id=organization_id, session_id=session_id
|
|
)
|
|
if not session.external_session_id:
|
|
raise ProviderUnavailableError("نشست در ارائهدهنده ثبت نشده است.")
|
|
if session.status != SessionStatus.RUNNING:
|
|
raise ConflictError("نشست در وضعیت اجرا نیست.")
|
|
|
|
cleaned = command.strip()
|
|
if not cleaned:
|
|
raise ValidationError("دستور نمیتواند خالی باشد.")
|
|
if len(cleaned) > MAX_COMMAND_LENGTH:
|
|
raise ValidationError("دستور بیش از حد طولانی است.")
|
|
|
|
workspace = await self.workspaces.get_for_org(
|
|
organization_id, session.workspace_id
|
|
)
|
|
limits = await self.effective_limits(
|
|
organization_id=organization_id,
|
|
project_id=workspace.project_id if workspace else None,
|
|
)
|
|
|
|
result = await self.provider.run_command(
|
|
session_external_id=session.external_session_id,
|
|
command=cleaned,
|
|
# Timeout and output cap are decided here, not by the caller.
|
|
timeout_seconds=limits.timeout_seconds,
|
|
max_output_bytes=limits.max_output_bytes,
|
|
env={},
|
|
)
|
|
|
|
run = await self.commands.create(
|
|
CommandRun(
|
|
session_id=session.id,
|
|
command=cleaned,
|
|
exit_code=result.exit_code,
|
|
stdout_ref=result.stdout[: limits.max_output_bytes],
|
|
stderr_ref=result.stderr[: limits.max_output_bytes],
|
|
duration_ms=result.duration_ms,
|
|
truncated=result.truncated,
|
|
)
|
|
)
|
|
await self.sessions.set_status(session.id, SessionStatus.RUNNING)
|
|
|
|
# Live push: the terminal panel on the project page streams the command
|
|
# and its output as the row is written, not on the next poll.
|
|
await emit_sandbox_command(
|
|
session_id=session.id,
|
|
organization_id=organization_id,
|
|
command_run_id=run.id,
|
|
command=cleaned,
|
|
exit_code=result.exit_code,
|
|
stdout=run.stdout_ref,
|
|
stderr=run.stderr_ref,
|
|
duration_ms=result.duration_ms,
|
|
truncated=result.truncated,
|
|
)
|
|
|
|
# CPU seconds are billed. 1 second of wall time ~= 1 CPU second at MVP.
|
|
cpu_seconds = max(0.0, result.duration_ms / 1000.0)
|
|
if cpu_seconds > 0:
|
|
card = await self.rate_cards.get_by_unit(RateUnit.SANDBOX_CPU_SECOND)
|
|
unit_price = card.customer_price if card else 0.0
|
|
internal_cost = card.internal_cost_estimate if card else 0.0
|
|
await self.ledger.record(
|
|
usage=UsageRecord(
|
|
organization_id=organization_id,
|
|
project_id=workspace.project_id if workspace else None,
|
|
user_id=user_id,
|
|
unit=RateUnit.SANDBOX_CPU_SECOND,
|
|
quantity=cpu_seconds,
|
|
unit_price=unit_price,
|
|
internal_cost_estimate=cpu_seconds * internal_cost,
|
|
total_price=cpu_seconds * unit_price,
|
|
metadata_json={"session_id": session.id, "exit_code": result.exit_code},
|
|
)
|
|
)
|
|
|
|
return run, result
|
|
|
|
async def list_commands(self, *, session_id: str) -> list[CommandRun]:
|
|
return await self.commands.list_for_session(session_id)
|
|
|
|
# -- files ---------------------------------------------------------------
|
|
|
|
async def read_file(
|
|
self, *, organization_id: str, session_id: str, path: str
|
|
) -> str:
|
|
session = await self.get_session_or_404(
|
|
organization_id=organization_id, session_id=session_id
|
|
)
|
|
if not session.external_session_id:
|
|
raise ProviderUnavailableError("نشست در ارائهدهنده ثبت نشده است.")
|
|
return await self.provider.read_file(
|
|
session_external_id=session.external_session_id, path=path
|
|
)
|
|
|
|
async def write_file(
|
|
self, *, organization_id: str, session_id: str, path: str, content: str
|
|
) -> None:
|
|
session = await self.get_session_or_404(
|
|
organization_id=organization_id, session_id=session_id
|
|
)
|
|
if not session.external_session_id:
|
|
raise ProviderUnavailableError("نشست در ارائهدهنده ثبت نشده است.")
|
|
await self.provider.write_file(
|
|
session_external_id=session.external_session_id, path=path, content=content
|
|
)
|
|
|
|
# -- snapshots & previews ------------------------------------------------
|
|
|
|
async def snapshot(
|
|
self, *, organization_id: str, session_id: str, name: str
|
|
) -> Snapshot:
|
|
session = await self.get_session_or_404(
|
|
organization_id=organization_id, session_id=session_id
|
|
)
|
|
if not session.external_session_id:
|
|
raise ProviderUnavailableError("نشست در ارائهدهنده ثبت نشده است.")
|
|
info = await self.provider.snapshot(
|
|
session_external_id=session.external_session_id, name=name
|
|
)
|
|
return await self.snapshots.create(
|
|
Snapshot(
|
|
workspace_id=session.workspace_id, name=name,
|
|
external_snapshot_id=info.external_id,
|
|
)
|
|
)
|
|
|
|
async def get_preview_url(
|
|
self, *, organization_id: str, session_id: str, port: int
|
|
) -> PreviewUrl:
|
|
session = await self.get_session_or_404(
|
|
organization_id=organization_id, session_id=session_id
|
|
)
|
|
if not session.external_session_id:
|
|
raise ProviderUnavailableError("نشست در ارائهدهنده ثبت نشده است.")
|
|
url = await self.provider.get_preview_url(
|
|
session_external_id=session.external_session_id, port=port
|
|
)
|
|
return await self.previews.create(
|
|
PreviewUrl(workspace_id=session.workspace_id, port=port, url=url)
|
|
)
|
|
|
|
# -- policy administration ----------------------------------------------
|
|
|
|
async def set_limits(
|
|
self,
|
|
*,
|
|
scope: str,
|
|
scope_id: str | None,
|
|
cpu: float,
|
|
memory_mb: int,
|
|
disk_mb: int,
|
|
timeout_seconds: int,
|
|
max_output_bytes: int,
|
|
network_policy: NetworkPolicy,
|
|
) -> SandboxResourceLimit:
|
|
if scope not in ("organization", "project", "platform"):
|
|
raise ValidationError("دامنهٔ محدودیت نامعتبر است.")
|
|
return await self.limits.upsert(
|
|
self._clamp(
|
|
SandboxResourceLimit(
|
|
scope=scope, scope_id=scope_id, cpu=cpu, memory_mb=memory_mb,
|
|
disk_mb=disk_mb, timeout_seconds=timeout_seconds,
|
|
max_output_bytes=max_output_bytes, network_policy=network_policy,
|
|
)
|
|
)
|
|
)
|
|
|
|
async def destroy_workspace(
|
|
self, *, organization_id: str, workspace_id: str
|
|
) -> Workspace:
|
|
workspace = await self.get_workspace_or_404(
|
|
organization_id=organization_id, workspace_id=workspace_id
|
|
)
|
|
for session in await self.sessions.list_for_workspace(workspace.id):
|
|
if session.external_session_id:
|
|
# Teardown is best-effort: a remote sandbox that is already gone
|
|
# must not block us from marking the local row stopped.
|
|
with contextlib.suppress(Exception):
|
|
await self.provider.destroy(
|
|
session_external_id=session.external_session_id
|
|
)
|
|
await self.sessions.set_status(
|
|
session.id, SessionStatus.STOPPED, stopped=True
|
|
)
|
|
await self.workspaces.set_status(workspace.id, WorkspaceStatus.DESTROYED)
|
|
refreshed = await self.workspaces.get_for_org(organization_id, workspace_id)
|
|
return refreshed or workspace
|
|
|
|
@staticmethod
|
|
def now():
|
|
return utcnow()
|
|
|
|
# -- platform maintenance ------------------------------------------------
|
|
|
|
async def reap_idle_sessions(
|
|
self, *, idle_before: datetime, limit: int = 200
|
|
) -> list[SandboxSession]:
|
|
"""Stop sandbox sessions untouched since ``idle_before``.
|
|
|
|
**Deliberately not tenant-scoped.** This is the gateway's own housekeeping
|
|
and spans every organization, so it is never reachable from a
|
|
tenant-scoped route — only from the sandbox-gateway worker. Callers are
|
|
responsible for the audit trail.
|
|
|
|
A provider that has already forgotten the session must not keep it
|
|
"active" in our own record forever, so the remote teardown is
|
|
best-effort and the local row is always marked stopped.
|
|
"""
|
|
stopped: list[SandboxSession] = []
|
|
stale = await self.sessions.list_stale_active(
|
|
idle_before=idle_before, limit=limit
|
|
)
|
|
for session in stale:
|
|
if session.external_session_id:
|
|
with contextlib.suppress(Exception):
|
|
await self.provider.destroy(
|
|
session_external_id=session.external_session_id
|
|
)
|
|
await self.sessions.set_status(session.id, SessionStatus.STOPPED, stopped=True)
|
|
refreshed = await self.sessions.get(session.id)
|
|
stopped.append(refreshed or session)
|
|
return stopped
|