diff --git a/AGENTS.md b/AGENTS.md index 4a8536f..86656fa 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -271,11 +271,21 @@ a service public unless it authenticates or is meant to be open. ## 8. Mesh — multi-node coordination (opt-in) -Disabled by default. Enable via env on `castle-api`: -`CASTLE_API_MQTT_ENABLED=true` (+ `_MQTT_HOST`/`_PORT`), `CASTLE_API_MDNS_ENABLED=true`. -Nodes then advertise/discover over MQTT (Mosquitto, `castle-mqtt`) + mDNS; remote -deployments surface as `manager: none` **reference** kinds. Inspect: -`GET /mesh/status`, `GET /nodes`. Modules: `castle_api.mesh`, `.mqtt_client`, `.mdns`. +Runs on **NATS JetStream** (`castle-nats`, TLS + token). Enable via env on +`castle-api`: `CASTLE_API_NATS_ENABLED=true`, `CASTLE_API_NATS_URL=tls://castle-nats.:4222`, +`CASTLE_API_NATS_TOKEN=${secret:NATS_TOKEN}`. Each node publishes its +(secret-stripped) registry to a JetStream **KV** bucket, renews a **presence** +key, and watches for peers; remote deployments surface as `manager: none` +**reference** kinds. A static **`role`** (`authority`|`follower`, in `castle.yaml`) +gates who may write the shared-config bucket. A consumed cross-node service +(`requires: - ref: X` satisfied by a peer) is routed by the gateway with a +presence-gated circuit-breaker. + +Inspect + drive from the CLI: **`castle mesh status`** / **`castle mesh nodes`** / +**`castle mesh config list|get|set`** (or `GET /mesh/status`, `/nodes`, +`/mesh/config`). Modules: `castle_api.nats_client`, `.mesh`, `.mesh_gateway`, +`.mdns`; secrets via `core` `secret_backends` (file default, OpenBao opt-in). +→ Full history + operations: **`docs/fleet-mesh-plan.md`**. --- diff --git a/castle-api/src/castle_api/nats_client.py b/castle-api/src/castle_api/nats_client.py index b0ccb9e..b45c592 100644 --- a/castle-api/src/castle_api/nats_client.py +++ b/castle-api/src/castle_api/nats_client.py @@ -239,6 +239,15 @@ class CastleNATSClient: raise RuntimeError("config bucket not available") await self._config_kv.put(key, value.encode()) + async def list_shared_config(self) -> list[str]: + """All shared-config keys (empty if none/unavailable).""" + if self._config_kv is None: + return [] + try: + return sorted(await self._config_kv.keys()) + except Exception: + return [] # empty bucket raises NoKeysError in nats-py + async def _config_watch_loop(self) -> None: """Watch shared config; announce changes so followers can reconcile. diff --git a/castle-api/src/castle_api/nodes.py b/castle-api/src/castle_api/nodes.py index 9fd31da..48b6d22 100644 --- a/castle-api/src/castle_api/nodes.py +++ b/castle-api/src/castle_api/nodes.py @@ -3,6 +3,7 @@ from __future__ import annotations from fastapi import APIRouter, HTTPException, Request, status +from pydantic import BaseModel from castle_api.config import get_registry, settings from castle_api.mesh import mesh_state @@ -11,6 +12,42 @@ from castle_api.models import DeploymentSummary, MeshStatus, NodeDetail, NodeSum router = APIRouter(tags=["nodes"]) +class ConfigValue(BaseModel): + value: str + + +@router.get("/mesh/config") +async def list_mesh_config(request: Request) -> dict: + """List shared-config keys + this node's role (only the authority may write).""" + client = getattr(request.app.state, "nats_client", None) + if client is None: + return {"keys": [], "role": get_registry().node.role} + return {"keys": await client.list_shared_config(), "role": client.role} + + +@router.get("/mesh/config/{key:path}") +async def get_mesh_config(key: str, request: Request) -> dict: + """Read a shared-config value.""" + client = getattr(request.app.state, "nats_client", None) + value = await client.get_shared_config(key) if client else None + if value is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, f"config key '{key}' not set") + return {"key": key, "value": value} + + +@router.put("/mesh/config/{key:path}") +async def set_mesh_config(key: str, body: ConfigValue, request: Request) -> dict: + """Write a shared-config value (authority only).""" + client = getattr(request.app.state, "nats_client", None) + if client is None: + raise HTTPException(status.HTTP_503_SERVICE_UNAVAILABLE, "mesh not enabled") + try: + await client.put_shared_config(key, body.value) + except PermissionError as exc: + raise HTTPException(status.HTTP_403_FORBIDDEN, str(exc)) from exc + return {"key": key, "ok": True} + + def _local_node_summary(registry: object) -> NodeSummary: """Build a NodeSummary for the local node from the registry.""" return NodeSummary( diff --git a/castle-api/tests/conftest.py b/castle-api/tests/conftest.py index a5a99a1..e707acc 100644 --- a/castle-api/tests/conftest.py +++ b/castle-api/tests/conftest.py @@ -1,5 +1,9 @@ """Test fixtures for castle-api.""" +import socket +import subprocess +import time +import urllib.request from collections.abc import Generator from pathlib import Path @@ -17,6 +21,48 @@ from castle_core.registry import ( ) +def _free_port() -> int: + s = socket.socket() + s.bind(("", 0)) + port = s.getsockname()[1] + s.close() + return port + + +def _docker_available() -> bool: + try: + return subprocess.run( + ["docker", "info"], capture_output=True, timeout=5 + ).returncode == 0 + except Exception: + return False + + +@pytest.fixture +def nats_url() -> Generator[str, None, None]: + """A throwaway NATS+JetStream broker in docker (fresh per test for clean + buckets). Skips if docker is unavailable.""" + if not _docker_available(): + pytest.skip("docker unavailable — skipping NATS integration tests") + cport, mport = _free_port(), _free_port() + name = f"castle-test-nats-{cport}" + subprocess.run( + ["docker", "run", "-d", "--rm", "--name", name, + "-p", f"{cport}:4222", "-p", f"{mport}:8222", "nats:2", "-js", "-m", "8222"], + check=True, capture_output=True, + ) + try: + for _ in range(50): # wait for readiness + try: + urllib.request.urlopen(f"http://127.0.0.1:{mport}/healthz", timeout=1) + break + except Exception: + time.sleep(0.2) + yield f"nats://127.0.0.1:{cport}" + finally: + subprocess.run(["docker", "rm", "-f", name], capture_output=True) + + def _write_castle_config(root: Path, config: dict) -> None: """Scatter a nested castle config dict into the directory-per-resource layout.""" globals_data = {k: v for k, v in config.items() if k in ("gateway", "repo")} diff --git a/castle-api/tests/test_mesh_config_api.py b/castle-api/tests/test_mesh_config_api.py new file mode 100644 index 0000000..0ac9074 --- /dev/null +++ b/castle-api/tests/test_mesh_config_api.py @@ -0,0 +1,26 @@ +"""HTTP-layer tests for the /mesh/config endpoints (mesh-disabled paths). + +The write/read-through-the-client behavior is covered by test_nats_integration; +here we pin the endpoint wiring when no mesh client is attached. +""" + +from __future__ import annotations + +from fastapi.testclient import TestClient + + +def test_list_config_reports_role_when_mesh_disabled(client: TestClient) -> None: + r = client.get("/mesh/config") + assert r.status_code == 200 + body = r.json() + assert body["keys"] == [] + assert body["role"] == "follower" # test-node has no explicit role + + +def test_get_missing_config_is_404(client: TestClient) -> None: + assert client.get("/mesh/config/does/not/exist").status_code == 404 + + +def test_write_without_mesh_is_503(client: TestClient) -> None: + r = client.put("/mesh/config/fleet/motd", json={"value": "x"}) + assert r.status_code == 503 diff --git a/castle-api/tests/test_nats_integration.py b/castle-api/tests/test_nats_integration.py new file mode 100644 index 0000000..76872dd --- /dev/null +++ b/castle-api/tests/test_nats_integration.py @@ -0,0 +1,117 @@ +"""Integration tests for CastleNATSClient against a real NATS/JetStream broker. + +Exercises the runtime the unit tests can't: connect, KV publish, peer discovery +via watch, presence, graceful-offline, and shared-config. Requires docker (the +`nats_url` fixture); skipped otherwise. +""" + +from __future__ import annotations + +import asyncio + +import pytest +from castle_core.registry import Deployment, NodeConfig, NodeRegistry + +import castle_api.nats_client as ncmod +from castle_api.mesh import mesh_state +from castle_api.nats_client import CastleNATSClient + + +@pytest.fixture(autouse=True) +def _isolate(monkeypatch): + """Stub the gateway-regen side effect (so tests never touch the host's real + Caddyfile) and clear the shared mesh_state singleton around each test.""" + + async def _stub(*_a, **_k): + return False + + monkeypatch.setattr(ncmod, "refresh_remote_routes", _stub) + mesh_state._nodes.clear() + yield + mesh_state._nodes.clear() + + +def _reg(host: str, deployed=None, role: str = "follower") -> NodeRegistry: + return NodeRegistry( + node=NodeConfig(hostname=host, role=role), deployed=deployed or {} + ) + + +def _widget_reg(host: str) -> NodeRegistry: + w = Deployment( + manager="systemd", launcher="python", run_cmd=[], name="widget", + kind="service", port=9099, subdomain="widget", + ) + return _reg(host, {NodeRegistry.key("service", "widget"): w}) + + +def test_publish_peer_discovery_and_offline(nats_url: str) -> None: + async def run() -> None: + a = CastleNATSClient("node-a", _reg("node-a"), servers=nats_url) + b = CastleNATSClient("node-b", _widget_reg("node-b"), servers=nats_url) + await a.start() + await b.start() + await asyncio.sleep(1.0) # let watches propagate + + nodes = mesh_state.all_nodes(include_stale=True) + assert "node-b" in nodes, "node-a should discover node-b via the KV watch" + assert nodes["node-b"].registry.get("service", "widget") is not None + + # Graceful stop deletes b's key -> the DELETE watch marks it offline. + await b.stop() + await asyncio.sleep(1.0) + nb = mesh_state.get_node("node-b") + assert nb is None or not nb.online + + await a.stop() + + asyncio.run(run()) + + +def test_presence_key_written(nats_url: str) -> None: + async def run() -> None: + c = CastleNATSClient("solo", _reg("solo"), servers=nats_url) + await c.start() + keys = await c._presence_kv.keys() + assert "solo" in keys + await c.stop() + + asyncio.run(run()) + + +def test_shared_config_authority_write_read(nats_url: str) -> None: + async def run() -> None: + auth = CastleNATSClient("auth", _reg("auth", role="authority"), servers=nats_url) + await auth.start() + await auth.put_shared_config("fleet/motd", "hello") + assert await auth.get_shared_config("fleet/motd") == "hello" + assert "fleet/motd" in await auth.list_shared_config() + await auth.stop() + + asyncio.run(run()) + + +def test_secrets_never_on_the_wire(nats_url: str) -> None: + """The registry a peer receives must carry no env/run_cmd.""" + async def run() -> None: + secretful = Deployment( + manager="systemd", launcher="python", + run_cmd=["uv", "run", "svc"], env={"API_KEY": "s3cr3t"}, + name="svc", kind="service", port=9001, subdomain="svc", + ) + a = CastleNATSClient("wa", _reg("wa"), servers=nats_url) + b = CastleNATSClient( + "wb", _reg("wb", {NodeRegistry.key("service", "svc"): secretful}), + servers=nats_url, + ) + await a.start() + await b.start() + await asyncio.sleep(1.0) + svc = mesh_state.get_node("wb").registry.get("service", "svc") + assert svc is not None + assert not svc.env, "env must not cross the wire" + assert not svc.run_cmd, "run_cmd must not cross the wire" + await a.stop() + await b.stop() + + asyncio.run(run()) diff --git a/cli/src/castle_cli/commands/mesh.py b/cli/src/castle_cli/commands/mesh.py new file mode 100644 index 0000000..01d64ad --- /dev/null +++ b/cli/src/castle_cli/commands/mesh.py @@ -0,0 +1,105 @@ +"""castle mesh — inspect the NATS mesh and manage shared config. + +The mesh lives in the running castle-api (it holds the live peer state), so this +command talks to the local API over HTTP rather than reading files. +""" + +from __future__ import annotations + +import argparse +import json +import urllib.error +import urllib.request + + +def _api_base() -> str: + port = None + try: + from castle_core.config import load_config + + config = load_config() + dep = next((d for _k, d in config.deployments_named("castle-api")), None) + internal = getattr(getattr(getattr(dep, "expose", None), "http", None), "internal", None) + port = getattr(internal, "port", None) + except Exception: + pass + return f"http://localhost:{port or 9020}" + + +def _get(path: str): + with urllib.request.urlopen(_api_base() + path, timeout=5) as r: # noqa: S310 + return json.load(r) + + +def _put(path: str, body: dict): + req = urllib.request.Request( # noqa: S310 + _api_base() + path, + data=json.dumps(body).encode(), + method="PUT", + headers={"Content-Type": "application/json"}, + ) + with urllib.request.urlopen(req, timeout=5) as r: # noqa: S310 + return json.load(r) + + +def run_mesh(args: argparse.Namespace) -> int: + sub = getattr(args, "mesh_command", None) or "status" + try: + if sub == "status": + return _status() + if sub == "nodes": + return _nodes() + if sub == "config": + return _config(args) + except urllib.error.HTTPError as e: + detail = e.read().decode(errors="replace") + print(f"error: HTTP {e.code} — {detail}") + return 1 + except urllib.error.URLError as e: + print(f"castle-api not reachable ({e.reason}). Is it running + mesh enabled?") + return 1 + print(f"unknown mesh command: {sub}") + return 2 + + +def _status() -> int: + s = _get("/mesh/status") + on = "connected" if s.get("connected") else "disconnected" + print(f"mesh: {'enabled' if s.get('enabled') else 'disabled'} ({on})") + print(f" transport: {s.get('nats_url')}") + print(f" peers ({s.get('peer_count', 0)}): {', '.join(s.get('peers', [])) or '—'}") + return 0 + + +def _nodes() -> int: + nodes = _get("/nodes") + print(f"{'NODE':<14}{'STATUS':<10}{'DEPLOYED':<10}LOCAL") + for n in nodes: + if n.get("online"): + status = "online" + else: + status = "stale" if n.get("is_stale") else "offline" + local = "yes" if n.get("is_local") else "" + print(f"{n['hostname']:<14}{status:<10}{n.get('deployed_count', 0):<10}{local}") + return 0 + + +def _config(args: argparse.Namespace) -> int: + cmd = getattr(args, "mesh_config_command", None) or "list" + if cmd == "list": + data = _get("/mesh/config") + print(f"shared config (this node role: {data.get('role')}):") + for k in data.get("keys", []): + print(f" {k}") + if not data.get("keys"): + print(" (none)") + return 0 + if cmd == "get": + print(_get(f"/mesh/config/{args.key}").get("value", "")) + return 0 + if cmd == "set": + _put(f"/mesh/config/{args.key}", {"value": args.value}) + print(f"set {args.key}") + return 0 + print(f"unknown config command: {cmd}") + return 2 diff --git a/cli/src/castle_cli/main.py b/cli/src/castle_cli/main.py index d656cc2..e1da21e 100644 --- a/cli/src/castle_cli/main.py +++ b/cli/src/castle_cli/main.py @@ -168,6 +168,20 @@ def build_parser() -> argparse.ArgumentParser: gw_sub = gw.add_subparsers(dest="gateway_command") gw_sub.add_parser("status", help="Show gateway status + routes (the default)") + # Mesh — inspect nodes + manage authority-written shared config. + mesh = subparsers.add_parser("mesh", help="Inspect the mesh + shared config") + mesh_sub = mesh.add_subparsers(dest="mesh_command") + mesh_sub.add_parser("status", help="Mesh coordination status (the default)") + mesh_sub.add_parser("nodes", help="List mesh nodes (local + remote)") + mc = mesh_sub.add_parser("config", help="Shared config (only the authority writes)") + mc_sub = mc.add_subparsers(dest="mesh_config_command") + mc_sub.add_parser("list", help="List shared-config keys") + mc_get = mc_sub.add_parser("get", help="Get a shared-config value") + mc_get.add_argument("key") + mc_set = mc_sub.add_parser("set", help="Set a shared-config value (authority only)") + mc_set.add_argument("key") + mc_set.add_argument("value") + # TLS material for raw-TCP services (cert cut from the gateway wildcard). tls = subparsers.add_parser( "tls", help="Manage castle-materialized TLS certs for raw-TCP services" @@ -320,6 +334,10 @@ def main() -> int: from castle_cli.commands.gateway import run_gateway return run_gateway(args) + if cmd == "mesh": + from castle_cli.commands.mesh import run_mesh + + return run_mesh(args) if cmd == "tls": from castle_cli.commands.tls import run_tls