mirror of
https://github.com/MemeMeow-Studio/MemeMeow.git
synced 2026-10-18 21:52:59 +08:00
infstellar
3df9f7878c
Add authenticated in-container execution, shared runtime activity summaries, strict result-file validation, and OpenCode integration tests. 引入带认证的容器内执行器、共享 runtime 活动摘要、严格结果文件校验和 OpenCode 集成测试。
384 lines
17 KiB
Python
384 lines
17 KiB
Python
"""共享 Agent 容器边界和任务结果文件协议测试。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shlex
|
|
import shutil
|
|
import subprocess
|
|
import uuid
|
|
import asyncio
|
|
import stat
|
|
import time
|
|
from types import SimpleNamespace
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from starlette.requests import Request
|
|
|
|
from api import config_status
|
|
from backend.config import Settings
|
|
from backend.opencode import OpenCodeError, OpenCodeRunner
|
|
|
|
|
|
def make_settings(tmp_path: Path, **overrides: object) -> Settings:
|
|
"""构造仅使用临时目录的 Docker runtime 配置。"""
|
|
values: dict[str, object] = {
|
|
"data_root": tmp_path / "data",
|
|
"image_root": tmp_path / "data" / "images",
|
|
"opencode_model": "mememeow/gpt-5.6-luna",
|
|
"opencode_base_url": "https://example.invalid/v1",
|
|
"opencode_api_key": "test-key",
|
|
"agent_container_name": "mememeow-agent-runtime",
|
|
"agent_runtime_mode": "docker",
|
|
}
|
|
values.update(overrides)
|
|
return Settings(**values)
|
|
|
|
|
|
def candidate() -> dict[str, object]:
|
|
"""返回满足输出 schema 的最小业务对象。"""
|
|
return {
|
|
"title": "测试标题",
|
|
"summary": "测试摘要",
|
|
"subjects": ["主体"],
|
|
"visible_text": [],
|
|
"references": [],
|
|
"meaning": None,
|
|
"keywords": ["测试"],
|
|
"search_queries": [],
|
|
"uncertainties": [],
|
|
"source_urls": [],
|
|
}
|
|
|
|
|
|
def test_container_command_whitelists_environment_and_maps_image(tmp_path: Path):
|
|
"""容器命令使用 /images 路径且不继承宿主密钥或数据库环境。"""
|
|
image_root = tmp_path / "data" / "images"
|
|
image_root.mkdir(parents=True)
|
|
image = image_root / "nested" / "meme.png"
|
|
image.parent.mkdir()
|
|
image.write_bytes(b"image")
|
|
runner = OpenCodeRunner(make_settings(tmp_path), project_root=tmp_path)
|
|
command = runner.container_command(image, "write result", task_id="task-1")
|
|
assert command[:3] == ["docker", "exec", "--workdir"]
|
|
assert "/images/nested/meme.png" in command
|
|
assert "--auto" in command
|
|
assert "--env" in command
|
|
assert "test-key" not in " ".join(command)
|
|
assert all("=" not in value for index, value in enumerate(command) if index and command[index - 1] == "--env")
|
|
assert not any("DATABASE_URL" in value or "HOME=" in value for value in command)
|
|
assert "SERPAPI_API_KEY" not in command
|
|
|
|
|
|
def test_docker_environment_contains_claim_task_id(tmp_path: Path):
|
|
"""Docker 白名单环境必须和 Host 一样传递当前 claim task id。"""
|
|
runner = OpenCodeRunner(make_settings(tmp_path), project_root=tmp_path)
|
|
environment = runner._allowed_container_environment(1, "claim-task-123")
|
|
assert environment["MEMEMEOW_AGENT_TASK_ID"] == "claim-task-123"
|
|
assert "SERPAPI_API_KEY" not in environment
|
|
|
|
|
|
def test_docker_environment_uses_agent_callback_urls(tmp_path: Path):
|
|
"""宿主后端与 Docker Agent 分离时，容器应使用专用 host-gateway 回调地址。"""
|
|
runner = OpenCodeRunner(
|
|
make_settings(
|
|
tmp_path,
|
|
agent_reverse_image_internal_url="http://host.docker.internal:8275/internal/reverse-image/search",
|
|
agent_visual_search_internal_url="http://host.docker.internal:8275/internal/visual-search/match",
|
|
),
|
|
project_root=tmp_path,
|
|
)
|
|
environment = runner._allowed_container_environment(0, "claim-task-urls")
|
|
assert environment["MEMEMEOW_REVERSE_IMAGE_INTERNAL_URL"].startswith("http://host.docker.internal:")
|
|
assert environment["MEMEMEOW_VISUAL_SEARCH_INTERNAL_URL"].startswith("http://host.docker.internal:")
|
|
|
|
|
|
def test_image_path_outside_root_is_rejected(tmp_path: Path):
|
|
"""图片根目录外的路径不能映射到 Agent。"""
|
|
outside = tmp_path / "outside.png"
|
|
outside.write_bytes(b"x")
|
|
runner = OpenCodeRunner(make_settings(tmp_path), project_root=tmp_path)
|
|
with pytest.raises(OpenCodeError) as error:
|
|
runner.map_host_image_path(outside)
|
|
assert error.value.code == "agent_image_path_forbidden"
|
|
|
|
|
|
def test_image_symlink_is_rejected_before_resolution(tmp_path: Path):
|
|
"""图片根目录内的符号链接也不能借路径解析绕过只读边界。"""
|
|
image_root = tmp_path / "data" / "images"
|
|
image_root.mkdir(parents=True)
|
|
source = image_root / "source.png"
|
|
source.write_bytes(b"image")
|
|
link = image_root / "link.png"
|
|
link.symlink_to(source)
|
|
runner = OpenCodeRunner(make_settings(tmp_path), project_root=tmp_path)
|
|
with pytest.raises(OpenCodeError) as error:
|
|
runner.map_host_image_path(link)
|
|
assert error.value.code == "agent_image_path_forbidden"
|
|
|
|
|
|
def test_docker_image_root_mismatch_is_rejected_before_container_exec(tmp_path: Path):
|
|
"""Docker 固定 `/images` 挂载与自定义图片根不一致时稳定阻断任务。"""
|
|
custom_root = tmp_path / "custom-images"
|
|
custom_root.mkdir()
|
|
runner = OpenCodeRunner(make_settings(tmp_path, image_root=custom_root))
|
|
with pytest.raises(OpenCodeError) as error:
|
|
runner.prepare_runtime()
|
|
assert error.value.code == "agent_image_root_mismatch"
|
|
|
|
|
|
def test_runtime_probe_strictly_checks_non_root_and_dependencies(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
|
"""运行时探针必须把 UID 0 和可写依赖目录判为不安全。"""
|
|
runner = OpenCodeRunner(make_settings(tmp_path))
|
|
monkeypatch.setattr(runner, "_container_ready", lambda: (True, "ok"))
|
|
monkeypatch.setattr(runner, "_container_exec_probe", lambda arguments: (False, "uid=0") if arguments[:2] == ["sh", "-lc"] and "id -u" in arguments[-1] else (True, "ok"))
|
|
result = runner.runtime_probe()
|
|
assert result["non_root"] is False
|
|
assert result["verified"] is False
|
|
|
|
|
|
def test_config_exposes_sanitized_agent_runtime_probe(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
|
"""`/config` 只返回固定 runtime 标识和布尔状态，不泄露诊断或密钥。"""
|
|
runner = OpenCodeRunner(make_settings(tmp_path))
|
|
monkeypatch.setattr(
|
|
runner,
|
|
"runtime_probe",
|
|
lambda: {
|
|
"mode": "shared-docker-container",
|
|
"container_name": "mememeow-agent-runtime",
|
|
"container_running": True,
|
|
"runtime_root_ready": True,
|
|
"workspace_ready": True,
|
|
"executable_ready": True,
|
|
"skills_ready": True,
|
|
"dependencies_ready": True,
|
|
"mounts_ready": True,
|
|
"non_root": True,
|
|
"network_ready": True,
|
|
"docker_socket_absent": True,
|
|
"verified": True,
|
|
"container_diagnostic": "宿主绝对路径不应返回",
|
|
},
|
|
)
|
|
app_stub = SimpleNamespace(
|
|
state=SimpleNamespace(
|
|
settings=runner.settings,
|
|
opencode=runner,
|
|
search_engine=SimpleNamespace(has_cache=lambda: False),
|
|
)
|
|
)
|
|
request = Request({
|
|
"type": "http",
|
|
"method": "GET",
|
|
"path": "/config",
|
|
"raw_path": b"/config",
|
|
"query_string": b"",
|
|
"headers": [],
|
|
"app": app_stub,
|
|
})
|
|
payload = asyncio.run(config_status(request))
|
|
assert payload["runtime_ready"] is True
|
|
assert payload["agent_runtime"]["mode"] == "shared-docker-container"
|
|
assert "container_diagnostic" not in payload["agent_runtime"]
|
|
assert "opencode_api_key" not in payload
|
|
|
|
|
|
def test_result_path_failure_releases_slot(tmp_path: Path):
|
|
"""任务结果路径冲突时不能泄漏已获取的 slot。"""
|
|
runner = OpenCodeRunner(make_settings(tmp_path))
|
|
runner.prepare_runtime = lambda: None
|
|
task_directory = runner.runtime_root / "task-results" / "slot-conflict"
|
|
task_directory.mkdir(parents=True)
|
|
(task_directory / "result.json.tmp").mkdir()
|
|
with pytest.raises(OpenCodeError) as error:
|
|
runner.run(tmp_path / "image.png", lambda *_: None, task_id="slot-conflict")
|
|
assert error.value.code == "agent_result_path_invalid"
|
|
assert runner._slot_semaphore.acquire(timeout=0.1)
|
|
runner._slot_semaphore.release()
|
|
|
|
|
|
def _run_container(runner: OpenCodeRunner, *arguments: str) -> subprocess.CompletedProcess[str]:
|
|
"""通过与生产运行器相同的 Docker/sg 包装执行容器命令。"""
|
|
command = runner._docker_command_for_execution(runner._container_exec(*arguments))
|
|
return subprocess.run(command, capture_output=True, text=True, timeout=20, check=False)
|
|
|
|
|
|
@pytest.mark.skipif(os.getenv("MEMEMEOW_AGENT_RUNTIME_E2E") != "1", reason="显式设置 MEMEMEOW_AGENT_RUNTIME_E2E=1 才运行 Docker 集成验收")
|
|
def test_shared_container_runs_distinct_sessions_and_result_directories():
|
|
"""两个真实 docker exec 任务共享容器，但使用不同 session 和结果目录。"""
|
|
project_root = Path(__file__).resolve().parent.parent
|
|
settings = Settings.from_env(project_root / ".env")
|
|
settings.agent_runtime_mode = "docker"
|
|
settings.agent_container_name = os.getenv("MEMEMEOW_AGENT_CONTAINER_NAME", "mememeow-agent-runtime")
|
|
runner = OpenCodeRunner(settings, project_root=project_root)
|
|
if not runner.runtime_probe().get("verified"):
|
|
pytest.skip("共享 Agent 容器未运行或探针未通过")
|
|
runner.prepare_runtime()
|
|
image = settings.image_root / "example_meme_1.jpg"
|
|
if not image.is_file():
|
|
pytest.skip("缺少 data/images/example_meme_1.jpg 集成图片")
|
|
|
|
suffix = uuid.uuid4().hex
|
|
task_ids = (f"integration-{suffix}-a", f"integration-{suffix}-b")
|
|
stub_path = f"/runtime/workspace/.mememeow-agent-stub-{suffix}"
|
|
stub = """#!/bin/sh
|
|
set -eu
|
|
task="${MEMEMEOW_AGENT_TASK_ID:?}"
|
|
directory="/runtime/task-results/$task"
|
|
mkdir -p "$directory"
|
|
printf '%s\\n' '{"title":"集成验收","summary":"共享容器结果","subjects":["主体"],"visible_text":[],"references":[],"meaning":null,"keywords":["集成"],"search_queries":[],"uncertainties":[],"source_urls":[]}' > "$directory/result.json.draft"
|
|
mv "$directory/result.json.draft" "$directory/result.json.tmp"
|
|
printf '{"type":"session.created","session_id":"stub-%s"}\\n' "$task"
|
|
"""
|
|
escaped_stub = shlex.quote(stub)
|
|
setup = _run_container(runner, "sh", "-lc", f"printf %s {escaped_stub} > {shlex.quote(stub_path)} && chmod 755 {shlex.quote(stub_path)}")
|
|
assert setup.returncode == 0, setup.stderr
|
|
original_executable = settings.opencode_executable
|
|
settings.opencode_executable = stub_path
|
|
try:
|
|
first, first_session = runner.run(image, lambda *_: None, task_id=task_ids[0])
|
|
second, second_session = runner.run(image, lambda *_: None, task_id=task_ids[1])
|
|
assert first["title"] == second["title"] == "集成验收"
|
|
assert first_session != second_session
|
|
assert (runner.runtime_root / "task-results" / task_ids[0] / "result.json.tmp").is_file()
|
|
assert (runner.runtime_root / "task-results" / task_ids[1] / "result.json.tmp").is_file()
|
|
boundary = _run_container(
|
|
runner,
|
|
"sh",
|
|
"-lc",
|
|
"test \"$(id -u)\" != 0 && test -r /images/example_meme_1.jpg && test ! -w /images && "
|
|
"test -r /skills/research-meme-context && test ! -w /skills/research-meme-context && "
|
|
"test -r /opt/mememeow/node_modules && test ! -w /opt/mememeow/node_modules && "
|
|
"test -r /runtime && test -w /runtime && test ! -e /.env && test ! -S /var/run/docker.sock",
|
|
)
|
|
assert boundary.returncode == 0, boundary.stderr
|
|
finally:
|
|
settings.opencode_executable = original_executable
|
|
_run_container(runner, "sh", "-lc", f"rm -f {shlex.quote(stub_path)}")
|
|
for task_id in task_ids:
|
|
shutil.rmtree(runner.runtime_root / "task-results" / task_id, ignore_errors=True)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("kind", "expected"),
|
|
[("missing", "agent_result_file_missing"), ("invalid", "agent_result_file_invalid_json"), ("schema", "agent_result_file_schema_invalid"), ("large", "agent_result_file_too_large")],
|
|
)
|
|
def test_result_file_failures_have_stable_codes(tmp_path: Path, kind: str, expected: str):
|
|
"""结果文件缺失、JSON 截断、schema 错误和超限均不回退 assistant 文本。"""
|
|
runner = OpenCodeRunner(make_settings(tmp_path, agent_result_max_bytes=1024))
|
|
_draft, result_path = runner.create_task_result_paths("task-1")
|
|
if kind == "invalid":
|
|
result_path.write_text('{"title":', encoding="utf-8")
|
|
elif kind == "schema":
|
|
result_path.write_text(json.dumps({"title": "only-title"}), encoding="utf-8")
|
|
elif kind == "large":
|
|
result_path.write_bytes(b"x" * 1025)
|
|
with pytest.raises(OpenCodeError) as error:
|
|
runner.read_result_file(result_path)
|
|
assert error.value.code == expected
|
|
|
|
|
|
def test_result_file_success_is_schema_validated_and_task_directories_are_independent(tmp_path: Path):
|
|
"""两个任务使用不同结果目录，成功结果只来自最终临时文件。"""
|
|
runner = OpenCodeRunner(make_settings(tmp_path))
|
|
first_draft, first_result = runner.create_task_result_paths("task-a")
|
|
second_draft, second_result = runner.create_task_result_paths("task-b")
|
|
assert first_result.parent != second_result.parent
|
|
first_draft.write_text(json.dumps(candidate(), ensure_ascii=False), encoding="utf-8")
|
|
first_draft.replace(first_result)
|
|
second_result.write_text(json.dumps(candidate(), ensure_ascii=False), encoding="utf-8")
|
|
assert runner.read_result_file(first_result)["title"] == "测试标题"
|
|
assert runner.read_result_file(second_result)["title"] == "测试标题"
|
|
|
|
|
|
def test_retry_clears_previous_result_artifact(tmp_path: Path):
|
|
"""同一任务重试前必须移除旧最终文件，防止失败尝试误读历史结果。"""
|
|
runner = OpenCodeRunner(make_settings(tmp_path))
|
|
draft, result = runner.create_task_result_paths("retry-task")
|
|
result.write_text(json.dumps(candidate(), ensure_ascii=False), encoding="utf-8")
|
|
runner._reset_task_result_files(draft, result)
|
|
assert not draft.exists()
|
|
assert not result.exists()
|
|
|
|
|
|
@pytest.mark.parametrize("kind", ["result", "draft", "directory"])
|
|
def test_result_paths_and_reads_reject_symlink_hijacking(tmp_path: Path, kind: str):
|
|
"""结果最终文件、草稿和任务目录均不能通过符号链接逃逸。"""
|
|
runner = OpenCodeRunner(make_settings(tmp_path))
|
|
root = runner.runtime_root / "task-results"
|
|
root.mkdir(parents=True)
|
|
if kind == "directory":
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
(root / "task-link").symlink_to(outside, target_is_directory=True)
|
|
with pytest.raises(OpenCodeError) as error:
|
|
runner.create_task_result_paths("task-link")
|
|
else:
|
|
draft, result = runner.create_task_result_paths("task-link")
|
|
target = draft if kind == "draft" else result
|
|
target.unlink(missing_ok=True)
|
|
outside = tmp_path / f"outside-{kind}.json"
|
|
outside.write_text(json.dumps(candidate()), encoding="utf-8")
|
|
target.symlink_to(outside)
|
|
with pytest.raises(OpenCodeError) as error:
|
|
if kind == "draft":
|
|
runner._reset_task_result_files(draft, result)
|
|
else:
|
|
runner.read_result_file(result)
|
|
assert error.value.code in {"agent_result_path_invalid", "agent_result_file_unreadable"}
|
|
|
|
|
|
def test_result_parent_symlink_is_rejected(tmp_path: Path):
|
|
"""task-results 本身被替换为符号链接时不能创建任务目录。"""
|
|
runner = OpenCodeRunner(make_settings(tmp_path))
|
|
runner.runtime_root.mkdir(parents=True)
|
|
outside = tmp_path / "outside-results"
|
|
outside.mkdir()
|
|
(runner.runtime_root / "task-results").symlink_to(outside, target_is_directory=True)
|
|
with pytest.raises(OpenCodeError) as error:
|
|
runner.create_task_result_paths("task-1")
|
|
assert error.value.code == "agent_result_path_invalid"
|
|
|
|
|
|
def test_cleanup_preserves_active_task_directories(tmp_path: Path):
|
|
"""清理旧产物时必须跳过其他仍在运行 task 的结果目录。"""
|
|
runner = OpenCodeRunner(make_settings(tmp_path, agent_result_retention_days=1, agent_result_max_tasks=1))
|
|
active = runner.runtime_root / "task-results" / "active-task"
|
|
old = runner.runtime_root / "task-results" / "old-task"
|
|
active.mkdir(parents=True)
|
|
old.mkdir(parents=True)
|
|
old_mtime = time.time() - 3 * 86400
|
|
os.utime(active, (old_mtime, old_mtime))
|
|
os.utime(old, (old_mtime, old_mtime))
|
|
runner._mark_task_active("active-task")
|
|
try:
|
|
assert runner.cleanup_task_results() == 1
|
|
assert active.is_dir()
|
|
assert not old.exists()
|
|
finally:
|
|
runner._unmark_task_active("active-task")
|
|
|
|
|
|
def test_terminate_container_session_waits_and_escalates(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
|
"""容器任务终止应发送 TERM、等待确认，再发送 KILL 兜底。"""
|
|
runner = OpenCodeRunner(make_settings(tmp_path))
|
|
calls: list[str] = []
|
|
alive = {"value": True}
|
|
ignore_term = {"value": False}
|
|
def signal_processes(pattern: str, signal_name: str) -> None:
|
|
calls.append(signal_name)
|
|
if signal_name == "TERM" and not ignore_term["value"]:
|
|
alive["value"] = False
|
|
monkeypatch.setattr(runner, "_container_signal_processes", signal_processes)
|
|
monkeypatch.setattr(runner, "_container_has_task_processes", lambda pattern: alive["value"])
|
|
runner._terminate_container_session("task-1")
|
|
assert calls == ["TERM"]
|
|
|
|
calls.clear()
|
|
alive["value"] = True
|
|
ignore_term["value"] = True
|
|
runner._terminate_container_session("task-2")
|
|
assert calls[:2] == ["TERM", "KILL"]
|