feat(extensions): add manifest discovery and policy config

This commit is contained in:
Xubin Ren 2026-07-26 16:20:42 +08:00
parent 336b2876d4
commit 863b02d215
8 changed files with 486 additions and 0 deletions

View File

@ -37,6 +37,11 @@ selects the active installation, applies allow/deny policy, and resolves
contribution ownership. Runtime adapters activate only the contributions the
host supports.
Packages expose this metadata as `nanobot.extension.json`. The same canonical
JSON shape is used on disk, over the Node sidecar protocol, and in market
indexes. Unknown fields are rejected so a misspelled permission or contribution
cannot silently change behavior.
The agent loop does not discover or execute plugins. Assembly code resolves
extensions before constructing the runtime and passes native tools, hooks, and
other contributions through the interfaces those subsystems already expose.
@ -84,6 +89,11 @@ and are exposed only through declared host interfaces. Existing workspace,
network, SSRF, and shell restrictions continue to apply to host-provided
operations.
The root `extensions` config controls explicit search paths, allow/deny policy,
per-extension enablement, package-owned config, and workspace trust. Discovery
does not import extension code. Installation does not imply workspace trust,
and activation does not rewrite `config.json` behind the user's back.
## Market boundary
The market is an index, not a runtime. It describes packages available from

View File

