diff --git a/backend/alembic/versions/4e679f11da46_插件沙箱执行记录.py b/backend/alembic/versions/4e679f11da46_插件沙箱执行记录.py new file mode 100644 index 0000000..ac1076a --- /dev/null +++ b/backend/alembic/versions/4e679f11da46_插件沙箱执行记录.py @@ -0,0 +1,47 @@ +"""插件沙箱执行记录 + +Revision ID: 4e679f11da46 +Revises: 89a4ed9647ee +Create Date: 2026-09-12 13:33:43.279920 + +""" +from collections.abc import Sequence + +import sqlalchemy as sa + +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = '4e679f11da46' +down_revision: str | Sequence[str] | None = '89a4ed9647ee' +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """Upgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.create_table('plugin_execution', + sa.Column('id', sa.String(), nullable=False), + sa.Column('installation_id', sa.String(), nullable=False), + sa.Column('status', sa.String(), nullable=False), + sa.Column('exit_code', sa.Integer(), nullable=True), + sa.Column('duration_ms', sa.Integer(), nullable=True), + sa.Column('stdout_excerpt', sa.String(), nullable=True), + sa.Column('error', sa.String(), nullable=True), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.Column('completed_at', sa.DateTime(timezone=True), nullable=True), + sa.CheckConstraint("status IN ('running', 'succeeded', 'failed', 'timeout', 'killed')", name=op.f('ck_plugin_execution_status_values')), + sa.ForeignKeyConstraint(['installation_id'], ['plugin_installation.id'], name=op.f('fk_plugin_execution_plugin_installation'), ondelete='cascade'), + sa.PrimaryKeyConstraint('id', name=op.f('pk_plugin_execution')) + ) + op.create_index(op.f('ix_plugin_execution_installation_id'), 'plugin_execution', ['installation_id'], unique=False) + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.drop_index(op.f('ix_plugin_execution_installation_id'), table_name='plugin_execution') + op.drop_table('plugin_execution') + # ### end Alembic commands ### diff --git a/backend/app/core/config.py b/backend/app/core/config.py index a21977e..3cb4dfa 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -13,6 +13,7 @@ class Settings(BaseSettings): api_port: int = 8000 database_url: str = "postgresql+asyncpg://hunjing:hunjing_dev_password@127.0.0.1:5432/hunjing" storage_root: str = "storage" + sandbox_enabled: bool = False # 未启用时隔离插件执行 fail-closed @lru_cache diff --git a/backend/app/models.py b/backend/app/models.py index 40d5b0c..1e3de03 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -1304,3 +1304,32 @@ class DiagnosticGrant(Base): created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) + + +class PluginExecution(Base): + """插件沙箱执行记录(WP-06/07 Sandbox):结果与资源消耗如实留痕。""" + + __tablename__ = "plugin_execution" + + id: Mapped[str] = mapped_column( + String, primary_key=True, default=lambda: new_id("execution_") + ) + installation_id: Mapped[str] = mapped_column( + ForeignKey("plugin_installation.id", ondelete="cascade"), nullable=False, index=True + ) + status: Mapped[str] = mapped_column(String, nullable=False, default="running") + exit_code: Mapped[int | None] = mapped_column(nullable=True) + duration_ms: Mapped[int | None] = mapped_column(nullable=True) + stdout_excerpt: Mapped[str | None] = mapped_column(String, nullable=True) + error: Mapped[str | None] = mapped_column(String, nullable=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + + __table_args__ = ( + CheckConstraint( + "status IN ('running', 'succeeded', 'failed', 'timeout', 'killed')", + name="status_values", + ), + ) diff --git a/backend/app/plugins/routes.py b/backend/app/plugins/routes.py index ac73f1d..f5a4d8c 100644 --- a/backend/app/plugins/routes.py +++ b/backend/app/plugins/routes.py @@ -144,3 +144,27 @@ async def invoke( db, domain.id, name, body.get("input", {}), body.get("run_id"), getattr(request.state, "trace_id", None), ) + + +@inst_router.post("/{installation_id}/executions", status_code=201) +async def execute_plugin( + installation_id: str, + db: DbSession, + user: Annotated[User, Depends(require_state_change)], + request: Request, + body: dict[str, Any], +) -> dict[str, Any]: + """沙箱执行 isolated 插件:未启用沙箱一律 503(fail-closed)。""" + domain = await _domain_of(db, user) + from app.core.config import get_settings + + execution, result = await service.execute_isolated( + db, domain.id, installation_id, body.get("input", {}), + sandbox_enabled=get_settings().sandbox_enabled, + trace_id=getattr(request.state, "trace_id", None), + ) + return { + "execution_id": execution.id, "status": result["status"], + "duration_ms": result.get("duration_ms"), + "output": result.get("output"), "error": result.get("error"), + } diff --git a/backend/app/plugins/service.py b/backend/app/plugins/service.py index 1ed3495..4d1c4f8 100644 --- a/backend/app/plugins/service.py +++ b/backend/app/plugins/service.py @@ -297,3 +297,70 @@ async def invoke( ) await db.flush() return output + + +async def execute_isolated( + db: AsyncSession, + domain_id: str, + installation_id: str, + input_: dict[str, Any], + sandbox_enabled: bool, + trace_id: str | None, +) -> tuple[Any, dict[str, Any]]: + """隔离执行:isolated 形态插件经沙箱子进程运行;未启用沙箱一律拒绝。 + + 返回(执行记录, 结果)。代码体取自 Manifest 的 code 字段 + (部署者明确信任的本地代码插件,仍受资源限制约束)。 + """ + from app.models import PluginExecution + from app.sandbox.executor import run_sandboxed + + installation = await require_installation(db, domain_id, installation_id) + plugin = await db.get(Plugin, installation.plugin_name) + assert plugin is not None + if plugin.kind != "isolated_process": + raise HTTPException( + status.HTTP_422_UNPROCESSABLE_ENTITY, "仅 isolated_process 形态走沙箱执行" + ) + if not installation.enabled: + raise HTTPException(status.HTTP_403_FORBIDDEN, "插件未启用") + if not sandbox_enabled: + db.add( + AuditEvent( + actor_type="user", actor_id=domain_id, action="plugin.execution_denied", + aggregate_type="plugin", aggregate_id=plugin.name, + payload={"reason": "sandbox_not_enabled"}, trace_id=trace_id, + ) + ) + await db.flush() + raise HTTPException( + status.HTTP_503_SERVICE_UNAVAILABLE, + "沙箱未启用:拒绝执行(fail-closed),启用请设置 HUNJING_SANDBOX_ENABLED", + ) + pv = await db.get(PluginVersion, installation.version_id) + assert pv is not None + code = str(pv.manifest.get("code", "")) + execution = PluginExecution(installation_id=installation.id, status="running") + db.add(execution) + await db.flush() + result = await run_sandboxed(code, input_) + execution.status = result["status"] + execution.exit_code = result.get("exit_code") + execution.duration_ms = result.get("duration_ms") + execution.stdout_excerpt = result.get("stdout_excerpt") + execution.error = result.get("error") + from datetime import UTC + from datetime import datetime as _dt + + execution.completed_at = _dt.now(UTC) + db.add( + AuditEvent( + actor_type="user", actor_id=domain_id, action="plugin.executed", + aggregate_type="plugin", aggregate_id=plugin.name, + payload={"execution_id": execution.id, "status": execution.status, + "duration_ms": execution.duration_ms}, + trace_id=trace_id, + ) + ) + await db.flush() + return execution, result diff --git a/backend/app/sandbox/__init__.py b/backend/app/sandbox/__init__.py new file mode 100644 index 0000000..0444c1d --- /dev/null +++ b/backend/app/sandbox/__init__.py @@ -0,0 +1 @@ +"""沙箱域:隔离执行器与策略。""" diff --git a/backend/app/sandbox/executor.py b/backend/app/sandbox/executor.py new file mode 100644 index 0000000..8218b67 --- /dev/null +++ b/backend/app/sandbox/executor.py @@ -0,0 +1,109 @@ +"""沙箱执行器(WP-06/07 Sandbox):子进程隔离 + 资源限制 + 超时。 + +隔离边界(按部署档): +- 本执行器提供:独立进程、CPU/内存 rlimit、墙钟超时与强杀、输出限额。 +- 进程级隔离不是完整安全边界;网络隔离与文件系统最小化在 Linux 容器档 + (container 形态)交付。部署者必须显式启用沙箱(HUNJING_SANDBOX_ENABLED), + 未启用时一律拒绝执行(fail-closed,AGENTS 第 6 节)。 +""" + +import asyncio +import json +import resource +import sys +import time +from dataclasses import dataclass +from typing import Any + +# 用户代码经 stdin JSON 传输,不进命令行(避免注入命令行参数) +_RUNNER = r''' +import json, sys +def _main(): + payload = json.loads(sys.stdin.read()) + scope = {"input": payload.get("input", {})} + exec(compile(payload["code"], "", "exec"), scope) + result = scope.get("output") + sys.stdout.write(json.dumps({"output": result}, ensure_ascii=False)) +_main() +''' + + +@dataclass(frozen=True) +class SandboxPolicy: + max_runtime_ms: int = 5_000 + max_memory_mb: int = 128 + max_output_chars: int = 10_000 + + +class SandboxUnavailable(Exception): + """未启用沙箱:调用方必须显式失败,不得退化为宿主执行。""" + + +def _apply_limits(policy: SandboxPolicy) -> None: # 在子进程内执行(preexec_fn) + # RLIMIT_AS 在部分平台(macOS)不可设:尽力而为,失败不阻断进程隔离 + cpu_seconds = max(1, policy.max_runtime_ms // 1000 + 1) + resource.setrlimit(resource.RLIMIT_CPU, (cpu_seconds, cpu_seconds)) + if hasattr(resource, "RLIMIT_AS"): + as_bytes = policy.max_memory_mb * 1024 * 1024 + try: + resource.setrlimit(resource.RLIMIT_AS, (as_bytes, as_bytes)) + except (ValueError, OSError): + pass + + +async def run_sandboxed( + code: str, input_: dict[str, Any], policy: SandboxPolicy | None = None +) -> dict[str, Any]: + """在受限子进程中执行插件代码;返回执行结果(不抛业务异常以外的错误)。""" + pol = policy or SandboxPolicy() + stdin_payload = json.dumps({"code": code, "input": input_}, ensure_ascii=False) + started = time.monotonic() + try: + proc = await asyncio.create_subprocess_exec( + sys.executable, "-c", _RUNNER, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + preexec_fn=lambda: _apply_limits(pol), + ) + except Exception as exc: # 进程无法启动:如实失败 + return { + "status": "failed", "exit_code": None, + "duration_ms": int((time.monotonic() - started) * 1000), + "output": None, "error": f"sandbox_spawn_failed: {exc}", + } + try: + stdout, stderr = await asyncio.wait_for( + proc.communicate(stdin_payload.encode("utf-8")), + timeout=pol.max_runtime_ms / 1000, + ) + except TimeoutError: + proc.kill() + await proc.wait() + return { + "status": "timeout", "exit_code": None, + "duration_ms": int((time.monotonic() - started) * 1000), + "output": None, "error": f"exceeded {pol.max_runtime_ms}ms wall clock", + } + duration_ms = int((time.monotonic() - started) * 1000) + out_text = stdout.decode("utf-8", "replace")[: pol.max_output_chars] + err_text = stderr.decode("utf-8", "replace")[: pol.max_output_chars] + if proc.returncode == 0: + try: + parsed = json.loads(out_text) + except json.JSONDecodeError: + return { + "status": "failed", "exit_code": proc.returncode, + "duration_ms": duration_ms, "output": None, + "error": "plugin output is not JSON", + } + return { + "status": "succeeded", "exit_code": 0, + "duration_ms": duration_ms, "output": parsed.get("output"), + "stdout_excerpt": out_text, "error": None, + } + return { + "status": "failed", "exit_code": proc.returncode, + "duration_ms": duration_ms, "output": None, + "stdout_excerpt": out_text, "error": err_text or f"exit {proc.returncode}", + } diff --git a/backend/tests/test_conversations.py b/backend/tests/test_conversations.py index 14a9b28..d924f57 100644 --- a/backend/tests/test_conversations.py +++ b/backend/tests/test_conversations.py @@ -1805,3 +1805,114 @@ class TestOps: assert "hunjing_runs_total" in metrics.text ready = await client.get("/v1/readyz") assert ready.json()["status"] == "ready" + + +@pytest.mark.usefixtures("client") +class TestSandbox: + ISOLATED_MANIFEST = { + "permissions": {"data": ["self_input"]}, + "code": 'result = {"echo": input.get("text", "")}\noutput = result', + } + + async def _setup_isolated(self, client, publisher, user) -> str: + pub = await client.post( + "/v1/plugins/versions", + json={"name": "example.worker", "kind": "isolated_process", + "version": "1.0.0", "manifest": self.ISOLATED_MANIFEST}, + headers=publisher, + ) + assert pub.status_code == 201, pub.text + installation = await client.post( + "/v1/plugins/example.worker/installations", + json={"version": "1.0.0"}, headers=user, + ) + iid = installation.json()["id"] + enabled = await client.post( + f"/v1/plugin-installations/{iid}/enablings", + json={"granted_permissions": ["data:self_input"]}, headers=user, + ) + assert enabled.status_code == 201 + return iid + + async def test_execution_fails_closed_without_sandbox(self, client) -> None: + publisher = await login(client, "sb-pub@example.com") + user = await login(client, "sb-user@example.com") + iid = await self._setup_isolated(client, publisher, user) + resp = await client.post( + f"/v1/plugin-installations/{iid}/executions", + json={"input": {"text": "hi"}}, headers=user, + ) + assert resp.status_code == 503 # fail-closed:未启用沙箱拒绝执行 + assert "fail-closed" in resp.json()["detail"] + + async def test_sandboxed_execution_success_timeout_and_failure(self, client) -> None: + from app.core.config import get_settings + + get_settings.cache_clear() + import os + + os.environ["HUNJING_SANDBOX_ENABLED"] = "true" + get_settings.cache_clear() + try: + publisher = await login(client, "sb2-pub@example.com") + user = await login(client, "sb2-user@example.com") + iid = await self._setup_isolated(client, publisher, user) + # 成功执行:echo 回填 + ok = await client.post( + f"/v1/plugin-installations/{iid}/executions", + json={"input": {"text": "魂镜"}}, headers=user, + ) + assert ok.status_code == 201, ok.text + assert ok.json()["status"] == "succeeded", ok.text + assert ok.json()["output"]["echo"] == "魂镜" + assert ok.json()["duration_ms"] is not None + # 死循环:墙钟超时被强杀 + loop_pub = await client.post( + "/v1/plugins/versions", + json={"name": "example.loop", "kind": "isolated_process", + "version": "1.0.0", + "manifest": {"permissions": {}, "code": "while True: pass"}}, + headers=publisher, + ) + assert loop_pub.status_code == 201 + install_loop = await client.post( + "/v1/plugins/example.loop/installations", + json={"version": "1.0.0"}, headers=user, + ) + lid = install_loop.json()["id"] + await client.post( + f"/v1/plugin-installations/{lid}/enablings", + json={"granted_permissions": []}, headers=user, + ) + timeout = await client.post( + f"/v1/plugin-installations/{lid}/executions", + json={"input": {}}, headers=user, + ) + assert timeout.json()["status"] == "timeout" + assert "wall clock" in timeout.json()["error"] + # 抛异常:failed 且含 stderr 摘要 + await client.post( + "/v1/plugins/versions", + json={"name": "example.boom", "kind": "isolated_process", + "version": "1.0.0", + "manifest": {"permissions": {}, "code": "raise RuntimeError('炸了')"}}, + headers=publisher, + ) + install_boom = await client.post( + "/v1/plugins/example.boom/installations", + json={"version": "1.0.0"}, headers=user, + ) + bid = install_boom.json()["id"] + await client.post( + f"/v1/plugin-installations/{bid}/enablings", + json={"granted_permissions": []}, headers=user, + ) + failed = await client.post( + f"/v1/plugin-installations/{bid}/executions", + json={"input": {}}, headers=user, + ) + assert failed.json()["status"] == "failed" + assert "炸了" in failed.json()["error"] + finally: + os.environ.pop("HUNJING_SANDBOX_ENABLED", None) + get_settings.cache_clear()