""" Tests for the AgentPack feature (M2 Phase D). Covers: - lambdagent.agentpack: manifest validation (kernel, pure) - agentpaas.engine.agentpack_store: safe install / list / uninstall - zip-slip defense - SR-003 shell refusal for third-party packs Spec: docs/AGENTPACK_SPEC.md. """ from __future__ import annotations import os os.environ.setdefault("AGENTPAAS_DATABASE_URL", "sqlite:///:memory:") os.environ.setdefault("AGENTPAAS_TESTING", "1") import io import zipfile import pytest import yaml from lambdagent.agentpack import ( AgentPackManifest, PackPermissions, AgentPackError, load_manifest, validate_manifest_dict, enforce_config_permissions, ) from agentpaas.engine.agentpack_store import ( install_from_zip, list_installed, get_installed, uninstall, AgentPackInstallError, ) # ───────────────────────────────────────────────────────────────────────────── # Kernel: manifest validation # ───────────────────────────────────────────────────────────────────────────── _VALID = { "id": "research.top-journal-reviewer", "name": "Top Journal Reviewer", "version": "0.1.0", "domain": "research", "entrypoint": "agents/reviewer.yml", "permissions": {"network": False, "shell": False, "file_write": "workspace"}, "audience": ["professor", "phd_student"], "model": {"recommended": ["claude-code/sonnet"]}, } def test_manifest_valid(): m = validate_manifest_dict(_VALID) assert m.id == "research.top-journal-reviewer" assert m.version == "0.1.0" assert m.permissions.file_write == "workspace" assert m.permissions.shell is False assert "professor" in m.audience assert m.model_recommended == ["claude-code/sonnet"] @pytest.mark.parametrize("missing", ["id", "name", "version", "domain", "entrypoint", "permissions"]) def test_manifest_missing_required_field(missing): bad = dict(_VALID) bad.pop(missing) with pytest.raises(AgentPackError): validate_manifest_dict(bad) def test_manifest_bad_id_rejected(): bad = dict(_VALID, id="Bad ID With Spaces!") with pytest.raises(AgentPackError): validate_manifest_dict(bad) def test_manifest_bad_version_rejected(): bad = dict(_VALID, version="v1") with pytest.raises(AgentPackError): validate_manifest_dict(bad) def test_manifest_entrypoint_escape_rejected(): bad = dict(_VALID, entrypoint="../../etc/passwd") with pytest.raises(AgentPackError): validate_manifest_dict(bad) def test_manifest_bad_file_write_scope_rejected(): bad = dict(_VALID, permissions={"file_write": "everywhere"}) with pytest.raises(AgentPackError): validate_manifest_dict(bad) def test_permissions_deny_by_default(): # Empty permissions block → safe defaults. m = validate_manifest_dict(dict(_VALID, permissions={})) assert m.permissions.network is False assert m.permissions.shell is False assert m.permissions.file_write == "workspace" def test_permission_summary_zh(): m = validate_manifest_dict(_VALID) s = m.permissions.summary_zh() assert "读取本地知识库" in s assert "不执行 shell" in s assert "不访问网络" in s def test_permissions_are_enforced_on_runtime_config(): config = { "type": "react", "systemPrompt": "test", "mcp": { "localTools": [ "Bash", "RunTests", "WebSearch", "ReadFile", "WriteFile", "KBSearch", "KBAdd", "terminate", ], "onlineTool": {"remote": ["dangerous_remote_tool"]}, }, } effective, report = enforce_config_permissions(config, PackPermissions( network=False, shell=False, file_write="workspace", read_knowledge=True, read_filesystem=False, )) assert effective["mcp"]["localTools"] == ["WriteFile", "KBSearch", "terminate"] assert "onlineTool" not in effective["mcp"] assert set(report["removed_tools"]) == { "Bash", "RunTests", "WebSearch", "ReadFile", "KBAdd", } assert effective["_agentpack_permissions"]["file_write"] == "workspace" def test_gateway_wraps_validated_builtin_tools_case_insensitively(): from lambdagent.tool_gateway import GatedTool, GatewayPolicy, ToolGateway from lambdagent.validated_tool import ShellToolInput, ValidatedTool executed = [] bash = ValidatedTool("Bash", lambda value: executed.append(value) or "ran", ShellToolInput) gated = ToolGateway(GatewayPolicy()).wrap(bash) assert isinstance(gated, GatedTool) result = gated.apply('{"command":"rm -rf /"}') assert result.startswith("[BLOCKED]") assert executed == [] # ───────────────────────────────────────────────────────────────────────────── # Helpers to build pack zips # ───────────────────────────────────────────────────────────────────────────── def _make_pack_zip(tmp_path, manifest_dict, *, entrypoint_body="type: react\nname: x\nsystemPrompt: hi\n", nested=False, extra_members=None): """Build a .zip containing manifest.yml + the entrypoint config. If nested=True, wraps everything in a top-level dir.""" prefix = "mypack/" if nested else "" zpath = tmp_path / "pack.zip" with zipfile.ZipFile(zpath, "w") as zf: zf.writestr(prefix + "manifest.yml", yaml.safe_dump(manifest_dict, allow_unicode=True)) zf.writestr(prefix + manifest_dict["entrypoint"], entrypoint_body) for name, body in (extra_members or {}).items(): zf.writestr(prefix + name, body) return str(zpath) # ───────────────────────────────────────────────────────────────────────────── # Store: install / list / get / uninstall # ───────────────────────────────────────────────────────────────────────────── def test_install_list_get_uninstall_roundtrip(tmp_path): packs_dir = tmp_path / "agentpacks" packs_dir.mkdir() zp = _make_pack_zip(tmp_path, _VALID) pack = install_from_zip(zp, str(packs_dir)) assert pack.id == "research.top-journal-reviewer" assert pack.version == "0.1.0" assert os.path.isfile(os.path.join(pack.path, "manifest.yml")) assert os.path.isfile(os.path.join(pack.path, "agents", "reviewer.yml")) listed = list_installed(str(packs_dir)) assert len(listed) == 1 assert listed[0].id == pack.id got = get_installed(str(packs_dir), pack.id) assert got is not None and got.version == "0.1.0" n = uninstall(str(packs_dir), pack.id) assert n == 1 assert list_installed(str(packs_dir)) == [] def test_install_nested_pack_zip(tmp_path): """A zip with everything under a single top-level dir still installs.""" packs_dir = tmp_path / "agentpacks" packs_dir.mkdir() zp = _make_pack_zip(tmp_path, _VALID, nested=True) pack = install_from_zip(zp, str(packs_dir)) assert pack.id == "research.top-journal-reviewer" def test_install_rejects_missing_entrypoint(tmp_path): """Manifest references agents/reviewer.yml but the zip doesn't contain it.""" packs_dir = tmp_path / "agentpacks" packs_dir.mkdir() zpath = tmp_path / "pack.zip" with zipfile.ZipFile(zpath, "w") as zf: zf.writestr("manifest.yml", yaml.safe_dump(_VALID)) # entrypoint deliberately absent with pytest.raises((AgentPackInstallError, AgentPackError)): install_from_zip(str(zpath), str(packs_dir)) def test_install_refuses_third_party_shell(tmp_path): """SR-003: a pack declaring shell:true is refused unless allow_shell.""" packs_dir = tmp_path / "agentpacks" packs_dir.mkdir() shell_manifest = dict(_VALID, id="research.shelly", permissions={"shell": True, "file_write": "workspace"}) zp = _make_pack_zip(tmp_path, shell_manifest) with pytest.raises(AgentPackInstallError) as exc: install_from_zip(zp, str(packs_dir)) assert "shell" in str(exc.value).lower() # With allow_shell it goes through. pack = install_from_zip(zp, str(packs_dir), allow_shell=True) assert pack.manifest.permissions.shell is True def test_install_rejects_zip_slip(tmp_path): """A malicious member with ../ must be rejected (SPEC §6 zip-slip).""" packs_dir = tmp_path / "agentpacks" packs_dir.mkdir() zpath = tmp_path / "evil.zip" with zipfile.ZipFile(zpath, "w") as zf: zf.writestr("manifest.yml", yaml.safe_dump(_VALID)) zf.writestr("agents/reviewer.yml", "type: react\n") zf.writestr("../../escape.txt", "pwned") with pytest.raises(AgentPackInstallError) as exc: install_from_zip(str(zpath), str(packs_dir)) assert "escape" in str(exc.value).lower() or "unsafe" in str(exc.value).lower() def test_uninstall_nonexistent_returns_zero(tmp_path): packs_dir = tmp_path / "agentpacks" packs_dir.mkdir() assert uninstall(str(packs_dir), "does.not.exist") == 0 # ───────────────────────────────────────────────────────────────────────────── # Built-in packs (FR-008) — regression guard: the 3 bundled research packs # must always load, validate, and compile. # ───────────────────────────────────────────────────────────────────────────── _BUNDLED_IDS = { "research.top-journal-reviewer", "research.literature-mapper", "research.grant-planner", } def _bundled_pack_dirs(): import glob repo_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) return sorted(glob.glob(os.path.join(repo_root, "agentexample", "agentpacks", "*", "manifest.yml"))) def test_bundled_packs_present(): found = {load_manifest(os.path.dirname(p)).id for p in _bundled_pack_dirs()} assert _BUNDLED_IDS.issubset(found), f"missing bundled packs: {_BUNDLED_IDS - found}" @pytest.mark.parametrize("manifest_path", _bundled_pack_dirs()) def test_bundled_pack_valid_and_compiles(manifest_path): """Each bundled pack: manifest validates, entrypoint exists, and the entrypoint config compiles via from_config (it's a real runnable agent).""" from lambdagent import from_config pack_dir = os.path.dirname(manifest_path) m = load_manifest(pack_dir) assert os.path.isfile(m.entrypoint_path()), m.entrypoint # Built-in packs are local-first: no network. assert m.permissions.network is False # The entrypoint must compile to a runnable term. term = from_config(m.entrypoint_path()) assert term is not None # Policy: bundled packs default to no shell. The one exception — a # Claude-Code-style workspace assistant — may declare shell:true, but # ONLY if its agent guards it with dangerousCommandBlock (runtime safety # net for arbitrary Bash). 杜绝"内置包随便要 shell 又不设防护"。 if m.permissions.shell: import yaml as _yaml cfg = _yaml.safe_load(open(m.entrypoint_path(), encoding="utf-8")) guard = cfg.get("guard") or {} assert guard.get("dangerousCommandBlock") is True, ( f"bundled pack {m.id} 声明 shell:true 但未设 dangerousCommandBlock 防护" ) # ───────────────────────────────────────────────────────────────────────────── # Phase G: agentpack → agent creation endpoint integration # ───────────────────────────────────────────────────────────────────────────── def test_create_agent_from_pack_endpoint(tmp_path, monkeypatch): """End-to-end: install a synthetic pack, POST /agentpacks/{id}/create-agent, verify the new agent row carries pack.path as agent_dir and pack.id as agent_template.""" import json import secrets from fastapi.testclient import TestClient from agentpaas.api.app import app from agentpaas.api.middleware.auth import hash_key from agentpaas.config import settings from agentpaas.db.models import Database, gen_id, now_utc import agentpaas.db.session as _session_mod # Redirect settings.agentpacks_dir into tmp so we don't touch ~/... packs_dir = tmp_path / "agentpacks" packs_dir.mkdir() monkeypatch.setattr(settings, "agentpacks_dir", str(packs_dir)) # Install a pack via the public function. pack_zip = _make_pack_zip(tmp_path, _VALID) pack = install_from_zip(pack_zip, str(packs_dir)) assert pack.id == "research.top-journal-reviewer" # Fresh in-memory DB + a real tenant + key. prev_db = _session_mod._db _session_mod._db = Database("sqlite:///:memory:") db = _session_mod._db try: tid = gen_id("tn_") uid = gen_id("usr_") raw_key = f"ap_{secrets.token_hex(16)}" now = now_utc() db.execute( "INSERT INTO tenants (id, name, plan, status, created_at) " "VALUES (?, 'test', 'free', 'active', ?)", (tid, now), ) db.execute( "INSERT INTO users (id, tenant_id, email, role, created_at) " "VALUES (?, ?, '', 'admin', ?)", (uid, tid, now), ) db.execute( "INSERT INTO api_keys " "(id, tenant_id, user_id, key_hash, key_prefix, name, scopes, " " rate_limit, status, created_at) " "VALUES (?, ?, ?, ?, ?, 'test', ?, 600, 'active', ?)", (gen_id("key_"), tid, uid, hash_key(raw_key), raw_key[:8], json.dumps(["agents:*", "keys:*"]), now), ) db.commit() with TestClient(app) as client: r = client.post( f"/api/v1/agentpacks/{pack.id}/create-agent", json={"name": "My Reviewer", "description": "from a pack"}, headers={"Authorization": f"Bearer {raw_key}"}, ) assert r.status_code == 201, r.text[:300] body = r.json() assert body["ok"] is True agent_id = body["agent_id"] assert body["pack"]["id"] == pack.id assert body["pack"]["version"] == pack.version # Verify DB state: agent_dir = pack.path, agent_template = pack.id. row = db.fetchone( "SELECT agent_dir, agent_template, name FROM agents WHERE id = ?", (agent_id,), ) assert row["agent_dir"] == pack.path assert row["agent_template"] == pack.id assert row["name"] == "My Reviewer" # The initial version's config carries _config_dir = pack.path so # any sub-agent ref inside the pack resolves correctly. ver = db.fetchone( "SELECT config FROM agent_versions WHERE agent_id = ? AND version = 1", (agent_id,), ) cfg = json.loads(ver["config"]) assert cfg.get("_config_dir") == pack.path finally: _session_mod._db = prev_db def test_create_agent_from_unknown_pack_returns_404(tmp_path, monkeypatch): """POST against a pack_id that isn't installed → 404.""" from fastapi.testclient import TestClient from agentpaas.api.app import app from agentpaas.config import settings monkeypatch.setattr(settings, "agentpacks_dir", str(tmp_path / "empty")) (tmp_path / "empty").mkdir() with TestClient(app) as client: # No auth — gate-blocking 401 is the first dependency that fires, # which is fine; the assertion we care about is "no 5xx, no # accidental success". 401 covers both paths cleanly. r = client.post( "/api/v1/agentpacks/does.not.exist/create-agent", json={"name": "x"}, ) assert r.status_code in (401, 404)