@ -407,6 +407,34 @@ class ToolsConfig(Base):
ssrf_whitelist: list[str] = Field(default_factory=list) # CIDR ranges to exempt from SSRF blocking (e.g. ["100.64.0.0/10"] for Tailscale)
class ExtensionEntryConfig(Base):
"""Activation and package-owned config for one installed extension."""
enabled: bool = True
trusted: bool = False
config: dict[str, Any] = Field(default_factory=dict)
class ExtensionsConfig(Base):
"""Discovery and trust policy for first-class extensions."""
enabled: bool = True
paths: list[str] = Field(default_factory=list)
allow: list[str] = Field(default_factory=list)
deny: list[str] = Field(default_factory=list)
entries: dict[str, ExtensionEntryConfig] = Field(default_factory=dict)
workspace_trust: Literal["ask", "allow", "deny"] = "ask"
@model_validator(mode="after")
def _validate_policy(self) -> "ExtensionsConfig":
overlap = set(self.allow) & set(self.deny)
if overlap:
raise ValueError(
f"extension IDs cannot appear in both allow and deny: {sorted(overlap)}"
)
return self
class Config(BaseSettings):
"""Root configuration for nanobot."""
@ -417,6 +445,7 @@ class Config(BaseSettings):
api: ApiConfig = Field(default_factory=ApiConfig)
gateway: GatewayConfig = Field(default_factory=GatewayConfig)
tools: ToolsConfig = Field(default_factory=ToolsConfig)
extensions: ExtensionsConfig = Field(default_factory=ExtensionsConfig)
model_presets: dict[str, ModelPresetConfig] = Field(
default_factory=dict,
validation_alias=AliasChoices("modelPresets", "model_presets"),

View File

@ -1,5 +1,13 @@
"""First-class extension metadata and discovery primitives."""
from nanobot.extensions.codec import (
MANIFEST_FILENAME,
ManifestFormatError,
dump_manifest,
load_manifest,
manifest_from_mapping,
manifest_to_mapping,
)
from nanobot.extensions.manifest import (
EXTENSION_API_VERSION,
ContributionKind,
@ -35,5 +43,11 @@ __all__ = [
"ExtensionRuntime",
"ExtensionScope",
"ExtensionSnapshot",
"MANIFEST_FILENAME",
"ManifestFormatError",
"ResolvedContribution",
"dump_manifest",
"load_manifest",
"manifest_from_mapping",
"manifest_to_mapping",
]

203
nanobot/extensions/codec.py Normal file
View File

@ -0,0 +1,203 @@
"""Strict JSON codec for portable extension manifests."""
from __future__ import annotations
import json
from collections.abc import Mapping
from pathlib import Path
from typing import Any
from nanobot.extensions.manifest import (
EXTENSION_API_VERSION,
ContributionKind,
DependencyKind,
ExtensionContribution,
ExtensionDependency,
ExtensionManifest,
ExtensionPermission,
ExtensionRuntime,
)
MANIFEST_FILENAME = "nanobot.extension.json"
_MANIFEST_KEYS = frozenset(
{
"id",
"name",
"version",
"runtime",
"contributions",
"description",
"dependencies",
"permissions",
"apiVersion",
"homepage",
"license",
}
)
_CONTRIBUTION_KEYS = frozenset(
{"kind", "name", "target", "description", "replaces"}
)
_DEPENDENCY_KEYS = frozenset({"kind", "name", "specifier", "optional"})
_PERMISSION_KEYS = frozenset({"name", "reason"})
class ManifestFormatError(ValueError):
"""Raised when a manifest cannot be decoded unambiguously."""
def load_manifest(path: Path) -> ExtensionManifest:
"""Read and validate one canonical JSON manifest."""
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
raise ManifestFormatError(f"cannot read extension manifest {path}: {exc}") from exc
return manifest_from_mapping(data)
def dump_manifest(manifest: ExtensionManifest, path: Path) -> None:
"""Write one canonical JSON manifest."""
payload = json.dumps(
manifest_to_mapping(manifest),
ensure_ascii=False,
indent=2,
)
path.write_text(payload + "\n", encoding="utf-8")
def manifest_from_mapping(data: object) -> ExtensionManifest:
"""Decode a mapping while rejecting misspelled or ambiguous fields."""
mapping = _mapping(data, "extension manifest")
_reject_unknown(mapping, _MANIFEST_KEYS, "extension manifest")
try:
contributions = tuple(
_contribution_from_mapping(item)
for item in _sequence(mapping.get("contributions", ()), "contributions")
)
dependencies = tuple(
_dependency_from_mapping(item)
for item in _sequence(mapping.get("dependencies", ()), "dependencies")
)
permissions = tuple(
_permission_from_mapping(item)
for item in _sequence(mapping.get("permissions", ()), "permissions")
)
return ExtensionManifest(
id=mapping["id"],
name=mapping["name"],
version=mapping["version"],
runtime=ExtensionRuntime(mapping["runtime"]),
contributions=contributions,
description=mapping.get("description", ""),
dependencies=dependencies,
permissions=permissions,
api_version=mapping.get("apiVersion", EXTENSION_API_VERSION),
homepage=mapping.get("homepage", ""),
license=mapping.get("license", ""),
)
except (KeyError, TypeError, ValueError) as exc:
raise ManifestFormatError(f"invalid extension manifest: {exc}") from exc
def manifest_to_mapping(manifest: ExtensionManifest) -> dict[str, Any]:
"""Return the stable wire representation used by sidecars and catalogs."""
return {
"id": manifest.id,
"name": manifest.name,
"version": manifest.version,
"apiVersion": manifest.api_version,
"runtime": manifest.runtime.value,
"description": manifest.description,
"homepage": manifest.homepage,
"license": manifest.license,
"contributions": [
{
"kind": item.kind.value,
"name": item.name,
"target": item.target,
"description": item.description,
"replaces": list(item.replaces),
}
for item in manifest.contributions
],
"dependencies": [
{
"kind": item.kind.value,
"name": item.name,
"specifier": item.specifier,
"optional": item.optional,
}
for item in manifest.dependencies
],
"permissions": [
{"name": item.name, "reason": item.reason}
for item in manifest.permissions
],
}
def _contribution_from_mapping(data: object) -> ExtensionContribution:
mapping = _mapping(data, "extension contribution")
_reject_unknown(mapping, _CONTRIBUTION_KEYS, "extension contribution")
try:
return ExtensionContribution(
kind=ContributionKind(mapping["kind"]),
name=mapping["name"],
target=mapping.get("target", ""),
description=mapping.get("description", ""),
replaces=tuple(
_sequence(mapping.get("replaces", ()), "contribution replaces")
),
)
except (KeyError, TypeError, ValueError) as exc:
raise ManifestFormatError(f"invalid extension contribution: {exc}") from exc
def _dependency_from_mapping(data: object) -> ExtensionDependency:
mapping = _mapping(data, "extension dependency")
_reject_unknown(mapping, _DEPENDENCY_KEYS, "extension dependency")
try:
return ExtensionDependency(
kind=DependencyKind(mapping["kind"]),
name=mapping["name"],
specifier=mapping.get("specifier", ""),
optional=mapping.get("optional", False),
)
except (KeyError, TypeError, ValueError) as exc:
raise ManifestFormatError(f"invalid extension dependency: {exc}") from exc
def _permission_from_mapping(data: object) -> ExtensionPermission:
mapping = _mapping(data, "extension permission")
_reject_unknown(mapping, _PERMISSION_KEYS, "extension permission")
try:
return ExtensionPermission(
name=mapping["name"],
reason=mapping.get("reason", ""),
)
except (KeyError, TypeError, ValueError) as exc:
raise ManifestFormatError(f"invalid extension permission: {exc}") from exc
def _mapping(data: object, label: str) -> Mapping[str, Any]:
if not isinstance(data, Mapping) or not all(
isinstance(key, str) for key in data
):
raise ManifestFormatError(f"{label} must be a JSON object")
return data
def _sequence(data: object, label: str) -> list[Any] | tuple[Any, ...]:
if not isinstance(data, (list, tuple)):
raise ManifestFormatError(f"{label} must be a JSON array")
return data
def _reject_unknown(
mapping: Mapping[str, Any],
allowed: frozenset[str],
label: str,
) -> None:
unknown = sorted(set(mapping) - allowed)
if unknown:
raise ManifestFormatError(f"{label} has unknown fields: {', '.join(unknown)}")

View File

@ -0,0 +1,77 @@
"""Side-effect-free discovery of extension manifests on disk."""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from nanobot.extensions.codec import MANIFEST_FILENAME, load_manifest
from nanobot.extensions.registry import (
ExtensionCandidate,
ExtensionDiagnostic,
ExtensionScope,
)
@dataclass(frozen=True, slots=True)
class ExtensionDiscoveryResult:
candidates: tuple[ExtensionCandidate, ...] = ()
diagnostics: tuple[ExtensionDiagnostic, ...] = ()
def discover_manifest_root(
root: Path,
*,
scope: ExtensionScope,
trusted: bool = False,
enabled_ids: frozenset[str] = frozenset(),
) -> ExtensionDiscoveryResult:
"""Discover direct children containing ``nanobot.extension.json``."""
if not root.exists():
return ExtensionDiscoveryResult()
if not root.is_dir():
return ExtensionDiscoveryResult(
diagnostics=(
ExtensionDiagnostic(
code="invalid_extension_root",
extension_id="",
message=f"extension root is not a directory: {root}",
),
)
)
manifests = []
direct_manifest = root / MANIFEST_FILENAME
if direct_manifest.is_file():
manifests.append(direct_manifest)
manifests.extend(
sorted(
path / MANIFEST_FILENAME
for path in root.iterdir()
if path.is_dir() and (path / MANIFEST_FILENAME).is_file()
)
)
candidates: list[ExtensionCandidate] = []
diagnostics: list[ExtensionDiagnostic] = []
for path in manifests:
try:
manifest = load_manifest(path)
candidates.append(
ExtensionCandidate(
manifest=manifest,
scope=scope,
location=path.parent.resolve(),
enabled=not enabled_ids or manifest.id in enabled_ids,
trusted=trusted,
)
)
except Exception as exc:
diagnostics.append(
ExtensionDiagnostic(
code="invalid_manifest",
extension_id=path.parent.name,
message=str(exc),
)
)
return ExtensionDiscoveryResult(tuple(candidates), tuple(diagnostics))

View File

@ -0,0 +1,29 @@
import pytest
from pydantic import ValidationError
from nanobot.config.schema import Config, ExtensionsConfig
def test_extensions_config_accepts_camel_case_workspace_trust() -> None:
config = Config.model_validate(
{
"extensions": {
"workspaceTrust": "allow",
"entries": {
"acme": {
"enabled": True,
"trusted": True,
"config": {"endpoint": "https://example.com"},
}
},
}
}
)
assert config.extensions.workspace_trust == "allow"
assert config.extensions.entries["acme"].trusted is True
def test_extensions_config_rejects_overlapping_policy() -> None:
with pytest.raises(ValidationError, match="both allow and deny"):
ExtensionsConfig(allow=["acme"], deny=["acme"])

View File

@ -0,0 +1,62 @@
import json
import pytest
from nanobot.extensions import (
ContributionKind,
ExtensionContribution,
ExtensionManifest,
ExtensionRuntime,
ManifestFormatError,
dump_manifest,
load_manifest,
manifest_from_mapping,
)
def test_manifest_json_round_trip(tmp_path) -> None:
path = tmp_path / "nanobot.extension.json"
original = ExtensionManifest(
id="acme.tools",
name="Acme Tools",
version="2.0.0",
runtime=ExtensionRuntime.PI,
contributions=(
ExtensionContribution(
kind=ContributionKind.TOOL,
name="acme_search",
target="./index.ts#search",
),
),
)
dump_manifest(original, path)
assert load_manifest(path) == original
assert json.loads(path.read_text())["apiVersion"] == 1
def test_manifest_rejects_unknown_fields() -> None:
with pytest.raises(ManifestFormatError, match="unknown fields: typo"):
manifest_from_mapping(
{
"id": "bad",
"name": "Bad",
"version": "1.0.0",
"runtime": "python",
"typo": True,
}
)
def test_manifest_rejects_unknown_contribution_kind() -> None:
with pytest.raises(ManifestFormatError, match="invalid extension contribution"):
manifest_from_mapping(
{
"id": "bad",
"name": "Bad",
"version": "1.0.0",
"runtime": "python",
"contributions": [{"kind": "mystery", "name": "unknown"}],
}
)

View File

@ -0,0 +1,62 @@
import json
from nanobot.extensions import ExtensionScope
from nanobot.extensions.discovery import discover_manifest_root
def _write_manifest(root, name: str, extension_id: str) -> None:
package = root / name
package.mkdir()
(package / "nanobot.extension.json").write_text(
json.dumps(
{
"id": extension_id,
"name": extension_id,
"version": "1.0.0",
"runtime": "declarative",
}
)
)
def test_discovery_reads_metadata_without_importing_runtime(tmp_path) -> None:
_write_manifest(tmp_path, "one", "one")
_write_manifest(tmp_path, "two", "two")
result = discover_manifest_root(
tmp_path,
scope=ExtensionScope.WORKSPACE,
trusted=True,
)
assert [candidate.manifest.id for candidate in result.candidates] == ["one", "two"]
assert all(candidate.trusted for candidate in result.candidates)
assert result.diagnostics == ()
def test_discovery_reports_bad_manifest_without_hiding_good_packages(tmp_path) -> None:
_write_manifest(tmp_path, "good", "good")
bad = tmp_path / "bad"
bad.mkdir()
(bad / "nanobot.extension.json").write_text("{")
result = discover_manifest_root(tmp_path, scope=ExtensionScope.USER)
assert [candidate.manifest.id for candidate in result.candidates] == ["good"]
assert result.diagnostics[0].code == "invalid_manifest"
def test_discovery_can_leave_unselected_packages_disabled(tmp_path) -> None:
_write_manifest(tmp_path, "one", "one")
_write_manifest(tmp_path, "two", "two")
result = discover_manifest_root(
tmp_path,
scope=ExtensionScope.USER,
enabled_ids=frozenset({"two"}),
)
assert [(item.manifest.id, item.enabled) for item in result.candidates] == [
("one", False),
("two", True),
]