diff --git a/claude_bottle/manifest.py b/claude_bottle/manifest.py index 91eaa39..f1674ec 100644 --- a/claude_bottle/manifest.py +++ b/claude_bottle/manifest.py @@ -28,18 +28,58 @@ from __future__ import annotations import json import os from pathlib import Path -from typing import Any +from typing import Any, Literal, TypedDict, cast from .log import die -Manifest = dict[str, Any] + +class SshEntry(TypedDict, total=False): + """One entry from bottle.ssh. Host and IdentityFile are required at + runtime (enforced by manifest_validate); the rest configure the + generated ~/.ssh/config block inside the container.""" + + Host: str + IdentityFile: str + Hostname: str + User: str + Port: int | str + KnownHostKey: str + + +class BottleEgress(TypedDict, total=False): + allowlist: list[str] + + +class Bottle(TypedDict, total=False): + """Shared infrastructure referenced by agents. env values are + sentinel-prefixed strings (see env_resolve); runtime defaults to + "runc" when absent.""" + + env: dict[str, str] + ssh: list[SshEntry] + egress: BottleEgress + runtime: Literal["runc", "runsc"] + + +class Agent(TypedDict, total=False): + skills: list[str] + prompt: str + bottle: str + + +class Manifest(TypedDict, total=False): + bottles: dict[str, Bottle] + agents: dict[str, Agent] + + +_SUPPORTED_RUNTIMES: tuple[str, ...] = ("runc", "runsc") def manifest_resolve(cwd: str) -> Manifest: """Look for claude-bottle.json in and in $HOME, deep-merge them (cwd entries override home entries on key conflict for both - bottles and agents). Dies if neither file is found or either is - invalid JSON.""" + bottles and agents), then validate. Dies if neither file is found, + either is invalid JSON, or the merged shape violates the schema.""" cwd_file = Path(cwd) / "claude-bottle.json" home_file = Path(os.environ["HOME"]) / "claude-bottle.json" @@ -49,18 +89,135 @@ def manifest_resolve(cwd: str) -> Manifest: if cwd_doc is None and home_doc is None: die(f"no claude-bottle.json found in {cwd} or {os.environ['HOME']}") - if cwd_doc is None: - return home_doc # type: ignore[return-value] - if home_doc is None: - return cwd_doc - - return { - "bottles": {**(home_doc.get("bottles") or {}), **(cwd_doc.get("bottles") or {})}, - "agents": {**(home_doc.get("agents") or {}), **(cwd_doc.get("agents") or {})}, + h = home_doc if home_doc is not None else cast(Manifest, {}) + c = cwd_doc if cwd_doc is not None else cast(Manifest, {}) + manifest: Manifest = { + "bottles": {**(h.get("bottles") or {}), **(c.get("bottles") or {})}, + "agents": {**(h.get("agents") or {}), **(c.get("agents") or {})}, } + manifest_validate(manifest) + return manifest -def _load_json_or_die(path: Path) -> Manifest: +def manifest_validate(manifest: Manifest) -> None: + """Deep-validate the manifest, dying on the first schema violation. + After this returns, getters can trust that bottles and agents are + dicts, every agent has a 'bottle' field that references a defined + bottle, and optional fields (when present) have correct types.""" + bottles = manifest.get("bottles") or {} + agents = manifest.get("agents") or {} + + for bname, bottle in bottles.items(): + _validate_bottle(bname, bottle) + + for aname, agent in agents.items(): + _validate_agent(aname, agent, bottles) + + +def _validate_bottle(name: str, bottle: Any) -> None: + if not isinstance(bottle, dict): + die(f"bottle '{name}' must be a JSON object (was {_json_type(bottle)})") + bottle = cast(dict[Any, Any], bottle) + + env = bottle.get("env") + if env is not None: + if not isinstance(env, dict): + die(f"bottle '{name}' env must be a JSON object (was {_json_type(env)})") + env = cast(dict[Any, Any], env) + for var, value in env.items(): + if not isinstance(value, str): + die( + f"env entry {var} in bottle '{name}' must be a JSON string " + f"(was {_json_type(value)}). Use \"?\" for prompt-at-runtime." + ) + + ssh = bottle.get("ssh") + if ssh is not None: + if not isinstance(ssh, list): + die(f"bottle '{name}' ssh must be an array (was {_json_type(ssh)})") + ssh = cast(list[Any], ssh) + for i, entry in enumerate(ssh): + if not isinstance(entry, dict): + die( + f"bottle '{name}' ssh[{i}] must be a JSON object " + f"(was {_json_type(entry)})" + ) + entry = cast(dict[Any, Any], entry) + host = entry.get("Host") + if not isinstance(host, str) or not host: + die(f"bottle '{name}' ssh[{i}] missing required string field 'Host'") + ident = entry.get("IdentityFile") + if not isinstance(ident, str) or not ident: + die( + f"bottle '{name}' ssh '{host}' missing required string field " + f"'IdentityFile'" + ) + + egress = bottle.get("egress") + if egress is not None: + if not isinstance(egress, dict): + die(f"bottle '{name}' egress must be a JSON object (was {_json_type(egress)})") + egress = cast(dict[Any, Any], egress) + allowlist = egress.get("allowlist") + if allowlist is not None: + if not isinstance(allowlist, list): + die( + f"bottle '{name}' egress.allowlist must be an array " + f"(was {_json_type(allowlist)})" + ) + allowlist = cast(list[Any], allowlist) + for i, host in enumerate(allowlist): + if not isinstance(host, str): + die( + f"bottle '{name}' egress.allowlist[{i}] must be a string " + f"(was {_json_type(host)})" + ) + + runtime = bottle.get("runtime") + if runtime is not None: + if not isinstance(runtime, str): + die(f"bottle '{name}' runtime must be a string (was {_json_type(runtime)})") + if runtime not in _SUPPORTED_RUNTIMES: + die( + f"bottle '{name}' runtime '{runtime}' is not supported. " + f"Use one of: {', '.join(_SUPPORTED_RUNTIMES)}." + ) + + +def _validate_agent(name: str, agent: Any, bottles: dict[str, Any]) -> None: + if not isinstance(agent, dict): + die(f"agent '{name}' must be a JSON object (was {_json_type(agent)})") + agent = cast(dict[Any, Any], agent) + + skills = agent.get("skills") + if skills is not None: + if not isinstance(skills, list): + die(f"agent '{name}' skills must be an array (was {_json_type(skills)})") + skills = cast(list[Any], skills) + for i, skill in enumerate(skills): + if not isinstance(skill, str): + die(f"agent '{name}' skills[{i}] must be a string (was {_json_type(skill)})") + + prompt = agent.get("prompt") + if prompt is not None and not isinstance(prompt, str): + die(f"agent '{name}' prompt must be a string (was {_json_type(prompt)})") + + bottle = agent.get("bottle") + if not isinstance(bottle, str) or not bottle: + die(f"agent '{name}' must declare a 'bottle' field naming a defined bottle") + if bottle not in bottles: + available = ", ".join(bottles.keys()) or "(none defined)" + die( + f"agent '{name}' references bottle '{bottle}', which is not defined. " + f"Available: {available}" + ) + + +def _load_json_or_die(path: Path) -> dict[Any, Any]: + """Load and shallow-check. Confirms bottles/agents (when present) + are JSON objects so the merge in manifest_resolve cannot TypeError + before manifest_validate runs.""" + doc: Any = None try: with path.open() as f: doc = json.load(f) @@ -68,18 +225,18 @@ def _load_json_or_die(path: Path) -> Manifest: die(f"claude-bottle.json at {path} is not valid JSON") if not isinstance(doc, dict): die(f"claude-bottle.json at {path} must be a JSON object") - return doc + return cast(dict[Any, Any], doc) def manifest_has_agent(manifest: Manifest, name: str) -> bool: - return name in (manifest.get("agents") or {}) + return name in manifest["agents"] def manifest_require_agent(manifest: Manifest, name: str) -> None: """Like has_agent but dies with the available agent names listed.""" if manifest_has_agent(manifest, name): return - available = ", ".join((manifest.get("agents") or {}).keys()) + available = ", ".join(manifest["agents"].keys()) if available: die(f"agent '{name}' not defined in claude-bottle.json. Available: {available}") else: @@ -88,58 +245,38 @@ def manifest_require_agent(manifest: Manifest, name: str) -> None: def manifest_env_names(manifest: Manifest, name: str) -> list[str]: """Names (not values) of bottles[agent.bottle].env, in declaration - order. Empty list if the agent has no bottle or the bottle has no env.""" - agent = (manifest.get("agents") or {}).get(name) or {} - bottle_name = agent.get("bottle") or "" - if not bottle_name: - return [] - bottle = (manifest.get("bottles") or {}).get(bottle_name) or {} - return list((bottle.get("env") or {}).keys()) + order. Empty if the agent's bottle has no env.""" + bottle_name = manifest["agents"][name]["bottle"] + return list(manifest["bottles"][bottle_name].get("env", {}).keys()) def manifest_env_entry(manifest: Manifest, agent: str, var: str) -> str: """Raw string value of one env entry. Used by env_resolve, which - classifies the result by sentinel. Dies if the agent has no bottle, - or the entry is not a string.""" - agent_def = (manifest.get("agents") or {}).get(agent) or {} - bottle_name = agent_def.get("bottle") or "" - if not bottle_name: - die(f"env entry {var} for agent {agent}: agent has no 'bottle' field") - bottle = (manifest.get("bottles") or {}).get(bottle_name) or {} - env = bottle.get("env") or {} - value = env.get(var) - if not isinstance(value, str): - actual = _json_type(value) - die( - f"env entry {var} for agent {agent} must be a JSON string " - f"(was {actual}). Use \"?\" for prompt-at-runtime." - ) - return value + classifies the result by sentinel.""" + bottle_name = manifest["agents"][agent]["bottle"] + return manifest["bottles"][bottle_name]["env"][var] def manifest_skills(manifest: Manifest, name: str) -> list[str]: - agent = (manifest.get("agents") or {}).get(name) or {} - return list(agent.get("skills") or []) + return list(manifest["agents"][name].get("skills", [])) def manifest_prompt(manifest: Manifest, name: str) -> str: - agent = (manifest.get("agents") or {}).get(name) or {} - return agent.get("prompt") or "" + return manifest["agents"][name].get("prompt", "") def manifest_agent_bottle(manifest: Manifest, name: str) -> str: - agent = (manifest.get("agents") or {}).get(name) or {} - return agent.get("bottle") or "" + return manifest["agents"][name]["bottle"] def manifest_has_bottle(manifest: Manifest, bottle_name: str) -> bool: - return bottle_name in (manifest.get("bottles") or {}) + return bottle_name in manifest["bottles"] def manifest_require_bottle(manifest: Manifest, bottle_name: str) -> None: if manifest_has_bottle(manifest, bottle_name): return - available = ", ".join((manifest.get("bottles") or {}).keys()) + available = ", ".join(manifest["bottles"].keys()) if available: die( f"bottle '{bottle_name}' not defined in claude-bottle.json. " @@ -149,56 +286,24 @@ def manifest_require_bottle(manifest: Manifest, bottle_name: str) -> None: die(f"bottle '{bottle_name}' not defined in claude-bottle.json (no bottles defined).") -def manifest_bottle_ssh(manifest: Manifest, bottle_name: str) -> list[dict[str, Any]]: - bottle = (manifest.get("bottles") or {}).get(bottle_name) or {} - return list(bottle.get("ssh") or []) - - -_SUPPORTED_RUNTIMES: tuple[str, ...] = ("runc", "runsc") +def manifest_bottle_ssh(manifest: Manifest, bottle_name: str) -> list[SshEntry]: + return list(manifest["bottles"][bottle_name].get("ssh", [])) def manifest_bottle_runtime(manifest: Manifest, bottle_name: str) -> str: """Container runtime for the bottle's agent container. Returns - "runc" (Docker default) or "runsc" (gVisor opt-in). Dies if the - field is present but not one of the supported values.""" - bottle = (manifest.get("bottles") or {}).get(bottle_name) or {} - raw = bottle.get("runtime") - if raw is None: - return "runc" - if not isinstance(raw, str): - die( - f"bottle '{bottle_name}' runtime must be a string " - f"(was {_json_type(raw)})." - ) - if raw not in _SUPPORTED_RUNTIMES: - die( - f"bottle '{bottle_name}' runtime '{raw}' is not supported. " - f"Use one of: {', '.join(_SUPPORTED_RUNTIMES)}." - ) - return raw + "runc" (Docker default) or "runsc" (gVisor opt-in).""" + return manifest["bottles"][bottle_name].get("runtime", "runc") def manifest_bottle_egress_allowlist(manifest: Manifest, bottle_name: str) -> list[str]: - """Hostnames in bottles[bottle_name].egress.allowlist. Dies if the - field is present but not an array. Per-element string typing is - re-checked at use-time in pipelock.""" - bottle = (manifest.get("bottles") or {}).get(bottle_name) or {} - allowlist = (bottle.get("egress") or {}).get("allowlist") - if allowlist is None: - return [] - if not isinstance(allowlist, list): - die( - f"bottle '{bottle_name}' egress.allowlist must be an array " - f"(was {_json_type(allowlist)})." - ) - return list(allowlist) + """Hostnames in bottles[bottle_name].egress.allowlist.""" + return list(manifest["bottles"][bottle_name].get("egress", {}).get("allowlist", [])) -def manifest_ssh(manifest: Manifest, agent_name: str) -> list[dict[str, Any]]: - """SSH entries resolved via the agent's "bottle" field; empty if no bottle set.""" +def manifest_ssh(manifest: Manifest, agent_name: str) -> list[SshEntry]: + """SSH entries resolved via the agent's "bottle" field.""" bottle_name = manifest_agent_bottle(manifest, agent_name) - if not bottle_name: - return [] return manifest_bottle_ssh(manifest, bottle_name) diff --git a/tests/test_manifest_runtime.py b/tests/test_manifest_runtime.py index 4ba013d..eeec914 100644 --- a/tests/test_manifest_runtime.py +++ b/tests/test_manifest_runtime.py @@ -1,10 +1,11 @@ -"""Unit: manifest_bottle_runtime — defaults to runc, accepts runsc, -rejects unknown values and non-strings.""" +"""Unit: bottle runtime — manifest_bottle_runtime returns the configured +runtime (defaulting to runc); manifest_validate rejects unknown values, +non-strings, and empty strings.""" import unittest from claude_bottle.log import Die -from claude_bottle.manifest import manifest_bottle_runtime +from claude_bottle.manifest import manifest_bottle_runtime, manifest_validate def _bottle(runtime_value: object | None) -> dict: @@ -34,15 +35,15 @@ class TestManifestBottleRuntime(unittest.TestCase): def test_rejects_unknown_runtime(self): with self.assertRaises(Die): - manifest_bottle_runtime(_bottle("kata-runtime"), "dev") + manifest_validate(_bottle("kata-runtime")) def test_rejects_non_string(self): with self.assertRaises(Die): - manifest_bottle_runtime(_bottle(42), "dev") + manifest_validate(_bottle(42)) def test_rejects_empty_string(self): with self.assertRaises(Die): - manifest_bottle_runtime(_bottle(""), "dev") + manifest_validate(_bottle("")) if __name__ == "__main__":