From 5394403917d24487e263ddb407a8a5022dad0f14 Mon Sep 17 00:00:00 2001 From: Aly Raffauf Date: Sat, 27 Jun 2026 21:56:09 -0400 Subject: [PATCH] manifest: migrate to pydantic --- tartarus/manifest.py | 337 +++++++++++++++-- tartarus/manifest_loader.py | 692 ++++++++-------------------------- tests/test_broker.py | 28 +- tests/test_bundle.py | 2 +- tests/test_manifest.py | 25 +- tests/test_manifest_loader.py | 66 ++-- tests/test_policy.py | 9 +- 7 files changed, 544 insertions(+), 615 deletions(-) diff --git a/tartarus/manifest.py b/tartarus/manifest.py index f22f947..dcd2937 100644 --- a/tartarus/manifest.py +++ b/tartarus/manifest.py @@ -5,43 +5,118 @@ from the same source: `build_manifest` projects every non-deny capability into a tool. """ -from typing import Any +from __future__ import annotations -from dataclasses import dataclass, field +import string +from typing import Any, Literal +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator +from typing_extensions import Self -@dataclass(frozen=True) -class Grant: + +# ── Grant ──────────────────────────────────────────────────────────────────── + + +class Grant(BaseModel): """The host reach a capability opens. Empty means "nothing beyond the shell".""" - package_bins: list[str] = field(default_factory=list) - allowed_hosts: list[str] = field(default_factory=list) - writable: list[str] = field(default_factory=list) - unrestricted: bool = False + model_config = ConfigDict(frozen=True) + + package_bins: list[str] = Field(default_factory=list) + allowed_hosts: list[str] = Field(default_factory=list) + writable: list[str] = Field(default_factory=list) + unrestricted: bool = Field(default=False, strict=True) # The store path of the `closureInfo` `store-paths` file for this grant's # packages (emitted by Nix). `closure_paths` is its realized contents — the # exact store paths the jail binds, so the capability reaches its declared # closure and nothing else. Populated after realization (manifest_loader). closure_file: str = "" - closure_paths: list[str] = field(default_factory=list) + closure_paths: list[str] = Field(default_factory=list) + + @field_validator("package_bins") + @classmethod + def _validate_package_bins(cls, v: list[str]) -> list[str]: + for entry in v: + if not entry.startswith("/"): + raise ValueError( + f"packageBins entry '{entry}' must be an absolute path" + ) + if not entry.startswith("/nix/store/"): + raise ValueError( + f"packageBins entry '{entry}' must be under /nix/store" + ) + if not entry.endswith("/bin"): + raise ValueError( + f"packageBins entry '{entry}' must end with /bin" + ) + return v + + @field_validator("writable") + @classmethod + def _validate_writable(cls, v: list[str]) -> list[str]: + for entry in v: + if entry.startswith("/"): + raise ValueError( + f"writable path '{entry}' must be relative" + ) + if ".." in entry.split("/"): + raise ValueError( + f"writable path '{entry}' escapes the work tree" + ) + return v + @field_validator("closure_file") + @classmethod + def _validate_closure_file(cls, v: str) -> str: + if not v: + return v + if not v.startswith("/nix/store/"): + raise ValueError("closure must be under /nix/store") + if not v.endswith("/store-paths"): + raise ValueError("closure must end with /store-paths") + return v -@dataclass(frozen=True) -class Param: - type: str # JSON Schema scalar: "string" | "integer" | "boolean" | "array" - description: str - required: bool = False + +# ── Param ──────────────────────────────────────────────────────────────────── + + +class Param(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + type: Literal["string", "integer", "boolean", "array"] + description: str = "" + required: bool = Field(default=False, strict=True) enum: list[Any] | None = None + @model_validator(mode="after") + def _validate_enum_entries(self) -> Self: + if self.enum is None: + return self + for entry in self.enum: + if self.type == "string" and not isinstance(entry, str): + raise ValueError("enum entries must match type 'string'") + if self.type == "integer": + if isinstance(entry, bool) or not isinstance(entry, int): + raise ValueError("enum entries must match type 'integer'") + if self.type == "boolean" and not isinstance(entry, bool): + raise ValueError("enum entries must match type 'boolean'") + if self.type == "array" and not isinstance(entry, list): + raise ValueError("enum entries must match type 'array'") + return self + + +# ── Capability ─────────────────────────────────────────────────────────────── + + +class Capability(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") -@dataclass(frozen=True) -class Capability: name: str - description: str - policy: str # "auto" | "ask-once" | "ask-always" | "deny" + description: str = "" + policy: Literal["auto", "ask-once", "ask-always", "deny"] params: dict[str, Param] grants: Grant - runner: str + runner: str = "" # Per-capability wall-clock budget in seconds. None means the capability runs # unbounded; a declared value caps it at that many seconds. timeout: int | None = None @@ -49,35 +124,144 @@ class Capability: # "command" — run in the jail, capture output, return one result (default). # "background" — launch detached, return a handle; track in the registry. # "control" — operate on the background registry (see `control`); no jail. - kind: str = "command" + kind: Literal["command", "background", "control"] = "command" # For kind == "control": which registry operation this tool performs, # one of "status" | "output" | "stop". None for every other kind. - control: str | None = None + control: Literal["status", "output", "stop"] | None = None + + @field_validator("timeout", mode="before") + @classmethod + def _reject_bool_timeout(cls, v: object) -> int | None: + if v is None: + return None + if isinstance(v, bool): + raise ValueError("timeout must be a positive integer") + if not isinstance(v, int) or v <= 0: + raise ValueError("timeout must be a positive integer") + return v + + @model_validator(mode="after") + def _validate_unrestricted(self) -> Self: + if self.grants.unrestricted and self.policy == "auto": + raise ValueError( + "unrestricted grant is not allowed under 'auto' policy; " + "the unrestricted escape must never be silent" + ) + return self + + @model_validator(mode="after") + def _validate_kind_rules(self) -> Self: + if self.kind == "control": + if self.control is None: + raise ValueError("control capability must declare a control op") + if self.runner: + raise ValueError("control capability must not declare a runner") + if _grant_opens_reach(self.grants): + raise ValueError("control capability must not declare grants") + else: + if self.control is not None: + raise ValueError( + "'control' is only valid for kind 'control'" + ) + if self.kind == "background": + if self.grants.unrestricted: + raise ValueError("background capability cannot be unrestricted") + if self.timeout is not None: + raise ValueError( + "background capability cannot declare a timeout; " + "a background task runs until it exits or is stopped" + ) + return self + + @model_validator(mode="after") + def _validate_runner_placeholders(self) -> Self: + for _, field_name, _, _ in string.Formatter().parse(self.runner): + if field_name and field_name not in self.params: + raise ValueError( + f"runner references undeclared param '{field_name}'" + ) + return self + + +def _grant_opens_reach(grant: Grant) -> bool: + return bool( + grant.package_bins + or grant.allowed_hosts + or grant.writable + or grant.unrestricted + ) + + +# ── ModelConfig ────────────────────────────────────────────────────────────── Sampling = dict[str, int | float] -@dataclass(frozen=True) -class ModelConfig: +_RESERVED_SAMPLING_KEYS = frozenset( + {"model", "max_tokens", "messages", "stream", "tools", "tool_choice"} +) + + +class ModelConfig(BaseModel): """The model an agent declares: backend binding + inference knobs (PLAN.md §9). A model id is only meaningful next to the base_url that serves it, so they - travel together alongside provider-portable inference knobs. Secrets and + travel together alongside provider-portable inference knobs. Secrets and deployment-specific headers are sourced from the environment, never from the Nix store. """ + model_config = ConfigDict(frozen=True, extra="forbid") + base_url: str | None = None name: str | None = None provider: str | None = None # provider type, e.g. "openai-compat" max_tokens: int | None = None sampling: Sampling | None = None + @field_validator("base_url", "name", "provider") + @classmethod + def _reject_empty_strings(cls, v: str | None) -> str | None: + if v is not None and (not isinstance(v, str) or v == ""): + raise ValueError("must be a non-empty string") + return v + + @field_validator("max_tokens", mode="before") + @classmethod + def _reject_bool_max_tokens(cls, v: object) -> int | None: + if v is None: + return None + if isinstance(v, bool) or not isinstance(v, int) or v <= 0: + raise ValueError("maxTokens must be a positive integer") + return v + + @field_validator("sampling", mode="before") + @classmethod + def _validate_sampling(cls, v: object) -> object: + if v is None: + return None + if not isinstance(v, dict): + raise ValueError("sampling must be an object") + for key, val in v.items(): + if key in _RESERVED_SAMPLING_KEYS: + raise ValueError( + f"model sampling key '{key}' is reserved and cannot be overridden" + ) + if isinstance(val, bool) or not isinstance(val, (int, float)): + raise ValueError( + f"model sampling '{key}' must be a number" + ) + return v + -@dataclass(frozen=True) -class Manifest: - tools: list[dict] # provider-neutral tool defs (name/description/parameters) +# ── Manifest ───────────────────────────────────────────────────────────────── + + +class Manifest(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + tools: list[dict] capabilities: dict[str, Capability] # The agent's persona, declared in Nix. None when the agent declares none. system_prompt: str | None = None @@ -95,7 +279,99 @@ class Manifest: # bundle), as the `store-paths` file path emitted by Nix and its realized # contents. `shell_closure` is filled after realization (manifest_loader). shell_closure_file: str = "" - shell_closure: list[str] = field(default_factory=list) + shell_closure: list[str] = Field(default_factory=list) + + @field_validator("ca_bundle_file") + @classmethod + def _validate_ca_bundle(cls, v: str) -> str: + if not v or not v.startswith("/nix/store/"): + raise ValueError("caBundle must be under /nix/store") + return v + + @field_validator("shell_closure_file") + @classmethod + def _validate_shell_closure_file(cls, v: str) -> str: + if not v: + raise ValueError("shellClosure is required") + if not v.startswith("/nix/store/"): + raise ValueError("shellClosure must be under /nix/store") + if not v.endswith("/store-paths"): + raise ValueError("shellClosure must end with /store-paths") + return v + + @field_validator("shell_path") + @classmethod + def _validate_shell_path_entries(cls, v: str) -> str: + for entry in v.split(":"): + if not entry: + continue + if not entry.startswith("/nix/store/"): + raise ValueError( + f"shellPath entry '{entry}' must be under /nix/store" + ) + if not entry.endswith("/bin"): + raise ValueError( + f"shellPath entry '{entry}' must end with /bin" + ) + return v + + @model_validator(mode="after") + def _validate_tools_consistency(self) -> Self: + for tool in self.tools: + if not isinstance(tool, dict): + raise ValueError("tool entries must be objects") + name = tool.get("name") + if not isinstance(name, str): + raise ValueError("tool 'name' must be a string") + capability = self.capabilities.get(name) + if capability is None: + raise ValueError(f"tool '{name}' has no matching capability") + if capability.policy == "deny": + raise ValueError( + f"tool '{name}' is exposed but its capability policy is 'deny'" + ) + _validate_tool_schema_against_params(tool, capability) + return self + + +def _validate_tool_schema_against_params( + tool: dict[str, Any], capability: Capability +) -> None: + schema = tool.get("parameters", {}) + if not isinstance(schema, dict): + raise ValueError(f"tool '{capability.name}' parameters must be an object") + properties = schema.get("properties", {}) + if not isinstance(properties, dict): + raise ValueError( + f"tool '{capability.name}' parameters properties must be an object" + ) + required = schema.get("required", []) + if not isinstance(required, list): + raise ValueError( + f"tool '{capability.name}' parameters required must be a list" + ) + + schema_properties = set(properties.keys()) + schema_required = set(required) + + declared = set(capability.params.keys()) + declared_required = { + name for name, param in capability.params.items() if param.required + } + + if schema_properties != declared: + raise ValueError( + f"tool '{capability.name}' parameters {sorted(schema_properties)} " + f"do not match capability params {sorted(declared)}" + ) + if schema_required != declared_required: + raise ValueError( + f"tool '{capability.name}' required {sorted(schema_required)} does not " + f"match capability required {sorted(declared_required)}" + ) + + +# ── Tool projection helpers ────────────────────────────────────────────────── def _params_to_json_schema(params: dict[str, Param]) -> dict[str, Any]: @@ -103,7 +379,10 @@ def _params_to_json_schema(params: dict[str, Param]) -> dict[str, Any]: properties: dict[str, Any] = {} required: list[str] = [] for name, param in params.items(): - schema: dict[str, Any] = {"type": param.type, "description": param.description} + schema: dict[str, Any] = { + "type": param.type, + "description": param.description, + } if param.enum is not None: schema["enum"] = param.enum properties[name] = schema diff --git a/tartarus/manifest_loader.py b/tartarus/manifest_loader.py index dab859f..c6937f0 100644 --- a/tartarus/manifest_loader.py +++ b/tartarus/manifest_loader.py @@ -11,16 +11,11 @@ realized store artifacts the manifest points at. import os import platform -import string -from dataclasses import replace from typing import Any, cast -from tartarus.manifest import Capability, Grant, Manifest, ModelConfig, Param, Sampling +from pydantic import ValidationError -VALID_POLICIES = frozenset({"auto", "ask-once", "ask-always", "deny"}) -VALID_PARAM_TYPES = frozenset({"string", "integer", "boolean", "array"}) -VALID_KINDS = frozenset({"command", "background", "control"}) -VALID_CONTROL_OPS = frozenset({"status", "output", "stop"}) +from tartarus.manifest import Capability, Manifest class ManifestError(Exception): @@ -45,505 +40,198 @@ def host_system() -> str: def build_manifest_from_raw(raw: object) -> Manifest: - """Validate decoded manifest JSON against §5 and build typed objects.""" - if not isinstance(raw, dict): - raise ManifestError("manifest must be a JSON object") - - typed_raw = cast(dict[str, Any], raw) - tools = typed_raw.get("tools") - raw_capabilities = typed_raw.get("capabilities") - if not isinstance(tools, list): - raise ManifestError("manifest 'tools' must be a list") - if not isinstance(raw_capabilities, dict): - raise ManifestError("manifest 'capabilities' must be an object") - - capabilities = { - name: _build_capability(name, body) for name, body in raw_capabilities.items() - } - _validate_tools(tools, capabilities) - - system_prompt = typed_raw.get("systemPrompt") - if system_prompt is not None and not isinstance(system_prompt, str): - raise ManifestError("manifest 'systemPrompt' must be a string") - - model = _build_model_config(typed_raw.get("model")) - - ca_bundle_file = _required_store_path(typed_raw, "caBundle") - - shell_closure_file = _required_store_paths_file(typed_raw, "shellClosure") - - shell_path = typed_raw.get("shellPath", "") - if not isinstance(shell_path, str): - raise ManifestError("manifest 'shellPath' must be a string") - _validate_shell_path(shell_path) - - return Manifest( - tools=tools, - capabilities=capabilities, - system_prompt=system_prompt, - model=model, - ca_bundle_file=ca_bundle_file, - shell_path=shell_path, - shell_closure_file=shell_closure_file, - ) + """Validate decoded manifest JSON and build typed objects. - -def _required_store_path(raw: dict[str, Any], field_name: str) -> str: - if field_name not in raw: - raise ManifestError(f"manifest '{field_name}' is required") - - value = raw[field_name] - if not isinstance(value, str): - raise ManifestError(f"manifest '{field_name}' must be a string") - if not value: - raise ManifestError(f"manifest '{field_name}' is required") - if not value.startswith("/nix/store/"): - raise ManifestError(f"manifest '{field_name}' must be under /nix/store") - return value - - -def _required_store_paths_file(raw: dict[str, Any], field_name: str) -> str: - if field_name not in raw: - raise ManifestError(f"manifest '{field_name}' is required") - - value = raw[field_name] - _validate_store_paths_file(f"manifest '{field_name}'", value) - return value - - -def _object_field(raw: dict[str, Any], key: str, label: str) -> dict[str, Any]: - value = raw.get(key) - if value is None: - return {} - if not isinstance(value, dict): - raise ManifestError(f"{label} '{key}' must be an object") - return cast(dict[str, Any], value) - - -def _list_field(raw: dict[str, Any], key: str, label: str) -> list[Any]: - value = raw.get(key) - if value is None: - return [] - if not isinstance(value, list): - raise ManifestError(f"{label} '{key}' must be a list") - return value - - -def _optional_string_field( - raw: dict[str, Any], key: str, label: str, default: str = "" -) -> str: - value = raw.get(key, default) - if not isinstance(value, str): - raise ManifestError(f"{label} '{key}' must be a string") - return value - - -def _optional_bool_field(raw: dict[str, Any], key: str, label: str) -> bool: - value = raw.get(key, False) - if not isinstance(value, bool): - raise ManifestError(f"{label} '{key}' must be a boolean") - return value - - -def _validate_store_paths_file(label: str, value: object) -> None: - if not isinstance(value, str): - raise ManifestError(f"{label} must be a string") - if not value: - raise ManifestError(f"{label} is required") - if not value.startswith("/nix/store/"): - raise ManifestError(f"{label} must be under /nix/store") - if not value.endswith("/store-paths"): - raise ManifestError(f"{label} must end with /store-paths") - - -def _validate_shell_path(shell_path: str) -> None: - """Each baked PATH entry must be an absolute `/nix/store/**/bin` dir. - - Absent is tolerated (the Phase-0 / no-Nix path); a real bundle always bakes - it from the declared shell packages. + Maps JSON camelCase keys to Python snake_case, flattens the nested + `network.allowedHosts` grant field, then delegates all type/shape/cross-field + validation to Pydantic models. """ - for entry in shell_path.split(":"): - if not entry: - continue - if not entry.startswith("/nix/store/"): - raise ManifestError( - f"manifest shellPath entry '{entry}' must be under /nix/store" - ) - if not entry.endswith("/bin"): - raise ManifestError( - f"manifest shellPath entry '{entry}' must end with /bin" - ) - - -def _build_model_config(raw: object) -> ModelConfig | None: - """Validate the optional `model` block (PLAN.md §9). Fails closed.""" - if raw is None: - return None if not isinstance(raw, dict): - raise ManifestError("manifest 'model' must be an object") - - _reject_unknown_model_keys(raw) - - base_url = _optional_nonempty_str(raw, "baseUrl") - name = _optional_nonempty_str(raw, "name") - provider = _optional_nonempty_str(raw, "provider") - - return ModelConfig( - base_url=base_url, - name=name, - provider=provider, - max_tokens=_build_max_tokens(raw.get("maxTokens")), - sampling=_build_sampling(raw.get("sampling")), - ) + raise ManifestError("manifest must be a JSON object") + mapped = _map_manifest_raw(cast(dict[str, Any], raw)) -_MODEL_KEYS = frozenset({"provider", "baseUrl", "name", "maxTokens", "sampling"}) + try: + return Manifest.model_validate(mapped) + except ValidationError as error: + raise ManifestError(_format_validation_error(error)) from error -def _reject_unknown_model_keys(raw: dict) -> None: - unknown_keys = sorted(set(raw) - _MODEL_KEYS) - if unknown_keys: +def validate_realized_package_bins(manifest: Manifest) -> None: + missing_bins = [ + package_bin + for capability in manifest.capabilities.values() + for package_bin in capability.grants.package_bins + if not os.path.isdir(package_bin) + ] + if missing_bins: raise ManifestError( - "manifest 'model' has unsupported keys: " + ", ".join(unknown_keys) + "realized packageBins are missing directories: " + ", ".join(missing_bins) ) -def _optional_nonempty_str(raw: dict, key: str) -> str | None: - value = raw.get(key) - if value is None: - return None - if not isinstance(value, str) or not value: - raise ManifestError(f"model '{key}' must be a non-empty string") - return value - - -def _build_max_tokens(raw: object) -> int | None: - if raw is None: - return None - # bool is an int subclass; reject it so `maxTokens = true` does not pass. - if isinstance(raw, bool) or not isinstance(raw, int) or raw <= 0: - raise ManifestError("model 'maxTokens' must be a positive integer") - return raw - - -_RESERVED_SAMPLING_KEYS = frozenset( - {"model", "max_tokens", "messages", "stream", "tools", "tool_choice"} -) - +def resolve_realized_closures(manifest: Manifest) -> Manifest: + """Read each emitted `store-paths` file into its grant's `closure_paths`. -def _build_sampling(raw: object) -> Sampling | None: - if raw is None: - return None - if not isinstance(raw, dict): - raise ManifestError("model 'sampling' must be an object") - sampling: Sampling = {} - for key, value in raw.items(): - if not isinstance(key, str): - raise ManifestError("model 'sampling' keys must be strings") - if key in _RESERVED_SAMPLING_KEYS: - raise ManifestError( - f"model sampling key '{key}' is reserved and cannot be overridden" - ) - if isinstance(value, bool) or not isinstance(value, (int, float)): - raise ManifestError(f"model sampling '{key}' must be a number") - sampling[key] = value - return sampling + Runs after the bundle is realized, so the files exist on disk. The jail + binds exactly these paths plus the agent's baseline `shell_closure`, so a + capability reaches its declared closure and nothing else. Fails closed: a + missing or malformed closure file refuses to start (mirrors + `validate_realized_package_bins`). + """ + shell_closure = _read_closure_file("agent shell", manifest.shell_closure_file) + capabilities = { + name: capability.model_copy( + update={ + "grants": capability.grants.model_copy( + update={ + "closure_paths": _read_closure_file( + f"capability '{name}'", capability.grants.closure_file + ), + } + ) + } + ) + for name, capability in manifest.capabilities.items() + } + return manifest.model_copy( + update={"capabilities": capabilities, "shell_closure": shell_closure} + ) def _build_capability(name: str, body: object) -> Capability: - if not isinstance(body, dict): - raise ManifestError(f"capability '{name}' must be an object") - typed_body = cast(dict[str, Any], body) - - policy = typed_body.get("policy") - if policy not in VALID_POLICIES: - raise ManifestError( - f"capability '{name}' has invalid policy {policy!r}; " - f"expected one of {sorted(VALID_POLICIES)}" - ) + """Build a single Capability from a raw body. Thin wrapper around Pydantic. - params = { - param_name: _build_param(name, param_name, param_body) - for param_name, param_body in _object_field( - typed_body, "params", f"capability '{name}'" - ).items() - } - grants = _build_grant( - name, _object_field(typed_body, "grants", f"capability '{name}'") - ) - description = _optional_string_field( - typed_body, "description", f"capability '{name}'" - ) - runner = _optional_string_field(typed_body, "runner", f"capability '{name}'") - timeout = _build_timeout(name, typed_body.get("timeout")) - kind = _build_kind(name, typed_body.get("kind")) - control = _build_control(name, kind, typed_body.get("control")) - - _validate_runner_placeholders(name, runner, params) - _validate_unrestricted(name, policy, grants) - _validate_kind(name, kind, grants, runner, timeout) - - return Capability( - name=name, - description=description, - policy=policy, - params=params, - grants=grants, - runner=runner, - timeout=timeout, - kind=kind, - control=control, - ) + Exposed so tests can validate capability-level rules in isolation + without building a full manifest. + """ + mapped = _map_capability_raw(name, body) + try: + return Capability.model_validate(mapped) + except ValidationError as error: + raise ManifestError(_format_validation_error(error)) from error -def _build_kind(capability: str, raw: object) -> str: - if raw is None: - return "command" - if not (isinstance(raw, str) and raw in VALID_KINDS): - raise ManifestError( - f"capability '{capability}' has invalid kind {raw!r}; " - f"expected one of {sorted(VALID_KINDS)}" - ) - return raw +# ── JSON → Python key mapping ──────────────────────────────────────────────── -def _build_control(capability: str, kind: str, raw: object) -> str | None: - if kind != "control": - if raw is not None: - raise ManifestError( - f"capability '{capability}' sets 'control' but kind is {kind!r}; " - "'control' is only valid for kind 'control'" - ) - return None - if not (isinstance(raw, str) and raw in VALID_CONTROL_OPS): - raise ManifestError( - f"control capability '{capability}' has invalid control op {raw!r}; " - f"expected one of {sorted(VALID_CONTROL_OPS)}" - ) - return raw +def _map_manifest_raw(raw: dict[str, Any]) -> dict[str, Any]: + mapped: dict[str, Any] = { + "tools": raw.get("tools", []), + "capabilities": _map_capabilities(raw.get("capabilities", {})), + "ca_bundle_file": raw.get("caBundle", ""), + "shell_closure_file": raw.get("shellClosure", ""), + "shell_path": raw.get("shellPath", ""), + } + if "systemPrompt" in raw: + mapped["system_prompt"] = raw["systemPrompt"] + if "model" in raw: + mapped["model"] = _map_model_raw(raw["model"]) + return mapped + + +def _map_capabilities(raw: object) -> dict[str, Any]: + capabilities = _require_object(raw, "manifest 'capabilities'") + return { + name: _map_capability_raw(name, body) + for name, body in capabilities.items() + } -def _validate_kind( - capability: str, kind: str, grants: Grant, runner: str, timeout: int | None -) -> None: - """Enforce the structural rules each capability kind requires.""" - if kind == "control": - # A control capability acts on the registry, never the jail: it must - # carry no runner and open no host reach. - if runner: - raise ManifestError( - f"control capability '{capability}' must not declare a runner" - ) - if _grant_opens_reach(grants): - raise ManifestError( - f"control capability '{capability}' must not declare grants" - ) - elif kind == "background": - # A detached run is unbounded by definition (stop it via a control tool), - # and the unrestricted host-escape is out of scope for background launch. - if grants.unrestricted: - raise ManifestError( - f"background capability '{capability}' cannot be unrestricted" - ) - if timeout is not None: - raise ManifestError( - f"background capability '{capability}' cannot declare a timeout; " - "a background task runs until it exits or is stopped" - ) +def _map_capability_raw(name: str, body: object) -> dict[str, Any]: + capability = _require_object(body, f"capability '{name}'") + return { + "name": name, + "description": capability.get("description", ""), + "policy": capability.get("policy"), + "params": _require_object( + capability.get("params", {}), f"capability '{name}' params" + ), + "grants": _map_grant_raw(name, capability.get("grants", {})), + "runner": capability.get("runner", ""), + "timeout": capability.get("timeout"), + "kind": capability.get("kind", "command"), + "control": capability.get("control"), + } -def _grant_opens_reach(grant: Grant) -> bool: - return bool( - grant.package_bins - or grant.allowed_hosts - or grant.writable - or grant.unrestricted +def _map_grant_raw(capability_name: str, body: object) -> dict[str, Any]: + grants = _require_object(body, f"capability '{capability_name}' grants") + network = _require_object( + grants.get("network", {}), f"capability '{capability_name}' grants network" ) + return { + "package_bins": grants.get("packageBins", []), + "allowed_hosts": network.get("allowedHosts", []), + "writable": grants.get("writable", []), + "unrestricted": grants.get("unrestricted", False), + "closure_file": grants.get("closure", ""), + } -def _build_timeout(capability: str, raw: object) -> int | None: - """Validate the optional per-capability timeout. - - Absent (None) means the capability runs unbounded; a declared value caps it - at that many seconds. - """ - if raw is None: - return None - # bool is an int subclass; reject it so `timeout = true` does not pass. - if isinstance(raw, bool) or not isinstance(raw, int) or raw <= 0: - raise ManifestError( - f"capability '{capability}' timeout must be a positive integer" - ) - return raw +_KNOWN_MODEL_KEYS = frozenset({"provider", "baseUrl", "name", "maxTokens", "sampling"}) -def _build_param(capability: str, name: str, body: object) -> Param: - if not isinstance(body, dict): - raise ManifestError(f"param '{capability}.{name}' must be an object") - typed_body = cast(dict[str, Any], body) +def _map_model_raw(raw: object) -> dict[str, Any]: + body = _require_object(raw, "manifest 'model'") - param_type = typed_body.get("type") - if param_type not in VALID_PARAM_TYPES: + unknown_keys = sorted(set(body) - _KNOWN_MODEL_KEYS) + if unknown_keys: raise ManifestError( - f"param '{capability}.{name}' has invalid type {param_type!r}; " - f"expected one of {sorted(VALID_PARAM_TYPES)}" + "manifest 'model' has unsupported keys: " + ", ".join(unknown_keys) ) - description = _optional_string_field( - typed_body, "description", f"param '{capability}.{name}'" - ) - required = _optional_bool_field( - typed_body, "required", f"param '{capability}.{name}'" - ) - enum = _build_param_enum(capability, name, param_type, typed_body.get("enum")) - return Param( - type=param_type, - description=description, - required=required, - enum=enum, - ) + mapped: dict[str, Any] = {} + for json_key, py_key in ( + ("baseUrl", "base_url"), + ("maxTokens", "max_tokens"), + ): + if json_key in body: + mapped[py_key] = body[json_key] -def _build_param_enum( - capability: str, name: str, param_type: str, raw: object -) -> list[Any] | None: - if raw is None: - return None - if not isinstance(raw, list): - raise ManifestError(f"param '{capability}.{name}' enum must be a list") - for entry in raw: - if not _enum_entry_matches_type(entry, param_type): - raise ManifestError( - f"param '{capability}.{name}' enum entries must match " - f"type '{param_type}'" - ) - return raw - - -def _enum_entry_matches_type(entry: object, param_type: str) -> bool: - if param_type == "string": - return isinstance(entry, str) - if param_type == "integer": - return not isinstance(entry, bool) and isinstance(entry, int) - if param_type == "boolean": - return isinstance(entry, bool) - if param_type == "array": - return isinstance(entry, list) - return False - - -def _build_grant(capability: str, body: dict[str, Any]) -> Grant: - network = _object_field(body, "network", f"capability '{capability}' grants") - package_bins = _list_field(body, "packageBins", f"capability '{capability}' grants") - writable = _list_field(body, "writable", f"capability '{capability}' grants") - allowed_hosts = _list_field( - network, "allowedHosts", f"capability '{capability}' grants network" - ) - closure_file = _optional_string_field( - body, "closure", f"capability '{capability}' grants" - ) - unrestricted = _optional_bool_field( - body, "unrestricted", f"capability '{capability}' grants" - ) - _validate_package_bins(capability, package_bins) - _validate_writable_paths(capability, writable) - _validate_allowed_hosts(capability, allowed_hosts) - _validate_closure_file(capability, closure_file) - - return Grant( - package_bins=package_bins, - allowed_hosts=allowed_hosts, - writable=writable, - unrestricted=unrestricted, - closure_file=closure_file, - ) + for key in ("name", "provider", "sampling"): + if key in body: + mapped[key] = body[key] + return mapped -def _validate_closure_file(capability: str, closure_file: object) -> None: - """Validate the closure store-paths reference's shape (contents read later). - Absence is tolerated here (like an empty `packageBins`) so the pure builder - stays testable without Nix; a real Nix manifest always emits it, and the - realized read in `resolve_realized_closures` fails closed if it is missing. - """ - if not closure_file: - return - _validate_store_paths_file(f"capability '{capability}' closure", closure_file) +def _require_object(value: object, label: str) -> dict[str, Any]: + """Return value as a dict, or fail closed with a contextual message. - -def _validate_package_bins(capability: str, package_bins: list[str]) -> None: - for package_bin in package_bins: - if not isinstance(package_bin, str): - raise ManifestError( - f"capability '{capability}' packageBins entries must be strings" - ) - if not package_bin.startswith("/"): - raise ManifestError( - f"capability '{capability}' packageBins entry '{package_bin}' " - "must be an absolute path" - ) - if not package_bin.startswith("/nix/store/"): - raise ManifestError( - f"capability '{capability}' packageBins entry '{package_bin}' " - "must be under /nix/store" - ) - if not package_bin.endswith("/bin"): - raise ManifestError( - f"capability '{capability}' packageBins entry '{package_bin}' " - "must end with /bin" - ) - - -def _validate_allowed_hosts(capability: str, allowed_hosts: list[str]) -> None: - for host in allowed_hosts: - if not isinstance(host, str): - raise ManifestError( - f"capability '{capability}' allowedHosts entries must be strings" - ) + The mapping layer must traverse nested objects (`grants`, `network`, + `params`) to rename their keys, so a non-object there would crash before + Pydantic ever sees it. Guarding here keeps every shape error a ManifestError. + """ + if not isinstance(value, dict): + raise ManifestError(f"{label} must be an object") + return cast(dict[str, Any], value) -def validate_realized_package_bins(manifest: Manifest) -> None: - missing_bins = [ - package_bin - for capability in manifest.capabilities.values() - for package_bin in capability.grants.package_bins - if not os.path.isdir(package_bin) - ] - if missing_bins: - raise ManifestError( - "realized packageBins are missing directories: " + ", ".join(missing_bins) - ) +def _format_validation_error(error: ValidationError) -> str: + """Condense a Pydantic ValidationError into one readable clause per problem. + Pydantic's default string form is a multi-line dump carrying internal type + codes and a docs URL. The manifest is authored in Nix, so surface only the + field location and the message a human needs to fix it. + """ + problems: list[str] = [] + for detail in error.errors(): + location = ".".join(str(part) for part in detail["loc"]) + # Pydantic prefixes every validator ValueError with "Value error, "; drop it. + message = detail["msg"].removeprefix("Value error, ") + if location: + problems.append(f"{location}: {message}") + else: + problems.append(message) + return "; ".join(problems) -def resolve_realized_closures(manifest: Manifest) -> Manifest: - """Read each emitted `store-paths` file into its grant's `closure_paths`. - Runs after the bundle is realized, so the files exist on disk. The jail - binds exactly these paths plus the agent's baseline `shell_closure`, so a - capability reaches its declared closure and nothing else. Fails closed: a - missing or malformed closure file refuses to start (mirrors - `validate_realized_package_bins`). - """ - shell_closure = _read_closure_file("agent shell", manifest.shell_closure_file) - capabilities = { - name: replace( - capability, - grants=replace( - capability.grants, - closure_paths=_read_closure_file( - f"capability '{name}'", capability.grants.closure_file - ), - ), - ) - for name, capability in manifest.capabilities.items() - } - return replace(manifest, capabilities=capabilities, shell_closure=shell_closure) +# ── Closure file I/O ───────────────────────────────────────────────────────── def _read_closure_file(label: str, closure_file: str) -> list[str]: - # An absent reference binds nothing extra (the Phase-0 / no-Nix path). That - # under-permits — a tool whose closure is unbound simply fails to run — so it - # is fail-safe. A *declared* reference that cannot be read is a real error. if not closure_file: return [] try: @@ -559,89 +247,3 @@ def _read_closure_file(label: str, closure_file: str) -> list[str]: f"{label} closure file '{closure_file}' has a non-store path '{path}'" ) return paths - - -def _validate_runner_placeholders( - capability: str, runner: str, params: dict[str, Param] -) -> None: - for _, field_name, _, _ in string.Formatter().parse(runner): - if field_name and field_name not in params: - raise ManifestError( - f"capability '{capability}' runner references undeclared " - f"param '{field_name}'" - ) - - -def _validate_writable_paths(capability: str, writable: list[str]) -> None: - for path in writable: - if not isinstance(path, str): - raise ManifestError( - f"capability '{capability}' writable entries must be strings" - ) - if path.startswith("/"): - raise ManifestError( - f"capability '{capability}' writable path '{path}' must be relative" - ) - if ".." in path.split("/"): - raise ManifestError( - f"capability '{capability}' writable path '{path}' escapes the " - "work tree" - ) - - -def _validate_unrestricted(capability: str, policy: str, grants: Grant) -> None: - if grants.unrestricted and policy == "auto": - raise ManifestError( - f"capability '{capability}' is unrestricted under 'auto' policy; " - "the unrestricted escape must never be silent" - ) - - -def _validate_tools(tools: list, capabilities: dict[str, Capability]) -> None: - for tool in tools: - if not isinstance(tool, dict): - raise ManifestError("tool entries must be objects") - typed_tool = cast(dict[str, Any], tool) - name = typed_tool.get("name") - if not isinstance(name, str): - raise ManifestError("tool 'name' must be a string") - _optional_string_field(typed_tool, "description", f"tool '{name}'") - capability = capabilities.get(name) - if capability is None: - raise ManifestError(f"tool '{name}' has no matching capability") - if capability.policy == "deny": - raise ManifestError( - f"tool '{name}' is exposed but its capability policy is 'deny'" - ) - _validate_tool_consistency(typed_tool, capability) - - -def _validate_tool_consistency(tool: dict[str, Any], capability: Capability) -> None: - schema = _object_field(tool, "parameters", f"tool '{capability.name}'") - properties = _object_field( - schema, "properties", f"tool '{capability.name}' parameters" - ) - required = _list_field(schema, "required", f"tool '{capability.name}' parameters") - for required_name in required: - if not isinstance(required_name, str): - raise ManifestError( - f"tool '{capability.name}' required entries must be strings" - ) - schema_properties = set(properties.keys()) - schema_required = set(required) - - declared = set(capability.params.keys()) - declared_required = { - name for name, param in capability.params.items() if param.required - } - - if schema_properties != declared: - raise ManifestError( - f"tool '{capability.name}' parameters {sorted(schema_properties)} " - f"do not match capability params {sorted(declared)}" - ) - if schema_required != declared_required: - raise ManifestError( - f"tool '{capability.name}' required {sorted(schema_required)} does not " - f"match capability required {sorted(declared_required)}" - ) diff --git a/tests/test_broker.py b/tests/test_broker.py index df6468f..b26922b 100644 --- a/tests/test_broker.py +++ b/tests/test_broker.py @@ -241,7 +241,16 @@ def test_missing_required_argument_never_reaches_the_jail(): def test_deny_capability_never_reaches_the_jail(): jail = FakeJail() - capabilities = {"locked": Capability("locked", "no", "deny", {}, Grant(), "true")} + capabilities = { + "locked": Capability( + name="locked", + description="no", + policy="deny", + params={}, + grants=Grant(), + runner="true", + ) + } broker = Broker(build_manifest(capabilities), jail, PolicyEngine()) result = broker.handle(_call("locked", {})) @@ -256,7 +265,7 @@ def _ask_always_manifest(): name="run_command", description="run", policy="ask-always", - params={"command": Param("string", "", required=True)}, + params={"command": Param(type="string", description="", required=True)}, grants=Grant(), runner="bash -c {command}", ) @@ -268,7 +277,7 @@ def _unrestricted_manifest(): name="shell_escape", description="escape", policy="ask-always", - params={"command": Param("string", "", required=True)}, + params={"command": Param(type="string", description="", required=True)}, grants=Grant(unrestricted=True), runner="bash -c {command}", ) @@ -346,7 +355,7 @@ def _background_manifest(): name="run_bg", description="run detached", policy="ask-always", - params={"command": Param("string", "", required=True)}, + params={"command": Param(type="string", description="", required=True)}, grants=Grant(writable=["."]), runner="bash -c {command}", kind="background", @@ -359,7 +368,7 @@ def _control_manifest(name, op, policy="auto"): name=name, description="control", policy=policy, - params={"task": Param("string", "", required=True)}, + params={"task": Param(type="string", description="", required=True)}, grants=Grant(), runner="", kind="control", @@ -465,8 +474,13 @@ def test_control_output_rejects_non_integer_offset_via_broker_dispatch(): def test_validate_args_enforces_types_and_enums(): params = { - "direction": Param("string", "", required=True, enum=["up", "down"]), - "steps": Param("integer", ""), + "direction": Param( + type="string", + description="", + required=True, + enum=["up", "down"], + ), + "steps": Param(type="integer", description=""), } assert validate_args({"direction": "up"}, params) is None diff --git a/tests/test_bundle.py b/tests/test_bundle.py index ae05e72..56e56bc 100644 --- a/tests/test_bundle.py +++ b/tests/test_bundle.py @@ -118,7 +118,7 @@ def test_load_bundle_rejects_missing_ca_bundle(tmp_path): del manifest["caBundle"] (tmp_path / "manifest.json").write_text(json.dumps(manifest)) - with pytest.raises(BundleError, match="caBundle.*required"): + with pytest.raises(BundleError, match="caBundle"): load_bundle(str(tmp_path)) diff --git a/tests/test_manifest.py b/tests/test_manifest.py index e897c20..6782353 100644 --- a/tests/test_manifest.py +++ b/tests/test_manifest.py @@ -23,9 +23,12 @@ def test_tool_projection_builds_json_schema(): policy="auto", params={ "direction": Param( - "string", "Which way.", required=True, enum=["up", "down"] + type="string", + description="Which way.", + required=True, + enum=["up", "down"], ), - "steps": Param("integer", "How many."), + "steps": Param(type="integer", description="How many."), }, grants=Grant(), runner="demo {direction}", @@ -41,8 +44,22 @@ def test_tool_projection_builds_json_schema(): def test_deny_capabilities_are_not_projected_into_tools(): capabilities = { - "open": Capability("open", "ok", "auto", {}, Grant(), "true"), - "locked": Capability("locked", "no", "deny", {}, Grant(), "true"), + "open": Capability( + name="open", + description="ok", + policy="auto", + params={}, + grants=Grant(), + runner="true", + ), + "locked": Capability( + name="locked", + description="no", + policy="deny", + params={}, + grants=Grant(), + runner="true", + ), } manifest = build_manifest(capabilities) diff --git a/tests/test_manifest_loader.py b/tests/test_manifest_loader.py index 3a6170d..52bc8ae 100644 --- a/tests/test_manifest_loader.py +++ b/tests/test_manifest_loader.py @@ -1,5 +1,3 @@ -from dataclasses import replace - import pytest from tartarus.manifest_loader import ( @@ -118,7 +116,7 @@ def test_missing_ca_bundle_is_rejected(): raw = _valid_raw() del raw["caBundle"] - with pytest.raises(ManifestError, match="caBundle.*required"): + with pytest.raises(ManifestError, match="ca_bundle_file"): build_manifest_from_raw(raw) @@ -167,7 +165,7 @@ def test_non_string_system_prompt_is_rejected(): raw = _valid_raw() raw["systemPrompt"] = ["not", "a", "string"] - with pytest.raises(ManifestError, match="systemPrompt"): + with pytest.raises(ManifestError, match="system_prompt"): build_manifest_from_raw(raw) @@ -289,7 +287,7 @@ def test_invalid_policy_literal_is_rejected(): raw = _valid_raw() raw["capabilities"]["echo"]["policy"] = "sometimes" - with pytest.raises(ManifestError, match="invalid policy"): + with pytest.raises(ManifestError, match="policy"): build_manifest_from_raw(raw) @@ -309,11 +307,19 @@ def test_non_object_network_is_rejected(): build_manifest_from_raw(raw) +def test_non_object_capabilities_is_rejected(): + raw = _valid_raw() + raw["capabilities"] = ["not", "an", "object"] + + with pytest.raises(ManifestError, match="capabilities.*object"): + build_manifest_from_raw(raw) + + def test_non_string_writable_entry_is_rejected(): raw = _valid_raw() raw["capabilities"]["echo"]["grants"]["writable"] = [1] - with pytest.raises(ManifestError, match="writable entries must be strings"): + with pytest.raises(ManifestError, match="writable.*valid string"): build_manifest_from_raw(raw) @@ -321,7 +327,7 @@ def test_non_string_package_bin_entry_is_rejected(): raw = _valid_raw() raw["capabilities"]["echo"]["grants"]["packageBins"] = [1] - with pytest.raises(ManifestError, match="packageBins entries must be strings"): + with pytest.raises(ManifestError, match="package_bins.*valid string"): build_manifest_from_raw(raw) @@ -329,7 +335,7 @@ def test_non_string_capability_description_is_rejected(): raw = _valid_raw() raw["capabilities"]["echo"]["description"] = ["bad"] - with pytest.raises(ManifestError, match="description.*string"): + with pytest.raises(ManifestError, match="description.*valid string"): build_manifest_from_raw(raw) @@ -345,7 +351,7 @@ def test_non_object_param_body_is_rejected(): raw = _valid_raw() raw["capabilities"]["echo"]["params"]["message"] = "bad" - with pytest.raises(ManifestError, match="param 'echo.message' must be an object"): + with pytest.raises(ManifestError, match="params.message"): build_manifest_from_raw(raw) @@ -353,7 +359,7 @@ def test_non_boolean_param_required_is_rejected(): raw = _valid_raw() raw["capabilities"]["echo"]["params"]["message"]["required"] = "yes" - with pytest.raises(ManifestError, match="required.*boolean"): + with pytest.raises(ManifestError, match="required.*valid boolean"): build_manifest_from_raw(raw) @@ -361,7 +367,7 @@ def test_non_string_param_description_is_rejected(): raw = _valid_raw() raw["capabilities"]["echo"]["params"]["message"]["description"] = ["bad"] - with pytest.raises(ManifestError, match="description.*string"): + with pytest.raises(ManifestError, match="description.*valid string"): build_manifest_from_raw(raw) @@ -369,7 +375,7 @@ def test_non_list_param_enum_is_rejected(): raw = _valid_raw() raw["capabilities"]["echo"]["params"]["message"]["enum"] = "red" - with pytest.raises(ManifestError, match="enum must be a list"): + with pytest.raises(ManifestError, match="enum.*valid list"): build_manifest_from_raw(raw) @@ -488,19 +494,24 @@ def test_resolve_realized_closures_reads_store_paths(tmp_path): grant_file = tmp_path / "grant-store-paths" grant_file.write_text("/nix/store/coreutils\n/nix/store/jq\n") - # Inject temp file paths via replace so the read is exercised without forcing + # Inject temp file paths via model_copy so the read is exercised without forcing # the store-path shape rule (covered separately) onto a temp file. base = build_manifest_from_raw(_valid_raw()) echo = base.capabilities["echo"] - manifest = replace( - base, - shell_closure_file=str(shell_file), - capabilities={ - **base.capabilities, - "echo": replace( - echo, grants=replace(echo.grants, closure_file=str(grant_file)) - ), - }, + manifest = base.model_copy( + update={ + "shell_closure_file": str(shell_file), + "capabilities": { + **base.capabilities, + "echo": echo.model_copy( + update={ + "grants": echo.grants.model_copy( + update={"closure_file": str(grant_file)} + ) + } + ), + }, + } ) resolved = resolve_realized_closures(manifest) @@ -526,9 +537,8 @@ def test_resolve_realized_closures_fails_closed_on_missing_file(): def test_resolve_realized_closures_rejects_non_store_contents(tmp_path): bad_file = tmp_path / "store-paths" bad_file.write_text("/nix/store/ok\n/etc/passwd\n") - manifest = replace( - build_manifest_from_raw(_valid_raw()), - shell_closure_file=str(bad_file), + manifest = build_manifest_from_raw(_valid_raw()).model_copy( + update={"shell_closure_file": str(bad_file)} ) with pytest.raises(ManifestError, match="non-store path"): @@ -547,7 +557,7 @@ def test_non_object_tool_entry_is_rejected(): raw = _valid_raw() raw["tools"][0] = "bad" - with pytest.raises(ManifestError, match="tool entries must be objects"): + with pytest.raises(ManifestError, match="tools.0.*valid dictionary"): build_manifest_from_raw(raw) @@ -653,7 +663,7 @@ def test_control_kind_is_accepted_with_empty_runner_and_grants(): def test_invalid_kind_is_rejected(): - with pytest.raises(ManifestError, match="invalid kind"): + with pytest.raises(ManifestError, match="kind"): _build_capability("c", _cap_body(kind="weird")) @@ -663,7 +673,7 @@ def test_control_op_on_non_control_kind_is_rejected(): def test_invalid_control_op_is_rejected(): - with pytest.raises(ManifestError, match="invalid control op"): + with pytest.raises(ManifestError, match="control"): _build_capability( "c", _cap_body(kind="control", control="frobnicate", runner="") ) diff --git a/tests/test_policy.py b/tests/test_policy.py index 688820c..54b1fff 100644 --- a/tests/test_policy.py +++ b/tests/test_policy.py @@ -3,7 +3,14 @@ from tartarus.policy import PolicyEngine def _capability(policy, name="cap"): - return Capability(name, "desc", policy, {}, Grant(), "true") + return Capability( + name=name, + description="desc", + policy=policy, + params={}, + grants=Grant(), + runner="true", + ) def _always_yes(): -- 2.51.2