|
|
@@ -0,0 +1,236 @@
|
|
|
+"""
|
|
|
+lambdagent.agentpack — AgentPack manifest parsing, validation, permissions.
|
|
|
+
|
|
|
+Spec: docs/AGENTPACK_SPEC.md (CR-20260607-001 rev.2 §6.3 / §7.3).
|
|
|
+
|
|
|
+This is the kernel-side half of the AgentPack feature (CR §10.1): pure,
|
|
|
+testable, no HTTP and no filesystem mutation beyond reading a manifest.
|
|
|
+The install / unzip / HTTP layer lives in agentpaas.api.v1.agentpacks.
|
|
|
+
|
|
|
+An AgentPack is a thin packaging layer over the existing agent-template
|
|
|
+concept — `entrypoint` resolves to an ordinary lambdagent config. This
|
|
|
+module only deals with the manifest (metadata + permissions); running the
|
|
|
+pack is the same `from_config()` path as any other agent.
|
|
|
+"""
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import os
|
|
|
+import re
|
|
|
+from dataclasses import dataclass, field
|
|
|
+from typing import Any, Dict, List
|
|
|
+
|
|
|
+from .core import LambdagentError
|
|
|
+
|
|
|
+
|
|
|
+class AgentPackError(LambdagentError):
|
|
|
+ """Raised on an invalid or unsafe AgentPack manifest."""
|
|
|
+
|
|
|
+
|
|
|
+# ── Permission model (SR-003) ────────────────────────────────────────────
|
|
|
+
|
|
|
+_FILE_WRITE_SCOPES = ("none", "workspace", "knowledge")
|
|
|
+
|
|
|
+
|
|
|
+@dataclass
|
|
|
+class PackPermissions:
|
|
|
+ """Declared capabilities of a pack. Deny-by-default: any field absent
|
|
|
+ from the manifest is the safe value (False / 'none')."""
|
|
|
+ network: bool = False
|
|
|
+ shell: bool = False
|
|
|
+ file_write: str = "workspace" # none | workspace | knowledge
|
|
|
+ read_knowledge: bool = True
|
|
|
+ read_filesystem: bool = False
|
|
|
+
|
|
|
+ @classmethod
|
|
|
+ def from_dict(cls, d: Dict[str, Any]) -> "PackPermissions":
|
|
|
+ if not isinstance(d, dict):
|
|
|
+ raise AgentPackError("manifest.permissions must be a mapping")
|
|
|
+ fw = d.get("file_write", "workspace")
|
|
|
+ if fw not in _FILE_WRITE_SCOPES:
|
|
|
+ raise AgentPackError(
|
|
|
+ f"permissions.file_write must be one of {_FILE_WRITE_SCOPES}, "
|
|
|
+ f"got {fw!r}"
|
|
|
+ )
|
|
|
+ return cls(
|
|
|
+ network=bool(d.get("network", False)),
|
|
|
+ shell=bool(d.get("shell", False)),
|
|
|
+ file_write=fw,
|
|
|
+ read_knowledge=bool(d.get("read_knowledge", True)),
|
|
|
+ read_filesystem=bool(d.get("read_filesystem", False)),
|
|
|
+ )
|
|
|
+
|
|
|
+ def to_dict(self) -> Dict[str, Any]:
|
|
|
+ return {
|
|
|
+ "network": self.network,
|
|
|
+ "shell": self.shell,
|
|
|
+ "file_write": self.file_write,
|
|
|
+ "read_knowledge": self.read_knowledge,
|
|
|
+ "read_filesystem": self.read_filesystem,
|
|
|
+ }
|
|
|
+
|
|
|
+ def summary_zh(self) -> str:
|
|
|
+ """Plain-language permission summary for the install dialog (FR-007)."""
|
|
|
+ parts: List[str] = []
|
|
|
+ parts.append("读取本地知识库" if self.read_knowledge else "不读取知识库")
|
|
|
+ fw = {
|
|
|
+ "none": "不写入任何文件",
|
|
|
+ "workspace": "写入 workspace",
|
|
|
+ "knowledge": "写入 workspace 和知识库",
|
|
|
+ }[self.file_write]
|
|
|
+ parts.append(fw)
|
|
|
+ parts.append("执行 shell" if self.shell else "不执行 shell")
|
|
|
+ parts.append("访问网络" if self.network else "不访问网络")
|
|
|
+ if self.read_filesystem:
|
|
|
+ parts.append("读取知识库以外的文件")
|
|
|
+ return "此智能体包将" + "，".join(parts) + "。"
|
|
|
+
|
|
|
+
|
|
|
+# ── Manifest ──────────────────────────────────────────────────────────────
|
|
|
+
|
|
|
+_ID_RE = re.compile(r"^[a-z0-9]([a-z0-9.\-]*[a-z0-9])?$")
|
|
|
+_SEMVER_RE = re.compile(r"^\d+\.\d+\.\d+([.\-+][0-9A-Za-z.\-]+)?$")
|
|
|
+
|
|
|
+
|
|
|
+@dataclass
|
|
|
+class AgentPackManifest:
|
|
|
+ id: str
|
|
|
+ name: str
|
|
|
+ version: str
|
|
|
+ domain: str
|
|
|
+ entrypoint: str
|
|
|
+ permissions: PackPermissions
|
|
|
+ audience: List[str] = field(default_factory=list)
|
|
|
+ description: str = ""
|
|
|
+ model_recommended: List[str] = field(default_factory=list)
|
|
|
+ author: str = ""
|
|
|
+ homepage: str = ""
|
|
|
+ license: str = ""
|
|
|
+ # Absolute path to the pack root once loaded from disk (else "").
|
|
|
+ root_dir: str = ""
|
|
|
+
|
|
|
+ @classmethod
|
|
|
+ def from_dict(cls, data: Dict[str, Any], root_dir: str = "") -> "AgentPackManifest":
|
|
|
+ if not isinstance(data, dict):
|
|
|
+ raise AgentPackError("manifest must be a YAML mapping")
|
|
|
+
|
|
|
+ def _req(key: str) -> Any:
|
|
|
+ if key not in data or data[key] in (None, ""):
|
|
|
+ raise AgentPackError(f"manifest missing required field: {key!r}")
|
|
|
+ return data[key]
|
|
|
+
|
|
|
+ pid = str(_req("id"))
|
|
|
+ if not _ID_RE.match(pid):
|
|
|
+ raise AgentPackError(
|
|
|
+ f"manifest.id {pid!r} invalid — must match {_ID_RE.pattern}"
|
|
|
+ )
|
|
|
+
|
|
|
+ version = str(_req("version"))
|
|
|
+ if not _SEMVER_RE.match(version):
|
|
|
+ raise AgentPackError(
|
|
|
+ f"manifest.version {version!r} is not semver MAJOR.MINOR.PATCH"
|
|
|
+ )
|
|
|
+
|
|
|
+ entrypoint = str(_req("entrypoint"))
|
|
|
+ # zip-slip / escape guard: entrypoint must be relative and stay inside.
|
|
|
+ norm = os.path.normpath(entrypoint)
|
|
|
+ if os.path.isabs(norm) or norm.startswith(".."):
|
|
|
+ raise AgentPackError(
|
|
|
+ f"manifest.entrypoint {entrypoint!r} must be a relative path "
|
|
|
+ f"inside the pack (no absolute paths, no '..')"
|
|
|
+ )
|
|
|
+
|
|
|
+ perms = PackPermissions.from_dict(_req("permissions"))
|
|
|
+
|
|
|
+ model_block = data.get("model", {}) or {}
|
|
|
+ model_recommended = []
|
|
|
+ if isinstance(model_block, dict):
|
|
|
+ model_recommended = list(model_block.get("recommended", []) or [])
|
|
|
+
|
|
|
+ audience = data.get("audience", []) or []
|
|
|
+ if not isinstance(audience, list):
|
|
|
+ raise AgentPackError("manifest.audience must be a list")
|
|
|
+
|
|
|
+ return cls(
|
|
|
+ id=pid,
|
|
|
+ name=str(_req("name")),
|
|
|
+ version=version,
|
|
|
+ domain=str(_req("domain")),
|
|
|
+ entrypoint=norm,
|
|
|
+ permissions=perms,
|
|
|
+ audience=[str(a) for a in audience],
|
|
|
+ description=str(data.get("description", "") or ""),
|
|
|
+ model_recommended=[str(m) for m in model_recommended],
|
|
|
+ author=str(data.get("author", "") or ""),
|
|
|
+ homepage=str(data.get("homepage", "") or ""),
|
|
|
+ license=str(data.get("license", "") or ""),
|
|
|
+ root_dir=root_dir,
|
|
|
+ )
|
|
|
+
|
|
|
+ def entrypoint_path(self) -> str:
|
|
|
+ """Absolute path to the entrypoint config (requires root_dir set)."""
|
|
|
+ if not self.root_dir:
|
|
|
+ raise AgentPackError("manifest has no root_dir; load from disk first")
|
|
|
+ return os.path.join(self.root_dir, self.entrypoint)
|
|
|
+
|
|
|
+ def to_dict(self) -> Dict[str, Any]:
|
|
|
+ return {
|
|
|
+ "id": self.id,
|
|
|
+ "name": self.name,
|
|
|
+ "version": self.version,
|
|
|
+ "domain": self.domain,
|
|
|
+ "entrypoint": self.entrypoint,
|
|
|
+ "permissions": self.permissions.to_dict(),
|
|
|
+ "audience": self.audience,
|
|
|
+ "description": self.description,
|
|
|
+ "model": {"recommended": self.model_recommended},
|
|
|
+ "author": self.author,
|
|
|
+ "homepage": self.homepage,
|
|
|
+ "license": self.license,
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+def load_manifest(path: str) -> AgentPackManifest:
|
|
|
+ """Load + validate a manifest.yml from disk.
|
|
|
+
|
|
|
+ `path` may point at a manifest.yml file OR a pack directory containing
|
|
|
+ one. Returns a validated AgentPackManifest with root_dir set. Raises
|
|
|
+ AgentPackError on any problem (missing file, bad YAML, schema violation,
|
|
|
+ entrypoint escape, or — when loaded from disk — a missing entrypoint).
|
|
|
+ """
|
|
|
+ import yaml
|
|
|
+
|
|
|
+ if os.path.isdir(path):
|
|
|
+ manifest_path = os.path.join(path, "manifest.yml")
|
|
|
+ root_dir = os.path.abspath(path)
|
|
|
+ else:
|
|
|
+ manifest_path = path
|
|
|
+ root_dir = os.path.abspath(os.path.dirname(path))
|
|
|
+
|
|
|
+ if not os.path.isfile(manifest_path):
|
|
|
+ raise AgentPackError(f"manifest.yml not found: {manifest_path}")
|
|
|
+
|
|
|
+ try:
|
|
|
+ with open(manifest_path, encoding="utf-8") as f:
|
|
|
+ data = yaml.safe_load(f)
|
|
|
+ except yaml.YAMLError as e:
|
|
|
+ raise AgentPackError(f"manifest.yml is not valid YAML: {e}")
|
|
|
+
|
|
|
+ manifest = AgentPackManifest.from_dict(data or {}, root_dir=root_dir)
|
|
|
+
|
|
|
+ # When loaded from disk, the entrypoint must actually exist + stay inside.
|
|
|
+ ep = os.path.abspath(manifest.entrypoint_path())
|
|
|
+ if not (ep == root_dir or ep.startswith(root_dir + os.sep)):
|
|
|
+ raise AgentPackError(
|
|
|
+ f"entrypoint {manifest.entrypoint!r} escapes the pack root"
|
|
|
+ )
|
|
|
+ if not os.path.isfile(ep):
|
|
|
+ raise AgentPackError(
|
|
|
+ f"entrypoint config not found: {manifest.entrypoint}"
|
|
|
+ )
|
|
|
+ return manifest
|
|
|
+
|
|
|
+
|
|
|
+def validate_manifest_dict(data: Dict[str, Any]) -> AgentPackManifest:
|
|
|
+ """Validate a manifest given as a dict (no disk access, no entrypoint
|
|
|
+ existence check). Useful for tests + pre-publish linting."""
|
|
|
+ return AgentPackManifest.from_dict(data)
|
kenny67nju