From 38c33224e7effbd5ee9977218ed401caadf61c63 Mon Sep 17 00:00:00 2001 From: Paul Payne Date: Tue, 7 Jul 2026 04:54:50 -0700 Subject: [PATCH 1/7] refactor: remove dead provides/consumes capability model The program-level Capability model (provides/consumes) was a vestige of an earlier design superseded by the deployment-level requires/Requirement graph. It was read only into a diagnostic node and surfaced as an always-empty 'Capabilities' panel; nothing acted on it. Removed from manifest, relations, the CLI re-export, and the System Map. --- app/src/pages/SystemMap.tsx | 24 -- castle-api/src/castle_api/mqtt_client.py | 283 ----------------------- castle-api/tests/test_mqtt.py | 110 --------- cli/src/castle_cli/manifest.py | 1 - core/src/castle_core/manifest.py | 9 - core/src/castle_core/relations.py | 5 - 6 files changed, 432 deletions(-) delete mode 100644 castle-api/src/castle_api/mqtt_client.py delete mode 100644 castle-api/tests/test_mqtt.py diff --git a/app/src/pages/SystemMap.tsx b/app/src/pages/SystemMap.tsx index af38d50..e056356 100644 --- a/app/src/pages/SystemMap.tsx +++ b/app/src/pages/SystemMap.tsx @@ -399,8 +399,6 @@ interface NodeMeta { kind: string remote: boolean node: string | null - provides: string[] - capsConsumes: string[] reach: string | null exposable: boolean launchUrl?: string @@ -882,8 +880,6 @@ export function SystemMapPage() { kind: n.kind, remote: false, node: null, - provides: n.provides ?? [], - capsConsumes: n.consumes ?? [], reach: n.reach, exposable: n.kind === "service" || n.kind === "static", launchUrl: n.kind === "reference" ? (n.base_url ?? undefined) : launchOf(n), @@ -894,8 +890,6 @@ export function SystemMapPage() { kind: ri.md.kind, remote: true, node: ri.machine, - provides: [], - capsConsumes: [], reach: null, exposable: false, launchUrl: remoteLaunchOf(ri.md), @@ -1193,8 +1187,6 @@ export function SystemMapPage() { kind: m.kind, remote: m.remote, node: m.node, - provides: m.provides, - capsConsumes: m.capsConsumes, launchUrl: m.launchUrl, reach: m.reach, exposable: m.exposable, @@ -1422,8 +1414,6 @@ function InspectPanel({ remote: boolean node: string | null launchUrl?: string - provides: string[] - capsConsumes: string[] consumes: Consume[] consumedBy: Dependent[] } @@ -1488,20 +1478,6 @@ function InspectPanel({ )} - {(info.provides.length > 0 || info.capsConsumes.length > 0) && ( - - {info.provides.map((t) => ( - - provides {t} - - ))} - {info.capsConsumes.map((t) => ( - - needs {t} - - ))} - - )} {info.consumes.length === 0 && } {info.consumes.map((c) => ( diff --git a/castle-api/src/castle_api/mqtt_client.py b/castle-api/src/castle_api/mqtt_client.py deleted file mode 100644 index 7a107e5..0000000 --- a/castle-api/src/castle_api/mqtt_client.py +++ /dev/null @@ -1,283 +0,0 @@ -"""MQTT client for inter-node mesh coordination. - -Topics: - castle/{hostname}/registry — retained JSON NodeRegistry, published on connect - castle/{hostname}/status — "online" (retained) / "offline" (LWT) - -On incoming messages from other hostnames, updates MeshStateManager. -""" - -from __future__ import annotations - -import asyncio -import json -import logging - -import paho.mqtt.client as mqtt - -from castle_core.registry import ( - Deployment, - NodeConfig, - NodeRegistry, -) - -from castle_api.mesh import mesh_state -from castle_api.stream import broadcast - -logger = logging.getLogger(__name__) - - -def _registry_to_json(registry: NodeRegistry) -> str: - """Serialize a NodeRegistry to JSON for MQTT publishing. - - Only includes fields needed for mesh routing — env vars, run_cmd, - and castle_root are excluded to avoid leaking secrets. - """ - data: dict = { - "node": { - "hostname": registry.node.hostname, - "gateway_port": registry.node.gateway_port, - # acme domain — lets peers build launch URLs (.) - # for this node's exposed apps. Omitted when the node has no domain. - "gateway_domain": registry.node.gateway_domain, - }, - "deployed": {}, - } - - for _kind, name, comp in registry.all(): - entry: dict = { - "manager": comp.manager, - "launcher": comp.launcher, - "kind": comp.kind, - } - if comp.stack: - entry["stack"] = comp.stack - if comp.description: - entry["description"] = comp.description - if comp.port is not None: - entry["port"] = comp.port - if comp.health_path: - entry["health_path"] = comp.health_path - if comp.subdomain: - entry["subdomain"] = comp.subdomain - if comp.schedule: - entry["schedule"] = comp.schedule - if comp.managed: - entry["managed"] = comp.managed - # Socket surface + external target — so a peer can resolve cross-node - # consumption endpoints (still no secrets: only ports/URLs). - if getattr(comp, "tcp_port", None) is not None: - entry["tcp_port"] = comp.tcp_port - if getattr(comp, "base_url", None): - entry["base_url"] = comp.base_url - # requires — deployment refs (no secrets), so peers can draw cross-node deps. - if getattr(comp, "requires", None): - entry["requires"] = comp.requires - data["deployed"][NodeRegistry.key(comp.kind, name)] = entry - - return json.dumps(data) - - -def _json_to_registry(payload: str) -> NodeRegistry: - """Deserialize a NodeRegistry from MQTT JSON payload.""" - data = json.loads(payload) - node_data = data.get("node", {}) - node = NodeConfig( - hostname=node_data.get("hostname", ""), - castle_root=node_data.get("castle_root"), - gateway_port=node_data.get("gateway_port", 9000), - gateway_domain=node_data.get("gateway_domain"), - ) - deployed: dict[str, Deployment] = {} - for key, comp_data in data.get("deployed", {}).items(): - key_kind, name = key.split("/", 1) if "/" in key else (None, key) - kind = comp_data.get("kind") or key_kind or "service" - deployed[NodeRegistry.key(kind, name)] = Deployment( - manager=comp_data.get("manager", "systemd"), - launcher=comp_data.get("launcher"), - run_cmd=comp_data.get("run_cmd", []), - env=comp_data.get("env", {}), - description=comp_data.get("description"), - name=name, - kind=kind, - stack=comp_data.get("stack"), - port=comp_data.get("port"), - health_path=comp_data.get("health_path"), - subdomain=comp_data.get("subdomain"), - schedule=comp_data.get("schedule"), - managed=comp_data.get("managed", False), - tcp_port=comp_data.get("tcp_port"), - base_url=comp_data.get("base_url"), - requires=comp_data.get("requires", []), - ) - return NodeRegistry(node=node, deployed=deployed) - - -class CastleMQTTClient: - """Async wrapper around paho-mqtt for castle mesh coordination.""" - - def __init__( - self, - local_hostname: str, - local_registry: NodeRegistry, - broker_host: str = "localhost", - broker_port: int = 1883, - loop: asyncio.AbstractEventLoop | None = None, - ) -> None: - self._local_hostname = local_hostname - self._local_registry = local_registry - self._broker_host = broker_host - self._broker_port = broker_port - self._loop = loop or asyncio.get_event_loop() - - self._client = mqtt.Client( - callback_api_version=mqtt.CallbackAPIVersion.VERSION2, - client_id=f"castle-{local_hostname}", - clean_session=True, - ) - - # LWT: if we disconnect unexpectedly, broker publishes "offline" - self._client.will_set( - f"castle/{local_hostname}/status", - payload="offline", - qos=1, - retain=True, - ) - - self._connected = False - self._client.on_connect = self._on_connect - self._client.on_disconnect = self._on_disconnect - self._client.on_message = self._on_message - - @property - def connected(self) -> bool: - return self._connected - - @property - def broker_host(self) -> str: - return self._broker_host - - @property - def broker_port(self) -> int: - return self._broker_port - - def _on_disconnect( - self, - client: mqtt.Client, - userdata: object, - flags: mqtt.DisconnectFlags, - rc: mqtt.ReasonCode, - properties: mqtt.Properties | None = None, - ) -> None: - self._connected = False - logger.info("Disconnected from MQTT broker (rc=%s)", rc) - - def _on_connect( - self, - client: mqtt.Client, - userdata: object, - flags: mqtt.ConnectFlags, - rc: mqtt.ReasonCode, - properties: mqtt.Properties | None = None, - ) -> None: - """Called when connected to broker — publish our state and subscribe.""" - if rc.is_failure: - logger.error("MQTT connect failed: %s", rc) - self._connected = False - return - - self._connected = True - logger.info( - "Connected to MQTT broker at %s:%d", self._broker_host, self._broker_port - ) - - # Publish our status as online (retained) - client.publish( - f"castle/{self._local_hostname}/status", - payload="online", - qos=1, - retain=True, - ) - - # Publish our registry (retained) - self.publish_registry(self._local_registry) - - # Subscribe to all castle nodes - client.subscribe("castle/+/registry", qos=1) - client.subscribe("castle/+/status", qos=1) - - def _on_message( - self, - client: mqtt.Client, - userdata: object, - msg: mqtt.MQTTMessage, - ) -> None: - """Process incoming MQTT messages from other nodes.""" - try: - parts = msg.topic.split("/") - if len(parts) != 3 or parts[0] != "castle": - return - - hostname = parts[1] - msg_type = parts[2] - - # Ignore our own messages - if hostname == self._local_hostname: - return - - payload = msg.payload.decode() - - if msg_type == "registry": - registry = _json_to_registry(payload) - mesh_state.update_node(hostname, registry) - # Notify SSE clients about mesh change - asyncio.run_coroutine_threadsafe( - broadcast("mesh", {"event": "node_updated", "hostname": hostname}), - self._loop, - ) - - elif msg_type == "status": - if payload == "offline": - mesh_state.set_offline(hostname) - asyncio.run_coroutine_threadsafe( - broadcast( - "mesh", {"event": "node_offline", "hostname": hostname} - ), - self._loop, - ) - - except Exception: - logger.exception("Error processing MQTT message on %s", msg.topic) - - def publish_registry(self, registry: NodeRegistry) -> None: - """Publish (or re-publish) our local registry.""" - self._local_registry = registry - self._client.publish( - f"castle/{self._local_hostname}/registry", - payload=_registry_to_json(registry), - qos=1, - retain=True, - ) - - async def start(self) -> None: - """Connect to the broker and start the network loop.""" - self._client.connect_async(self._broker_host, self._broker_port) - self._client.loop_start() - logger.info( - "MQTT client starting (broker=%s:%d)", - self._broker_host, - self._broker_port, - ) - - async def stop(self) -> None: - """Disconnect and stop the network loop.""" - # Publish offline status before disconnecting - self._client.publish( - f"castle/{self._local_hostname}/status", - payload="offline", - qos=1, - retain=True, - ) - self._client.loop_stop() - self._client.disconnect() - logger.info("MQTT client stopped") diff --git a/castle-api/tests/test_mqtt.py b/castle-api/tests/test_mqtt.py deleted file mode 100644 index 541c8c7..0000000 --- a/castle-api/tests/test_mqtt.py +++ /dev/null @@ -1,110 +0,0 @@ -"""Tests for MQTT client serialization logic.""" - -import json - -from castle_core.registry import Deployment, NodeConfig, NodeRegistry - -from castle_api.mqtt_client import _json_to_registry, _registry_to_json - - -def _make_registry() -> NodeRegistry: - return NodeRegistry( - node=NodeConfig( - hostname="tower", castle_root="/data/repos/castle", gateway_port=9000 - ), - deployed={ - "my-svc": Deployment( - manager="systemd", - launcher="python", - run_cmd=["uv", "run", "my-svc"], - env={"PORT": "9001", "SECRET_KEY": "super-secret"}, - description="My service", - name="my-svc", - kind="service", - stack="python-fastapi", - port=9001, - health_path="/health", - subdomain="my-svc", - managed=True, - ), - "my-job": Deployment( - manager="systemd", - launcher="command", - run_cmd=["my-job"], - name="my-job", - kind="job", - stack="python-cli", - schedule="0 2 * * *", - ), - }, - ) - - -class TestRegistrySerialization: - """Round-trip serialization of NodeRegistry to/from JSON.""" - - def test_round_trip(self) -> None: - original = _make_registry() - json_str = _registry_to_json(original) - restored = _json_to_registry(json_str) - - assert restored.node.hostname == "tower" - assert restored.node.gateway_port == 9000 - - def test_deployed_components_preserved(self) -> None: - original = _make_registry() - restored = _json_to_registry(_registry_to_json(original)) - - svc = restored.get("service", "my-svc") - assert svc is not None - assert svc.manager == "systemd" - assert svc.launcher == "python" - assert svc.port == 9001 - assert svc.health_path == "/health" - assert svc.subdomain == "my-svc" - assert svc.managed is True - assert svc.kind == "service" - assert svc.stack == "python-fastapi" - - def test_job_fields_preserved(self) -> None: - original = _make_registry() - restored = _json_to_registry(_registry_to_json(original)) - - job = restored.get("job", "my-job") - assert job is not None - assert job.launcher == "command" - assert job.schedule == "0 2 * * *" - assert job.kind == "job" - assert job.stack == "python-cli" - - def test_optional_fields_omitted(self) -> None: - """Fields like port, health_path are None when not set.""" - reg = NodeRegistry( - node=NodeConfig(hostname="minimal"), - deployed={ - "bare": Deployment( - manager="systemd", launcher="command", run_cmd=["bare"], name="bare" - ), - }, - ) - restored = _json_to_registry(_registry_to_json(reg)) - bare = restored.get("service", "bare") - assert bare.port is None - assert bare.health_path is None - assert bare.subdomain is None - assert bare.schedule is None - assert bare.managed is False - - def test_no_secrets_in_payload(self) -> None: - """env vars, run_cmd, and castle_root must not appear in MQTT payload.""" - original = _make_registry() - json_str = _registry_to_json(original) - data = json.loads(json_str) - - # No castle_root in node - assert "castle_root" not in data["node"] - - # No env or run_cmd in any component - for name, comp in data["deployed"].items(): - assert "env" not in comp, f"{name} has env in MQTT payload" - assert "run_cmd" not in comp, f"{name} has run_cmd in MQTT payload" diff --git a/cli/src/castle_cli/manifest.py b/cli/src/castle_cli/manifest.py index 3b285e0..9b73fc1 100644 --- a/cli/src/castle_cli/manifest.py +++ b/cli/src/castle_cli/manifest.py @@ -4,7 +4,6 @@ from castle_core.manifest import * # noqa: F401, F403 from castle_core.manifest import ( # noqa: F401 — explicit re-exports for type checkers BuildSpec, CaddyDeployment, - Capability, CommandsSpec, DefaultsSpec, DeploymentBase, diff --git a/core/src/castle_core/manifest.py b/core/src/castle_core/manifest.py index 82a4ee5..06adc39 100644 --- a/core/src/castle_core/manifest.py +++ b/core/src/castle_core/manifest.py @@ -277,12 +277,6 @@ class CommandsSpec(BaseModel): # --------------------- -class Capability(BaseModel): - type: str - name: str | None = None - meta: dict[str, str] = Field(default_factory=dict) - - class Requirement(BaseModel): """A precondition — another **deployment** that must exist for this one to be *functional* (``ref`` = the target deployment's name). ``bind`` names the env @@ -390,9 +384,6 @@ class ProgramSpec(BaseModel): version: str | None = None build: BuildSpec | None = None - provides: list[Capability] = Field(default_factory=list) - consumes: list[Capability] = Field(default_factory=list) - tags: list[str] = Field(default_factory=list) @property diff --git a/core/src/castle_core/relations.py b/core/src/castle_core/relations.py index 3dfb929..ac2559b 100644 --- a/core/src/castle_core/relations.py +++ b/core/src/castle_core/relations.py @@ -77,8 +77,6 @@ class Node: reach: str | None = None # off|internal|public (systemd/caddy), else None endpoints: list[Endpoint] = field(default_factory=list) # sockets it exposes base_url: str | None = None # the target URL, for kind=="reference" (external) - provides: list[str] = field(default_factory=list) # capability types (from program) - consumes: list[str] = field(default_factory=list) # capability types (from program) @dataclass @@ -242,7 +240,6 @@ def build_model( ) prog_name = _program_of(name, dep) repo_key = repo_of.get(prog_name) - prog = config.programs.get(prog_name) if prog_name else None nodes.append( Node( name=name, @@ -257,8 +254,6 @@ def build_model( reach=getattr(getattr(dep, "reach", None), "value", None), endpoints=_endpoints_of(dep), base_url=getattr(dep, "base_url", None), - provides=[c.type for c in prog.provides] if prog else [], - consumes=[c.type for c in prog.consumes] if prog else [], ) ) return Model(repos=list(repos.values()), nodes=nodes, edges=edges) From f826d972b865e50c8544275a6de1e98b5fc3a5fd Mon Sep 17 00:00:00 2001 From: Paul Payne Date: Tue, 7 Jul 2026 04:55:02 -0700 Subject: [PATCH 2/7] docs: fleet mesh plan (OpenBao + NATS) Phased plan to evolve the mesh into a purpose-driven fleet: NATS (JetStream KV + presence) replacing Mosquitto, OpenBao as the secret authority, cross-node requires resolution + gateway circuit-breaker, static civil=authority role. Records resolved decisions and Phase 0/1 progress. --- docs/fleet-mesh-plan.md | 215 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 215 insertions(+) create mode 100644 docs/fleet-mesh-plan.md diff --git a/docs/fleet-mesh-plan.md b/docs/fleet-mesh-plan.md new file mode 100644 index 0000000..8af6288 --- /dev/null +++ b/docs/fleet-mesh-plan.md @@ -0,0 +1,215 @@ +# Fleet Mesh Plan — OpenBao + NATS + +**Status:** in progress on branch `feat/fleet-mesh-nats-openbao`. +Phases 0–1 complete + single-node verified (live cutover done). Phases 2–4 pending. + +## Context + +Castle's mesh today is a deliberately minimal, read-only gossip: each node +publishes a secret-stripped `NodeRegistry` to Mosquitto MQTT (retained) + an +online/offline LWT, and peers aggregate it into `MeshStateManager`. It is +LAN-only, unauthenticated, carries no secrets, and does nothing with what it +discovers beyond display. + +The goal is a **purpose-driven heterogeneous fleet** (media server, security +appliance, TV, comms handler, AI gateway) where nodes come and go as *expected*: +shared LAN-wide config + selected secrets, discoverability + presence, and +binding **by purpose** (a consumer needs `media-index`, not a hostname) with +**circuit-breakers** so churn degrades gracefully instead of hanging. + +Two adopted, genuinely-open components (see licensing analysis in chat history): + +- **NATS** (CNCF, Apache-2.0) — replaces Mosquitto as the mesh substrate and does + three jobs in one: pub/sub, **JetStream KV** (shared config + registry), and + **TTL keys as presence** (a provider's key vanishes on death → the breaker + trip signal, no separate health plumbing). +- **OpenBao** (Linux Foundation / OpenSSF, MPL-2.0, Vault fork) — the secret + authority behind castle's existing `${secret:...}` mechanism. + +Design stance: **static single-writer `role` authority, no consensus** — +availability comes from followers running cached local state, not failover. +NATS single-node matches that topology. Consensus stays explicitly out of scope. + +## Architecture — planes → components → code seams + +| Plane | Component | Primary code seam (verified) | +|-------|-----------|------------------------------| +| Transport + KV + presence | NATS (`castle-nats`) | `castle-api/.../mqtt_client.py` (whole file), `main.py` lifespan 62-77, `config.py` Settings | +| Shared config + registry + presence | NATS JetStream KV | `mesh.py` `MeshStateManager`; new KV buckets | +| Secret authority | OpenBao (`castle-openbao`) | `core/.../config.py:307` `_read_secret` (single chokepoint), `castle-api/.../secrets.py` CRUD | +| Binding-by-purpose + breaker | Caddy gateway | `generators/caddyfile.py` `compute_routes` (`remote` kind + ignored `remote_registries` are pre-stubbed), `deploy.py:509-540` `_target_url`/`_requires_env` (local-only today) | +| Authority role | `NodeConfig.role` | `castle_core/registry.py` / `manifest.py` | + +## Phased plan + +Each phase is independently useful; the risky secret-transport work is last. + +### Phase 0 — Stand up the two services (no behavior change) + +- **`castle-nats`**: a `SystemdDeployment` + `RunContainer` (`nats:2` with + JetStream `-js`), data volume `/data/castle/castle-nats`, `reach: internal` + (`nats.`). Follow the container YAML shape in `docs/tcp-exposure.md` + (postgres example). Replace the Mosquitto provisioning in `install.sh` + (`setup_mqtt`, `seed_mosquitto_config`, image/port consts) with NATS — or + better, promote it to a bootstrap deployment YAML (today the broker is only a + shell-provisioned container, not a declared deployment). +- **`castle-openbao`**: `RunContainer` (`openbao/openbao`), data volume, a file + storage backend, `reach: internal`, **auto-unseal** (local transit/key) so the + node boots unattended (Decision 2). + +### Phase 1 — NATS as the mesh transport (swap Mosquitto, behavior-preserving) + +- New `nats_client.py` (async-native `nats-py`) replacing `mqtt_client.py`. + Registry → a **JetStream KV bucket `castle-registry`** keyed by hostname + (KV last-value replaces MQTT retained). Presence → a **`castle-presence`** + bucket with a per-node **TTL key** the node renews (replaces LWT + the 300s + `STALE_TTL` poll; note `prune_stale` is currently uncalled). +- Wire in `main.py` lifespan; rename `mqtt_*` → `nats_*` in `config.py` (keep + `CASTLE_API_` prefix). `MeshStateManager` logic is reused; the KV **watch** + callback drives `update_node`/`set_offline` — and because `nats-py` is + asyncio-native, the `run_coroutine_threadsafe` cross-thread hop for + `broadcast("mesh", …)` **goes away**. +- **Preserve the secret-stripping invariant** from `_registry_to_json` (env / + run_cmd / castle_root never on the wire). +- Rename MQTT-named fields: `models.py` `MeshStatus` (217-226) + frontend + `types/index.ts` (326-330), `MeshPanel.tsx`. Drop `paho-mqtt`; add `nats-py`. + Keep zeroconf peer-advert for LAN discovery **and** support explicit NATS seed + URLs for cross-network (Decision 1); drop the dangling unused `_mqtt._tcp` + browse. Rewrite `test_mqtt.py` → `test_nats.py`. +- Fixes a latent gap: today the registry is published **on-connect only** (no + periodic/on-change republish). KV writes on `castle apply` fix this naturally. + +### Phase 2 — Shared config + presence via JetStream KV + +- Buckets: `castle-config` (shared LAN config, **authority-written only**), + `castle-registry` (per-node), `castle-presence` (TTL liveness). +- Add static **`role: authority | follower`** to `NodeConfig` (castle.yaml). + Pin **`civil` = authority** (Decision 3); only the authority may write + `castle-config`. Authority-down ⇒ shared state read-only, nodes serve cached. +- Followers **watch** `castle-config` and reconcile via `castle apply`. Presence + key renewal is the churn signal for Phase 3. + +### Phase 3 — Cross-node `requires` resolution + gateway binding + breaker (keystone) + +- Extend `_target_url`/`_requires_env` (`deploy.py:509-540`) to **fall through to + the mesh registry** when a `ref` isn't satisfied locally (today it's + local-only). The ref is already host-independent — this removes the + "must be local" restriction, not adds a concept. +- Produce the pre-stubbed **`remote` `GatewayRoute`**; thread `remote_registries` + into `compute_routes` (currently accepted-but-ignored) and into the Caddyfile + write at `deploy.py:157` (currently passed nothing). Binding = a **stable local + subdomain route** the gateway re-targets to the remote node's URL — the program + sees one unchanging URL. +- **Circuit-breaker:** gate the remote route on the `castle-presence` key + (regenerate + `_reload_gateway` when a provider appears/vanishes), backed by + Caddy passive health (`fail_duration` / `lb_try_duration`) on the + `reverse_proxy` line in `_host_matcher_block`. Consumer degraded-mode = + fallback target or clean 503. (Also carry `health_url` into the registry — + it's currently dropped on `manager: none` refs.) + +### Phase 4 — OpenBao as the secret backend (last; needs the security foundation) + +- Introduce a **`SecretBackend` seam** at the single read chokepoint + `_read_secret` (`config.py:307`) + the `castle-api/.../secrets.py` writer: + `FileSecretBackend` (default, unchanged) and `OpenBaoBackend`. `${secret:NAME}` + syntax is untouched; backend selected in castle.yaml or by env. +- Followers auth to OpenBao (token / AppRole); **need-to-know scoping** via + per-deployment policies. Keep the file backend as fallback; migrate secrets in. + +### Cross-cutting prerequisite (gates Phase 4 + any cross-network) + +- **Transport hardening**: NATS TLS + nkey/user auth (replace Mosquitto's + `allow_anonymous`); OpenBao TLS listener. No secret and no cross-network link + moves until this is in. Cross-network overlay (e.g. Nebula) is orthogonal and + layers under Caddy later. + +## Verification (per phase, end-to-end) + +- **P1:** bring up two nodes; confirm registry + presence propagate over NATS and + the mesh view (`/mesh/status`, System Map) is identical to the MQTT behavior. +- **P2:** write a key to `castle-config` on the authority; observe a follower + watch fire and `castle apply` reconcile. +- **P3:** define a service on node A that `requires` a ref provided on node B; + confirm the gateway routes cross-node, then **kill node B** and confirm the + consumer fails fast (curl returns the degraded 503, not a hang) and **recovers** + when B returns. +- **P4:** store a secret in OpenBao; confirm a service renders it via + `EnvironmentFile=` / `--env-file`; then disable the file fallback and confirm + it still resolves. + +## Progress + +### Phase 0 — DONE + verified (2026-07-07) + +Both services stood up **alongside** the live MQTT mesh (no cutover yet): + +- `castle-nats` (`nats:2`, JetStream) — deployment + `~/.castle/deployments/services/castle-nats.yaml`, config + `/data/castle/castle-nats/config/nats-server.conf`. Verified: container active, + `/healthz` ok, JetStream enabled (`/jsz` shows store dir + limits), and a real + **KV round-trip** — created `castle-registry` (put/get) and a TTL + `castle-presence` bucket (the presence primitive). Test buckets cleaned up. +- `castle-openbao` (`openbao/openbao:latest`, v2.5.5) — deployment + `~/.castle/deployments/services/castle-openbao.yaml`, config + `/data/castle/castle-openbao/config/openbao.hcl`. Verified: initialized (1 + share / threshold 1), unsealed, **KV-v2 secret put/get/delete** round-trip. + Unseal key + root token stored as castle secrets `OPENBAO_UNSEAL_KEY` / + `OPENBAO_ROOT_TOKEN`. + +**Consciously deferred (tracked):** +1. **Auto-unseal on boot.** OpenBao is unsealed now but a container restart + re-seals it. The manifest `SystemdSpec` has no `exec_start_post` hook, so + proper boot-unseal wiring lands with Phase 4 (either a small manifest + addition or a companion oneshot). Manual recovery: unseal with + `OPENBAO_UNSEAL_KEY`. +2. **`install.sh` bootstrap parity.** Fresh-install provisioning still seeds + Mosquitto; adding NATS/OpenBao there (and retiring Mosquitto) is bootstrap-only + (no runtime impact) and folds in with the Phase 1 cutover. + +### Phase 1 — DONE + single-node verified; live cutover done (2026-07-07) + +MQTT/Mosquitto transport replaced by NATS JetStream KV. + +- New `castle_api/mesh_wire.py` — transport-agnostic registry (de)serialization + (moved out of `mqtt_client.py`; preserves the secret-stripping invariant). +- New `castle_api/nats_client.py` — `CastleNATSClient`: connects, PUTs its + registry to the `castle-registry` KV bucket, seeds from existing keys, watches + for peer PUT/DELETE, heartbeats a re-PUT (crash liveness via the existing + stale-TTL), and DELETEs its key on graceful stop (immediate peer-offline). + Async-native → the paho cross-thread `run_coroutine_threadsafe` hop is gone. +- `main.py` lifespan, `config.py` (`nats_enabled`/`nats_url`), `models.py` + `MeshStatus` (`connected`/`nats_url`), `nodes.py` `/mesh/status` all rewired. +- Frontend: `types/index.ts` `MeshStatus` + `MeshPanel.tsx` updated to the new + fields. `mqtt_client.py` + `test_mqtt.py` deleted; `test_mesh_wire.py` added. + `mdns.py` dead `_mqtt._tcp` broker-browse removed. `paho-mqtt` → `nats-py`. +- **Live cutover:** `castle-api.yaml` env → `CASTLE_API_NATS_*`, `requires: + castle-nats`; applied. castle-api healthy on NATS, `/mesh/status` connected, + registry (9.7 KiB, 48 deployments) published to KV, **no secrets on the wire**. +- Verification run: ruff clean; 89 api + 196 core tests pass; frontend `tsc` + clean; runtime KV round-trip confirmed against the live `castle-nats`. + +**Pending second node:** two-node mesh-view parity (peer sees peer, offline on +departure) — needs the second LAN node running the NATS-enabled castle-api +pointed at civil's NATS (seed URL) or a clustered `castle-nats`. Revert path if +needed: restore `CASTLE_API_MQTT_*` env — but note `mqtt_client.py` is deleted, +so revert = `git checkout main -- castle-api` then re-apply. The old Mosquitto +`mqtt` service is left running (dormant) as a safety net; retire it once the +second node is proven on NATS. + +## Decisions (resolved) + +1. **Discovery — both.** Keep mDNS for zero-config LAN peer discovery *and* + support explicit NATS seed URLs so the mesh can span networks. mDNS is the + convenience path on-LAN; seed URLs are the cross-network path. Client connects + via seeds when present, falls back to mDNS-discovered peers on-LAN. +2. **OpenBao unseal — auto for now.** Auto-unseal (local transit/key) so a home + node boots unattended; designed so we can migrate to manual/stricter unseal + later (a supported rekey/unseal-migration, not zero-effort but planned for). + Keep the backend seam clean so the switch is config, not code. +3. **Authority = `civil`.** `civil` (the acme node, `civil.payne.io`) is pinned + as `role: authority`; all others are followers. **Confirmed:** when `civil` is + down, shared config/secrets go **read-only** fleet-wide and every node keeps + serving its own deployments from cached local state. +4. **NATS — single server now, cluster-ready.** Run one `castle-nats` today + (matches the single-writer authority). Clustering is deferred but must be a + pure config addition (cluster/routes block + a few more nodes) — no re-architecture. From 6c00a9d33c32877ad31fb978ecd6e68a54f7c05b Mon Sep 17 00:00:00 2001 From: Paul Payne Date: Tue, 7 Jul 2026 04:55:02 -0700 Subject: [PATCH 3/7] feat(mesh): NATS JetStream transport replacing MQTT MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the Mosquitto/paho mesh transport with NATS JetStream KV: - mesh_wire.py: transport-agnostic registry (de)serialization (secret-stripped) - nats_client.py: async CastleNATSClient — registry in the castle-registry KV bucket, heartbeat liveness, delete-on-stop offline, watch-driven mesh state - rewire main/config/models/nodes; MeshStatus -> connected/nats_url - frontend MeshStatus + MeshPanel to the new fields - drop paho-mqtt/mqtt_client + _mqtt._tcp mDNS browse; add nats-py - keep the secret-stripping invariant; mDNS peer discovery retained Single-node verified; two-node parity pending the second node. --- app/src/components/MeshPanel.tsx | 12 +- app/src/types/index.ts | 5 +- castle-api/pyproject.toml | 2 +- castle-api/src/castle_api/config.py | 5 +- castle-api/src/castle_api/main.py | 23 ++- castle-api/src/castle_api/mdns.py | 36 +---- castle-api/src/castle_api/mesh_wire.py | 99 ++++++++++++ castle-api/src/castle_api/models.py | 5 +- castle-api/src/castle_api/nats_client.py | 184 +++++++++++++++++++++++ castle-api/src/castle_api/nodes.py | 9 +- castle-api/tests/test_mesh_wire.py | 110 ++++++++++++++ uv.lock | 22 +-- 12 files changed, 436 insertions(+), 76 deletions(-) create mode 100644 castle-api/src/castle_api/mesh_wire.py create mode 100644 castle-api/src/castle_api/nats_client.py create mode 100644 castle-api/tests/test_mesh_wire.py diff --git a/app/src/components/MeshPanel.tsx b/app/src/components/MeshPanel.tsx index 516d09e..cab9885 100644 --- a/app/src/components/MeshPanel.tsx +++ b/app/src/components/MeshPanel.tsx @@ -19,25 +19,23 @@ export function MeshPanel({ mesh }: MeshPanelProps) { - {mesh.mqtt_connected ? ( + {mesh.connected ? ( ) : ( )} - {mesh.mqtt_connected ? "connected" : "disconnected"} + {mesh.connected ? "connected" : "disconnected"}
- {mesh.mqtt_broker_host && ( - - mqtt://{mesh.mqtt_broker_host}:{mesh.mqtt_broker_port} - + {mesh.nats_url && ( + {mesh.nats_url} )} {mesh.mdns_enabled && ( mDNS active diff --git a/app/src/types/index.ts b/app/src/types/index.ts index 8b6fc4a..915d51d 100644 --- a/app/src/types/index.ts +++ b/app/src/types/index.ts @@ -325,9 +325,8 @@ export interface NodeDetail extends NodeSummary { export interface MeshStatus { enabled: boolean - mqtt_connected: boolean - mqtt_broker_host: string | null - mqtt_broker_port: number | null + connected: boolean + nats_url: string | null mdns_enabled: boolean peer_count: number peers: string[] diff --git a/castle-api/pyproject.toml b/castle-api/pyproject.toml index aff58e3..4d3591e 100644 --- a/castle-api/pyproject.toml +++ b/castle-api/pyproject.toml @@ -9,7 +9,7 @@ dependencies = [ "pydantic-settings>=2.0.0", "httpx>=0.27.0", "castle-core", - "paho-mqtt>=2.0.0", + "nats-py>=2.9.0", "zeroconf>=0.131.0", ] diff --git a/castle-api/src/castle_api/config.py b/castle-api/src/castle_api/config.py index d641ad2..e85a8b0 100644 --- a/castle-api/src/castle_api/config.py +++ b/castle-api/src/castle_api/config.py @@ -15,9 +15,8 @@ class Settings(BaseSettings): port: int = 9020 # Mesh coordination (all off by default — single-node works without them) - mqtt_enabled: bool = False - mqtt_host: str = "localhost" - mqtt_port: int = 1883 + nats_enabled: bool = False + nats_url: str = "nats://localhost:4222" mdns_enabled: bool = False model_config = { diff --git a/castle-api/src/castle_api/main.py b/castle-api/src/castle_api/main.py index 7a71039..f291f39 100644 --- a/castle-api/src/castle_api/main.py +++ b/castle-api/src/castle_api/main.py @@ -56,25 +56,24 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: poll_task = asyncio.create_task(health_poll_loop()) # --- Mesh coordination (opt-in) --- - mqtt_client = None + nats_client = None mdns_service = None - if settings.mqtt_enabled: + if settings.nats_enabled: try: - from castle_api.mqtt_client import CastleMQTTClient + from castle_api.nats_client import CastleNATSClient registry = get_registry() - mqtt_client = CastleMQTTClient( + nats_client = CastleNATSClient( local_hostname=registry.node.hostname, local_registry=registry, - broker_host=settings.mqtt_host, - broker_port=settings.mqtt_port, + servers=settings.nats_url, ) - await mqtt_client.start() - app.state.mqtt_client = mqtt_client + await nats_client.start() + app.state.nats_client = nats_client except Exception: - logger.exception("Failed to start MQTT client") - mqtt_client = None + logger.exception("Failed to start NATS mesh client") + nats_client = None if settings.mdns_enabled: try: @@ -97,8 +96,8 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: _shutting_down = True poll_task.cancel() - if mqtt_client: - await mqtt_client.stop() + if nats_client: + await nats_client.stop() if mdns_service: mdns_service.stop() diff --git a/castle-api/src/castle_api/mdns.py b/castle-api/src/castle_api/mdns.py index 6205315..844cbaa 100644 --- a/castle-api/src/castle_api/mdns.py +++ b/castle-api/src/castle_api/mdns.py @@ -1,8 +1,7 @@ -"""mDNS discovery for castle mesh — zero-config LAN peer and broker discovery. +"""mDNS discovery for castle mesh — zero-config LAN peer discovery. -Advertises this node as _castle._tcp and browses for: - - _castle._tcp — peer castle nodes - - _mqtt._tcp — MQTT broker address +Advertises this node as _castle._tcp and browses for _castle._tcp peers. This is +the on-LAN convenience path; cross-network meshes use explicit NATS seed URLs. """ from __future__ import annotations @@ -15,11 +14,10 @@ from zeroconf import ServiceBrowser, ServiceInfo, ServiceStateChange, Zeroconf logger = logging.getLogger(__name__) CASTLE_SERVICE_TYPE = "_castle._tcp.local." -MQTT_SERVICE_TYPE = "_mqtt._tcp.local." class CastleMDNS: - """Advertise this castle node and discover peers + MQTT broker via mDNS.""" + """Advertise this castle node and discover peers via mDNS.""" def __init__( self, @@ -38,7 +36,6 @@ class CastleMDNS: self.peers: dict[ str, dict ] = {} # hostname -> {gateway_port, api_port, addresses} - self.mqtt_broker: dict | None = None # {host, port} or None def _on_service_state_change( self, @@ -55,8 +52,6 @@ class CastleMDNS: if service_type == CASTLE_SERVICE_TYPE: self._handle_castle_peer(info) - elif service_type == MQTT_SERVICE_TYPE: - self._handle_mqtt_broker(info) elif state_change == ServiceStateChange.Removed: if service_type == CASTLE_SERVICE_TYPE: @@ -89,20 +84,6 @@ class CastleMDNS: } logger.info("mDNS: discovered peer %s at %s", peer_hostname, addresses) - def _handle_mqtt_broker(self, info: ServiceInfo) -> None: - """Process a discovered MQTT broker.""" - addresses = [ - socket.inet_ntoa(addr) for addr in info.addresses if len(addr) == 4 - ] - if addresses: - self.mqtt_broker = { - "host": addresses[0], - "port": info.port, - } - logger.info( - "mDNS: discovered MQTT broker at %s:%d", addresses[0], info.port - ) - def start(self) -> None: """Start advertising and browsing.""" self._zeroconf = Zeroconf() @@ -123,7 +104,7 @@ class CastleMDNS: "mDNS: advertising %s on port %d", self._hostname, self._gateway_port ) - # Browse for peers and MQTT broker + # Browse for peer castle nodes self._browsers.append( ServiceBrowser( self._zeroconf, @@ -131,13 +112,6 @@ class CastleMDNS: handlers=[self._on_service_state_change], ) ) - self._browsers.append( - ServiceBrowser( - self._zeroconf, - MQTT_SERVICE_TYPE, - handlers=[self._on_service_state_change], - ) - ) def stop(self) -> None: """Stop advertising and close.""" diff --git a/castle-api/src/castle_api/mesh_wire.py b/castle-api/src/castle_api/mesh_wire.py new file mode 100644 index 0000000..9f66509 --- /dev/null +++ b/castle-api/src/castle_api/mesh_wire.py @@ -0,0 +1,99 @@ +"""Mesh wire format — (de)serialize a NodeRegistry for cross-node transport. + +Transport-agnostic (no MQTT/NATS imports). Only the fields needed for mesh +routing are included; env vars, run_cmd, and castle_root are **excluded** to +avoid leaking secrets — this invariant is load-bearing and must be preserved by +any transport that carries this payload. +""" + +from __future__ import annotations + +import json + +from castle_core.registry import ( + Deployment, + NodeConfig, + NodeRegistry, +) + + +def registry_to_json(registry: NodeRegistry) -> str: + """Serialize a NodeRegistry to JSON (secret-stripped).""" + data: dict = { + "node": { + "hostname": registry.node.hostname, + "gateway_port": registry.node.gateway_port, + # acme domain — lets peers build launch URLs (.) + # for this node's exposed apps. Omitted when the node has no domain. + "gateway_domain": registry.node.gateway_domain, + }, + "deployed": {}, + } + + for _kind, name, comp in registry.all(): + entry: dict = { + "manager": comp.manager, + "launcher": comp.launcher, + "kind": comp.kind, + } + if comp.stack: + entry["stack"] = comp.stack + if comp.description: + entry["description"] = comp.description + if comp.port is not None: + entry["port"] = comp.port + if comp.health_path: + entry["health_path"] = comp.health_path + if comp.subdomain: + entry["subdomain"] = comp.subdomain + if comp.schedule: + entry["schedule"] = comp.schedule + if comp.managed: + entry["managed"] = comp.managed + # Socket surface + external target — so a peer can resolve cross-node + # consumption endpoints (still no secrets: only ports/URLs). + if getattr(comp, "tcp_port", None) is not None: + entry["tcp_port"] = comp.tcp_port + if getattr(comp, "base_url", None): + entry["base_url"] = comp.base_url + # requires — deployment refs (no secrets), so peers can draw cross-node deps. + if getattr(comp, "requires", None): + entry["requires"] = comp.requires + data["deployed"][NodeRegistry.key(comp.kind, name)] = entry + + return json.dumps(data) + + +def json_to_registry(payload: str) -> NodeRegistry: + """Deserialize a NodeRegistry from a JSON payload.""" + data = json.loads(payload) + node_data = data.get("node", {}) + node = NodeConfig( + hostname=node_data.get("hostname", ""), + castle_root=node_data.get("castle_root"), + gateway_port=node_data.get("gateway_port", 9000), + gateway_domain=node_data.get("gateway_domain"), + ) + deployed: dict[str, Deployment] = {} + for key, comp_data in data.get("deployed", {}).items(): + key_kind, name = key.split("/", 1) if "/" in key else (None, key) + kind = comp_data.get("kind") or key_kind or "service" + deployed[NodeRegistry.key(kind, name)] = Deployment( + manager=comp_data.get("manager", "systemd"), + launcher=comp_data.get("launcher"), + run_cmd=comp_data.get("run_cmd", []), + env=comp_data.get("env", {}), + description=comp_data.get("description"), + name=name, + kind=kind, + stack=comp_data.get("stack"), + port=comp_data.get("port"), + health_path=comp_data.get("health_path"), + subdomain=comp_data.get("subdomain"), + schedule=comp_data.get("schedule"), + managed=comp_data.get("managed", False), + tcp_port=comp_data.get("tcp_port"), + base_url=comp_data.get("base_url"), + requires=comp_data.get("requires", []), + ) + return NodeRegistry(node=node, deployed=deployed) diff --git a/castle-api/src/castle_api/models.py b/castle-api/src/castle_api/models.py index d2459c3..d7fce04 100644 --- a/castle-api/src/castle_api/models.py +++ b/castle-api/src/castle_api/models.py @@ -218,9 +218,8 @@ class MeshStatus(BaseModel): """Current state of the mesh coordination layer.""" enabled: bool = False - mqtt_connected: bool = False - mqtt_broker_host: str | None = None - mqtt_broker_port: int | None = None + connected: bool = False + nats_url: str | None = None mdns_enabled: bool = False peer_count: int = 0 peers: list[str] = [] diff --git a/castle-api/src/castle_api/nats_client.py b/castle-api/src/castle_api/nats_client.py new file mode 100644 index 0000000..0194058 --- /dev/null +++ b/castle-api/src/castle_api/nats_client.py @@ -0,0 +1,184 @@ +"""NATS JetStream client for inter-node mesh coordination. + +Replaces the MQTT transport. State lives in a JetStream **KV bucket** rather than +retained MQTT messages: + + castle-registry — key=, value=secret-stripped NodeRegistry JSON + +Lifecycle: + * on connect: ensure the bucket, PUT our registry, seed local state from every + existing key, then watch the bucket for peer changes. + * heartbeat: re-PUT our registry every HEARTBEAT_SEC so peers refresh their + last-seen clock (crash liveness rides the existing stale-TTL). + * graceful stop: DELETE our key → peers get an immediate offline signal. + +Being asyncio-native, the watch callback calls ``broadcast`` directly — no +cross-thread ``run_coroutine_threadsafe`` hop (which the paho client needed). +""" + +from __future__ import annotations + +import asyncio +import contextlib +import logging + +import nats +from nats.js.api import KeyValueConfig + +from castle_core.registry import NodeRegistry + +from castle_api.mesh import mesh_state +from castle_api.mesh_wire import json_to_registry, registry_to_json +from castle_api.stream import broadcast + +logger = logging.getLogger(__name__) + +REGISTRY_BUCKET = "castle-registry" +HEARTBEAT_SEC = 30.0 +PRUNE_SEC = 30.0 + + +class CastleNATSClient: + """Async NATS/JetStream mesh client.""" + + def __init__( + self, + local_hostname: str, + local_registry: NodeRegistry, + servers: str | list[str] = "nats://localhost:4222", + ) -> None: + self._local_hostname = local_hostname + self._local_registry = local_registry + self._servers = servers + self._nc: nats.NATS | None = None + self._kv = None + self._tasks: list[asyncio.Task] = [] + self._last_json: dict[str, str] = {} + self._online: set[str] = set() + + @property + def connected(self) -> bool: + return self._nc is not None and self._nc.is_connected + + @property + def servers(self) -> str | list[str]: + return self._servers + + async def start(self) -> None: + """Connect, publish our registry, seed state, and start watchers.""" + self._nc = await nats.connect( + self._servers, + name=f"castle-{self._local_hostname}", + max_reconnect_attempts=-1, # reconnect forever — nodes come and go + ) + js = self._nc.jetstream() + try: + self._kv = await js.key_value(REGISTRY_BUCKET) + except Exception: + self._kv = await js.create_key_value( + config=KeyValueConfig(bucket=REGISTRY_BUCKET, history=1) + ) + + await self.publish_registry(self._local_registry) + await self._seed_existing() + + self._tasks = [ + asyncio.create_task(self._watch_loop()), + asyncio.create_task(self._heartbeat_loop()), + asyncio.create_task(self._prune_loop()), + ] + logger.info("NATS mesh client started (servers=%s)", self._servers) + + async def stop(self) -> None: + """Delete our key (immediate offline to peers) and disconnect.""" + for t in self._tasks: + t.cancel() + for t in self._tasks: + with contextlib.suppress(asyncio.CancelledError, Exception): + await t + self._tasks = [] + if self._kv is not None: + with contextlib.suppress(Exception): + await self._kv.delete(self._local_hostname) + if self._nc is not None: + with contextlib.suppress(Exception): + await self._nc.drain() + self._nc = None + logger.info("NATS mesh client stopped") + + async def publish_registry(self, registry: NodeRegistry) -> None: + """PUT (or refresh) our local registry into the KV bucket.""" + self._local_registry = registry + if self._kv is None: + return + await self._kv.put( + self._local_hostname, registry_to_json(registry).encode() + ) + + async def _seed_existing(self) -> None: + """Load every peer key already present in the bucket.""" + if self._kv is None: + return + try: + keys = await self._kv.keys() + except Exception: + keys = [] # empty bucket raises NoKeysError in nats-py + for key in keys: + if key == self._local_hostname: + continue + with contextlib.suppress(Exception): + entry = await self._kv.get(key) + if entry.value: + self._apply_put(key, entry.value.decode()) + + async def _watch_loop(self) -> None: + assert self._kv is not None + watcher = await self._kv.watchall() + async for entry in watcher: + if entry is None: # "caught up with current values" sentinel + continue + key = entry.key + if key == self._local_hostname: + continue + try: + if entry.operation in ("DEL", "PURGE"): + await self._apply_delete(key) + elif entry.value: + changed = self._apply_put(key, entry.value.decode()) + if changed: + await broadcast( + "mesh", {"event": "node_updated", "hostname": key} + ) + except Exception: + logger.exception("Error handling mesh entry for %s", key) + + def _apply_put(self, hostname: str, payload: str) -> bool: + """Update mesh state from a peer PUT. Returns True if content changed.""" + registry = json_to_registry(payload) + mesh_state.update_node(hostname, registry) # always refresh last-seen + self._online.add(hostname) + changed = self._last_json.get(hostname) != payload + self._last_json[hostname] = payload + return changed + + async def _apply_delete(self, hostname: str) -> None: + mesh_state.set_offline(hostname) + self._online.discard(hostname) + self._last_json.pop(hostname, None) + await broadcast("mesh", {"event": "node_offline", "hostname": hostname}) + + async def _heartbeat_loop(self) -> None: + while True: + await asyncio.sleep(HEARTBEAT_SEC) + with contextlib.suppress(Exception): + await self.publish_registry(self._local_registry) + + async def _prune_loop(self) -> None: + """Mark crashed peers (no refresh within the stale TTL) offline.""" + while True: + await asyncio.sleep(PRUNE_SEC) + all_nodes = mesh_state.all_nodes(include_stale=True) + for host in list(self._online): + node = all_nodes.get(host) + if node is None or node.is_stale: + await self._apply_delete(host) diff --git a/castle-api/src/castle_api/nodes.py b/castle-api/src/castle_api/nodes.py index cd5851c..9fd31da 100644 --- a/castle-api/src/castle_api/nodes.py +++ b/castle-api/src/castle_api/nodes.py @@ -66,15 +66,14 @@ def _deployed_to_summaries(registry: object, hostname: str) -> list[DeploymentSu @router.get("/mesh/status", response_model=MeshStatus) def get_mesh_status(request: Request) -> MeshStatus: """Get the current state of the mesh coordination layer.""" - mqtt_client = getattr(request.app.state, "mqtt_client", None) + nats_client = getattr(request.app.state, "nats_client", None) peers = list(mesh_state.all_nodes(include_stale=True).keys()) return MeshStatus( - enabled=settings.mqtt_enabled, - mqtt_connected=mqtt_client.connected if mqtt_client else False, - mqtt_broker_host=mqtt_client.broker_host if mqtt_client else None, - mqtt_broker_port=mqtt_client.broker_port if mqtt_client else None, + enabled=settings.nats_enabled, + connected=nats_client.connected if nats_client else False, + nats_url=str(nats_client.servers) if nats_client else None, mdns_enabled=settings.mdns_enabled, peer_count=len(peers), peers=peers, diff --git a/castle-api/tests/test_mesh_wire.py b/castle-api/tests/test_mesh_wire.py new file mode 100644 index 0000000..1467b69 --- /dev/null +++ b/castle-api/tests/test_mesh_wire.py @@ -0,0 +1,110 @@ +"""Tests for the transport-agnostic mesh wire format (registry (de)serialization).""" + +import json + +from castle_core.registry import Deployment, NodeConfig, NodeRegistry + +from castle_api.mesh_wire import json_to_registry, registry_to_json + + +def _make_registry() -> NodeRegistry: + return NodeRegistry( + node=NodeConfig( + hostname="tower", castle_root="/data/repos/castle", gateway_port=9000 + ), + deployed={ + "my-svc": Deployment( + manager="systemd", + launcher="python", + run_cmd=["uv", "run", "my-svc"], + env={"PORT": "9001", "SECRET_KEY": "super-secret"}, + description="My service", + name="my-svc", + kind="service", + stack="python-fastapi", + port=9001, + health_path="/health", + subdomain="my-svc", + managed=True, + ), + "my-job": Deployment( + manager="systemd", + launcher="command", + run_cmd=["my-job"], + name="my-job", + kind="job", + stack="python-cli", + schedule="0 2 * * *", + ), + }, + ) + + +class TestRegistrySerialization: + """Round-trip serialization of NodeRegistry to/from JSON.""" + + def test_round_trip(self) -> None: + original = _make_registry() + json_str = registry_to_json(original) + restored = json_to_registry(json_str) + + assert restored.node.hostname == "tower" + assert restored.node.gateway_port == 9000 + + def test_deployed_components_preserved(self) -> None: + original = _make_registry() + restored = json_to_registry(registry_to_json(original)) + + svc = restored.get("service", "my-svc") + assert svc is not None + assert svc.manager == "systemd" + assert svc.launcher == "python" + assert svc.port == 9001 + assert svc.health_path == "/health" + assert svc.subdomain == "my-svc" + assert svc.managed is True + assert svc.kind == "service" + assert svc.stack == "python-fastapi" + + def test_job_fields_preserved(self) -> None: + original = _make_registry() + restored = json_to_registry(registry_to_json(original)) + + job = restored.get("job", "my-job") + assert job is not None + assert job.launcher == "command" + assert job.schedule == "0 2 * * *" + assert job.kind == "job" + assert job.stack == "python-cli" + + def test_optional_fields_omitted(self) -> None: + """Fields like port, health_path are None when not set.""" + reg = NodeRegistry( + node=NodeConfig(hostname="minimal"), + deployed={ + "bare": Deployment( + manager="systemd", launcher="command", run_cmd=["bare"], name="bare" + ), + }, + ) + restored = json_to_registry(registry_to_json(reg)) + bare = restored.get("service", "bare") + assert bare.port is None + assert bare.health_path is None + assert bare.subdomain is None + assert bare.schedule is None + assert bare.managed is False + + def test_no_secrets_in_payload(self) -> None: + """env vars, run_cmd, and castle_root must never appear on the wire.""" + original = _make_registry() + json_str = registry_to_json(original) + data = json.loads(json_str) + + # No castle_root in node + assert "castle_root" not in data["node"] + + # No env or run_cmd in any component + for name, comp in data["deployed"].items(): + assert "env" not in comp, f"{name} has env in payload" + assert "run_cmd" not in comp, f"{name} has run_cmd in payload" diff --git a/uv.lock b/uv.lock index fb1eaf0..ecaf328 100644 --- a/uv.lock +++ b/uv.lock @@ -48,7 +48,7 @@ dependencies = [ { name = "castle-core" }, { name = "fastapi" }, { name = "httpx" }, - { name = "paho-mqtt" }, + { name = "nats-py" }, { name = "pydantic-settings" }, { name = "uvicorn", extra = ["standard"] }, { name = "zeroconf" }, @@ -67,7 +67,7 @@ requires-dist = [ { name = "castle-core", editable = "core" }, { name = "fastapi", specifier = ">=0.115.0" }, { name = "httpx", specifier = ">=0.27.0" }, - { name = "paho-mqtt", specifier = ">=2.0.0" }, + { name = "nats-py", specifier = ">=2.9.0" }, { name = "pydantic-settings", specifier = ">=2.0.0" }, { name = "uvicorn", extras = ["standard"], specifier = ">=0.34.0" }, { name = "zeroconf", specifier = ">=0.131.0" }, @@ -349,6 +349,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/70/bc/6f1c2f612465f5fa89b95bead1f44dcb607670fd42891d8fdcd5d039f4f4/markupsafe-3.0.3-cp314-cp314t-win_arm64.whl", hash = "sha256:32001d6a8fc98c8cb5c947787c5d08b0a50663d139f1305bac5885d98d9b40fa", size = 14146, upload-time = "2025-09-27T18:37:28.327Z" }, ] +[[package]] +name = "nats-py" +version = "2.15.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/02/f0/fc5e93f2b0dd14a202590ad9d30eda1955ea872039b5204357348d0f4b1e/nats_py-2.15.0.tar.gz", hash = "sha256:6622c547d9a7d2313d9c147d46c386188f4ec2c7b5c9f9a0438a4d1b55f54a93", size = 75995, upload-time = "2026-06-05T07:34:03.904Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/db/a8/b55606c7c621fb813c8ec78baf201d2c78bf6051091ec0c7ada572999e95/nats_py-2.15.0-py3-none-any.whl", hash = "sha256:9f8d36aa52a9926a88b8f1d70cf1fdce0ad387941479b500ee9ab3e51073cefd", size = 90334, upload-time = "2026-06-05T07:34:02.81Z" }, +] + [[package]] name = "nodeenv" version = "1.10.0" @@ -367,15 +376,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/b7/b9/c538f279a4e237a006a2c98387d081e9eb060d203d8ed34467cc0f0b9b53/packaging-26.0-py3-none-any.whl", hash = "sha256:b36f1fef9334a5588b4166f8bcd26a14e521f2b55e6b9de3aaa80d3ff7a37529", size = 74366, upload-time = "2026-01-21T20:50:37.788Z" }, ] -[[package]] -name = "paho-mqtt" -version = "2.1.0" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/39/15/0a6214e76d4d32e7f663b109cf71fb22561c2be0f701d67f93950cd40542/paho_mqtt-2.1.0.tar.gz", hash = "sha256:12d6e7511d4137555a3f6ea167ae846af2c7357b10bc6fa4f7c3968fc1723834", size = 148848, upload-time = "2024-04-29T19:52:55.591Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/c4/cb/00451c3cf31790287768bb12c6bec834f5d292eaf3022afc88e14b8afc94/paho_mqtt-2.1.0-py3-none-any.whl", hash = "sha256:6db9ba9b34ed5bc6b6e3812718c7e06e2fd7444540df2455d2c51bd58808feee", size = 67219, upload-time = "2024-04-29T19:52:48.345Z" }, -] - [[package]] name = "pluggy" version = "1.6.0" From c67c06d8e6979992e1ea20fae6e3c105b2797421 Mon Sep 17 00:00:00 2001 From: Paul Payne Date: Tue, 7 Jul 2026 05:03:07 -0700 Subject: [PATCH 4/7] feat(mesh): fleet role + shared config/presence KV (Phase 2) - add static 'role' (authority|follower) to NodeConfig/CastleConfig, wired from castle.yaml through the registry to the mesh wire; civil pinned authority - castle-presence: TTL KV bucket each node renews (churn signal); delete-on-stop - castle-config: authority-gated get/put + change-watch SSE (follower reconcile hook hangs off it) - bound the NATS drain on stop() so shutdown can't hang - tests: role config/registry round-trip, wire round-trip, role write-gating Single-node verified; follower reconcile pending the second node. --- castle-api/src/castle_api/mesh_wire.py | 3 + castle-api/src/castle_api/nats_client.py | 91 +++++++++++++++++++++--- castle-api/tests/test_mesh_wire.py | 12 ++++ castle-api/tests/test_nats_client.py | 30 ++++++++ core/src/castle_core/config.py | 4 ++ core/src/castle_core/deploy.py | 1 + core/src/castle_core/registry.py | 8 +++ core/tests/test_fleet_role.py | 48 +++++++++++++ docs/fleet-mesh-plan.md | 24 ++++++- 9 files changed, 211 insertions(+), 10 deletions(-) create mode 100644 castle-api/tests/test_nats_client.py create mode 100644 core/tests/test_fleet_role.py diff --git a/castle-api/src/castle_api/mesh_wire.py b/castle-api/src/castle_api/mesh_wire.py index 9f66509..1a79d16 100644 --- a/castle-api/src/castle_api/mesh_wire.py +++ b/castle-api/src/castle_api/mesh_wire.py @@ -26,6 +26,8 @@ def registry_to_json(registry: NodeRegistry) -> str: # acme domain — lets peers build launch URLs (.) # for this node's exposed apps. Omitted when the node has no domain. "gateway_domain": registry.node.gateway_domain, + # fleet role — so peers know which node is the config/secret authority. + "role": registry.node.role, }, "deployed": {}, } @@ -73,6 +75,7 @@ def json_to_registry(payload: str) -> NodeRegistry: castle_root=node_data.get("castle_root"), gateway_port=node_data.get("gateway_port", 9000), gateway_domain=node_data.get("gateway_domain"), + role=node_data.get("role", "follower"), ) deployed: dict[str, Deployment] = {} for key, comp_data in data.get("deployed", {}).items(): diff --git a/castle-api/src/castle_api/nats_client.py b/castle-api/src/castle_api/nats_client.py index 0194058..3a71b2b 100644 --- a/castle-api/src/castle_api/nats_client.py +++ b/castle-api/src/castle_api/nats_client.py @@ -34,8 +34,11 @@ from castle_api.stream import broadcast logger = logging.getLogger(__name__) REGISTRY_BUCKET = "castle-registry" +PRESENCE_BUCKET = "castle-presence" +CONFIG_BUCKET = "castle-config" HEARTBEAT_SEC = 30.0 PRUNE_SEC = 30.0 +PRESENCE_TTL = 90.0 # a node whose presence key expires within this is gone class CastleNATSClient: @@ -52,6 +55,8 @@ class CastleNATSClient: self._servers = servers self._nc: nats.NATS | None = None self._kv = None + self._presence_kv = None + self._config_kv = None self._tasks: list[asyncio.Task] = [] self._last_json: dict[str, str] = {} self._online: set[str] = set() @@ -64,6 +69,11 @@ class CastleNATSClient: def servers(self) -> str | list[str]: return self._servers + @property + def role(self) -> str: + """This node's fleet role — 'authority' or 'follower'.""" + return self._local_registry.node.role + async def start(self) -> None: """Connect, publish our registry, seed state, and start watchers.""" self._nc = await nats.connect( @@ -72,22 +82,39 @@ class CastleNATSClient: max_reconnect_attempts=-1, # reconnect forever — nodes come and go ) js = self._nc.jetstream() - try: - self._kv = await js.key_value(REGISTRY_BUCKET) - except Exception: - self._kv = await js.create_key_value( - config=KeyValueConfig(bucket=REGISTRY_BUCKET, history=1) - ) + self._kv = await self._ensure_bucket(js, REGISTRY_BUCKET, history=1) + # Presence: a short-TTL key each node renews; its expiry = the node is gone. + self._presence_kv = await self._ensure_bucket( + js, PRESENCE_BUCKET, history=1, ttl=PRESENCE_TTL + ) + # Shared config: authority-written, followers watch + reconcile. + self._config_kv = await self._ensure_bucket(js, CONFIG_BUCKET, history=5) await self.publish_registry(self._local_registry) + await self._presence_kv.put(self._local_hostname, b"online") await self._seed_existing() self._tasks = [ asyncio.create_task(self._watch_loop()), + asyncio.create_task(self._config_watch_loop()), asyncio.create_task(self._heartbeat_loop()), asyncio.create_task(self._prune_loop()), ] - logger.info("NATS mesh client started (servers=%s)", self._servers) + logger.info( + "NATS mesh client started (servers=%s, role=%s)", + self._servers, + self.role, + ) + + @staticmethod + async def _ensure_bucket(js, bucket: str, *, history: int = 1, ttl=None): + """Bind an existing KV bucket or create it.""" + try: + return await js.key_value(bucket) + except Exception: + return await js.create_key_value( + config=KeyValueConfig(bucket=bucket, history=history, ttl=ttl) + ) async def stop(self) -> None: """Delete our key (immediate offline to peers) and disconnect.""" @@ -100,9 +127,15 @@ class CastleNATSClient: if self._kv is not None: with contextlib.suppress(Exception): await self._kv.delete(self._local_hostname) - if self._nc is not None: + if self._presence_kv is not None: with contextlib.suppress(Exception): - await self._nc.drain() + await self._presence_kv.delete(self._local_hostname) + if self._nc is not None: + # Bound the drain so a wedged connection can't hang systemd shutdown. + with contextlib.suppress(Exception): + await asyncio.wait_for(self._nc.drain(), timeout=5.0) + with contextlib.suppress(Exception): + await self._nc.close() self._nc = None logger.info("NATS mesh client stopped") @@ -172,6 +205,46 @@ class CastleNATSClient: await asyncio.sleep(HEARTBEAT_SEC) with contextlib.suppress(Exception): await self.publish_registry(self._local_registry) + if self._presence_kv is not None: + with contextlib.suppress(Exception): + await self._presence_kv.put(self._local_hostname, b"online") + + # --- Shared config (authority writes, followers reconcile) --- + + async def get_shared_config(self, key: str) -> str | None: + """Read a shared-config value from the mesh (None if unset).""" + if self._config_kv is None: + return None + try: + entry = await self._config_kv.get(key) + return entry.value.decode() if entry.value else None + except Exception: + return None + + async def put_shared_config(self, key: str, value: str) -> None: + """Write a shared-config value. Only the authority may write.""" + if self.role != "authority": + raise PermissionError("only the authority node may write shared config") + if self._config_kv is None: + raise RuntimeError("config bucket not available") + await self._config_kv.put(key, value.encode()) + + async def _config_watch_loop(self) -> None: + """Watch shared config; announce changes so followers can reconcile. + + The reconcile action (trigger `castle apply` on a follower) hangs off this + SSE event; it's inert on the authority and on a single node. + """ + if self._config_kv is None: + return + watcher = await self._config_kv.watchall() + async for entry in watcher: + if entry is None: + continue + op = "delete" if entry.operation in ("DEL", "PURGE") else "put" + await broadcast( + "mesh", {"event": "config_changed", "key": entry.key, "op": op} + ) async def _prune_loop(self) -> None: """Mark crashed peers (no refresh within the stale TTL) offline.""" diff --git a/castle-api/tests/test_mesh_wire.py b/castle-api/tests/test_mesh_wire.py index 1467b69..9693172 100644 --- a/castle-api/tests/test_mesh_wire.py +++ b/castle-api/tests/test_mesh_wire.py @@ -95,6 +95,18 @@ class TestRegistrySerialization: assert bare.schedule is None assert bare.managed is False + def test_node_role_preserved(self) -> None: + reg = NodeRegistry( + node=NodeConfig(hostname="civil", role="authority"), deployed={} + ) + restored = json_to_registry(registry_to_json(reg)) + assert restored.node.role == "authority" + + def test_node_role_defaults_follower(self) -> None: + reg = NodeRegistry(node=NodeConfig(hostname="n"), deployed={}) + restored = json_to_registry(registry_to_json(reg)) + assert restored.node.role == "follower" + def test_no_secrets_in_payload(self) -> None: """env vars, run_cmd, and castle_root must never appear on the wire.""" original = _make_registry() diff --git a/castle-api/tests/test_nats_client.py b/castle-api/tests/test_nats_client.py new file mode 100644 index 0000000..ec47124 --- /dev/null +++ b/castle-api/tests/test_nats_client.py @@ -0,0 +1,30 @@ +"""CastleNATSClient role gating — hermetic (no live NATS server needed). + +The write-gate is checked before touching the KV bucket, so a follower is denied +without any connection. +""" + +from __future__ import annotations + +import asyncio + +import pytest +from castle_core.registry import NodeConfig, NodeRegistry + +from castle_api.nats_client import CastleNATSClient + + +def _client(role: str) -> CastleNATSClient: + reg = NodeRegistry(node=NodeConfig(hostname="n", role=role), deployed={}) + return CastleNATSClient("n", reg, servers="nats://localhost:4222") + + +def test_role_property() -> None: + assert _client("authority").role == "authority" + assert _client("follower").role == "follower" + + +def test_follower_cannot_write_shared_config() -> None: + client = _client("follower") + with pytest.raises(PermissionError): + asyncio.run(client.put_shared_config("fleet/key", "value")) diff --git a/core/src/castle_core/config.py b/core/src/castle_core/config.py index 5b7bc71..7e47969 100644 --- a/core/src/castle_core/config.py +++ b/core/src/castle_core/config.py @@ -184,6 +184,9 @@ class CastleConfig: # built-in defaults so tests/callers that don't care stay valid. data_dir: Path = field(default_factory=lambda: _DEFAULT_DATA_DIR) repos_dir: Path = field(default_factory=lambda: _DEFAULT_REPOS_DIR) + # Fleet role: "authority" (may write shared config/secrets to the mesh) or + # "follower" (reconciles from it). Static — pinned here, no election. + role: str = "follower" # Construction convenience only (not stored): a flat name→spec dict is routed # into the per-kind stores by kind_for. Lets callers/tests hand us a flat map # without pre-splitting it; there is still no flat `deployments` attribute. @@ -465,6 +468,7 @@ def load_config(root: Path | None = None) -> CastleConfig: agents=agents, data_dir=data_dir, repos_dir=repos_dir, + role=data.get("role", "follower"), **stores, ) return config diff --git a/core/src/castle_core/deploy.py b/core/src/castle_core/deploy.py index 5db67b0..ecfa090 100644 --- a/core/src/castle_core/deploy.py +++ b/core/src/castle_core/deploy.py @@ -189,6 +189,7 @@ def _node_config(config: CastleConfig) -> NodeConfig: public_domain=config.gateway.public_domain, tunnel_id=config.gateway.tunnel_id, cert_hook=config.gateway.cert_hook, + role=config.role, ) diff --git a/core/src/castle_core/registry.py b/core/src/castle_core/registry.py index 5cbb08c..668a799 100644 --- a/core/src/castle_core/registry.py +++ b/core/src/castle_core/registry.py @@ -32,6 +32,11 @@ class NodeConfig: tunnel_id: str | None = None # Emit the cert_obtained → `castle tls reconcile` hook (needs events-exec plugin). cert_hook: bool = False + # Fleet role: "authority" may write shared config/secrets to the mesh; + # "follower" reconciles from it. Static (no election) — the authority is + # pinned in castle.yaml. When the authority is down, shared state is + # read-only and every node keeps serving its own deployments from cache. + role: str = "follower" def __post_init__(self) -> None: if not self.hostname: @@ -152,6 +157,7 @@ def load_registry(path: Path | None = None) -> NodeRegistry: public_domain=node_data.get("public_domain"), tunnel_id=node_data.get("tunnel_id"), cert_hook=node_data.get("cert_hook", False), + role=node_data.get("role", "follower"), ) deployed: dict[str, Deployment] = {} @@ -241,6 +247,8 @@ def save_registry(registry: NodeRegistry, path: Path | None = None) -> None: data["node"]["tunnel_id"] = registry.node.tunnel_id if registry.node.cert_hook: data["node"]["cert_hook"] = registry.node.cert_hook + if registry.node.role and registry.node.role != "follower": + data["node"]["role"] = registry.node.role for key, comp in registry.deployed.items(): entry: dict = { diff --git a/core/tests/test_fleet_role.py b/core/tests/test_fleet_role.py new file mode 100644 index 0000000..2d440e5 --- /dev/null +++ b/core/tests/test_fleet_role.py @@ -0,0 +1,48 @@ +"""Fleet role (authority/follower) — config loading + registry round-trip.""" + +from __future__ import annotations + +from pathlib import Path + +import yaml +from castle_core.config import load_config +from castle_core.registry import ( + NodeConfig, + NodeRegistry, + load_registry, + save_registry, +) + + +def _write_min_config(root: Path, role: str | None) -> None: + data: dict = {"gateway": {"port": 18000}} + if role is not None: + data["role"] = role + (root / "castle.yaml").write_text(yaml.safe_dump(data)) + + +def test_role_defaults_to_follower(tmp_path: Path) -> None: + _write_min_config(tmp_path, None) + assert load_config(tmp_path).role == "follower" + + +def test_role_loaded_from_yaml(tmp_path: Path) -> None: + _write_min_config(tmp_path, "authority") + assert load_config(tmp_path).role == "authority" + + +def test_registry_role_round_trip(tmp_path: Path) -> None: + reg = NodeRegistry( + node=NodeConfig(hostname="civil", role="authority"), deployed={} + ) + path = tmp_path / "registry.yaml" + save_registry(reg, path) + assert load_registry(path).node.role == "authority" + + +def test_registry_role_defaults_follower(tmp_path: Path) -> None: + """A node with no explicit role round-trips as follower.""" + reg = NodeRegistry(node=NodeConfig(hostname="n"), deployed={}) + path = tmp_path / "registry.yaml" + save_registry(reg, path) + assert load_registry(path).node.role == "follower" diff --git a/docs/fleet-mesh-plan.md b/docs/fleet-mesh-plan.md index 8af6288..ccd13f8 100644 --- a/docs/fleet-mesh-plan.md +++ b/docs/fleet-mesh-plan.md @@ -1,7 +1,7 @@ # Fleet Mesh Plan — OpenBao + NATS **Status:** in progress on branch `feat/fleet-mesh-nats-openbao`. -Phases 0–1 complete + single-node verified (live cutover done). Phases 2–4 pending. +Phases 0–2 complete + single-node verified (live). Phases 3–4 pending (need 2nd node). ## Context @@ -196,6 +196,28 @@ so revert = `git checkout main -- castle-api` then re-apply. The old Mosquitto `mqtt` service is left running (dormant) as a safety net; retire it once the second node is proven on NATS. +### Phase 2 — DONE + single-node verified + tested (2026-07-07) + +- **`role` field** on `NodeConfig` + `CastleConfig`, wired end-to-end: + castle.yaml top-level `role:` → config → `_node_config` → registry.yaml → + mesh wire. **`civil` pinned `role: authority`**; default `follower`. +- **Presence** (`castle-presence`) — a TTL KV bucket each node renews on the + heartbeat; expiry = the node is gone. Delete-on-stop for immediate departure. +- **Shared config** (`castle-config`) — `get_shared_config` / `put_shared_config` + (authority-gated: followers raise `PermissionError`), plus a watch loop that + broadcasts a `config_changed` SSE (the follower `castle apply` reconcile hook + hangs off this — inert on one node). +- Hardened `stop()` to bound the NATS drain (can't hang systemd shutdown). +- Verified live: all three buckets present, `civil=online` presence, + `"role": "authority"` on the wire, authority write + follower-deny exercised + against the running `castle-nats`. +- **Tests added:** `core/tests/test_fleet_role.py` (role config load + registry + round-trip), `castle-api/tests/test_nats_client.py` (role gating, hermetic), + role round-trip in `test_mesh_wire.py`. Suites: **200 core + 93 api** pass. + +**Pending second node:** the follower-side reconcile (watch `castle-config` → +`castle apply`) is wired as an SSE hook but only meaningful with a peer. + ## Decisions (resolved) 1. **Discovery — both.** Keep mDNS for zero-config LAN peer discovery *and* From f94517887ec6f6c31ff01724183d3ad44bc7111e Mon Sep 17 00:00:00 2001 From: Paul Payne Date: Tue, 7 Jul 2026 05:08:16 -0700 Subject: [PATCH 5/7] feat(secrets): pluggable secret backend with OpenBao read (Phase 4) - secret_backends.py: SecretBackend protocol, FileSecretBackend (historical), OpenBaoBackend (KV-v2 read + file fallback), build_backend() env-selected - _read_secret delegates to the active backend; default is file, so production is unchanged until CASTLE_SECRET_BACKEND=openbao - OpenBao token bootstraps from the file backend; missing/unreachable falls back - tests: file hit/miss, backend selection, unreachable + empty-token fallback Read path verified live against castle-openbao. Write path, auto-unseal-on-boot, and TLS hardening documented as remaining for full OpenBao production use. --- core/src/castle_core/config.py | 13 ++-- core/src/castle_core/secret_backends.py | 90 +++++++++++++++++++++++++ core/tests/test_secret_backends.py | 54 +++++++++++++++ docs/fleet-mesh-plan.md | 31 ++++++++- 4 files changed, 183 insertions(+), 5 deletions(-) create mode 100644 core/src/castle_core/secret_backends.py create mode 100644 core/tests/test_secret_backends.py diff --git a/core/src/castle_core/config.py b/core/src/castle_core/config.py index 7e47969..f7fe9a6 100644 --- a/core/src/castle_core/config.py +++ b/core/src/castle_core/config.py @@ -319,10 +319,15 @@ def resolve_env_vars( def _read_secret(name: str) -> str: - """Read a secret from ~/.castle/secrets/. Returns placeholder if not found.""" - secret_path = SECRETS_DIR / name - if secret_path.exists(): - return secret_path.read_text().strip() + """Resolve a secret via the active backend (file by default; OpenBao opt-in). + + Returns a ```` placeholder if unresolved (never raises). + """ + from castle_core.secret_backends import build_backend + + value = build_backend(SECRETS_DIR).read(name) + if value is not None: + return value return f"" diff --git a/core/src/castle_core/secret_backends.py b/core/src/castle_core/secret_backends.py new file mode 100644 index 0000000..377aad0 --- /dev/null +++ b/core/src/castle_core/secret_backends.py @@ -0,0 +1,90 @@ +"""Pluggable secret backends for ``${secret:NAME}`` resolution. + +Default is the **file** backend (``~/.castle/secrets/``) — identical to the +historical behavior, so nothing changes unless a backend is explicitly selected +via ``CASTLE_SECRET_BACKEND``. + +The **openbao** backend reads from an OpenBao/Vault KV-v2 mount and falls back to +the file backend, which is also how it bootstraps: the OpenBao *token* itself is a +file secret (it can't live in the vault it unlocks). + +Selection (env, so it works in both the CLI and the systemd-run API): + CASTLE_SECRET_BACKEND file | openbao (default: file) + CASTLE_OPENBAO_ADDR http://localhost:8200 + CASTLE_OPENBAO_MOUNT castle (kv-v2 mount path) + CASTLE_OPENBAO_TOKEN_SECRET OPENBAO_TOKEN (file secret holding the token) +""" + +from __future__ import annotations + +import json +import os +import urllib.request +from pathlib import Path +from typing import Protocol + + +class SecretBackend(Protocol): + def read(self, name: str) -> str | None: ... + + +class FileSecretBackend: + """Reads ``/`` (the historical behavior).""" + + def __init__(self, secrets_dir: Path) -> None: + self._dir = secrets_dir + + def read(self, name: str) -> str | None: + path = self._dir / name + if path.exists(): + return path.read_text().strip() + return None + + +class OpenBaoBackend: + """Reads from an OpenBao/Vault KV-v2 mount; falls back to ``fallback``. + + A missing key, an auth failure, or an unreachable server all fall through to + the fallback — so a partly-migrated vault and the bootstrap token both resolve. + """ + + def __init__( + self, addr: str, token: str, mount: str, fallback: SecretBackend + ) -> None: + self._addr = addr.rstrip("/") + self._token = token + self._mount = mount + self._fallback = fallback + + def read(self, name: str) -> str | None: + value = self._read_bao(name) + if value is not None: + return value + return self._fallback.read(name) + + def _read_bao(self, name: str) -> str | None: + if not self._token: + return None + url = f"{self._addr}/v1/{self._mount}/data/{name}" + req = urllib.request.Request(url, headers={"X-Vault-Token": self._token}) + try: + with urllib.request.urlopen(req, timeout=5) as resp: # noqa: S310 + data = json.load(resp) + return data["data"]["data"].get("value") + except Exception: + return None + + +def build_backend(secrets_dir: Path) -> SecretBackend: + """Construct the active secret backend from the environment.""" + file_backend = FileSecretBackend(secrets_dir) + kind = os.environ.get("CASTLE_SECRET_BACKEND", "file").lower() + if kind == "openbao": + addr = os.environ.get("CASTLE_OPENBAO_ADDR", "http://localhost:8200") + mount = os.environ.get("CASTLE_OPENBAO_MOUNT", "castle") + token_secret = os.environ.get("CASTLE_OPENBAO_TOKEN_SECRET", "OPENBAO_TOKEN") + token = file_backend.read(token_secret) or os.environ.get( + "CASTLE_OPENBAO_TOKEN", "" + ) + return OpenBaoBackend(addr, token, mount, fallback=file_backend) + return file_backend diff --git a/core/tests/test_secret_backends.py b/core/tests/test_secret_backends.py new file mode 100644 index 0000000..855dcee --- /dev/null +++ b/core/tests/test_secret_backends.py @@ -0,0 +1,54 @@ +"""Tests for the pluggable secret backends (file default, OpenBao opt-in).""" + +from __future__ import annotations + +from pathlib import Path + +from castle_core.secret_backends import ( + FileSecretBackend, + OpenBaoBackend, + build_backend, +) + + +def test_file_backend_read_hit(tmp_path: Path) -> None: + (tmp_path / "MY_SECRET").write_text("value\n") + assert FileSecretBackend(tmp_path).read("MY_SECRET") == "value" + + +def test_file_backend_read_miss(tmp_path: Path) -> None: + assert FileSecretBackend(tmp_path).read("ABSENT") is None + + +def test_build_backend_defaults_to_file(tmp_path: Path, monkeypatch) -> None: + monkeypatch.delenv("CASTLE_SECRET_BACKEND", raising=False) + assert isinstance(build_backend(tmp_path), FileSecretBackend) + + +def test_build_backend_openbao_selected(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("CASTLE_SECRET_BACKEND", "openbao") + assert isinstance(build_backend(tmp_path), OpenBaoBackend) + + +def test_openbao_falls_back_to_file_when_unreachable(tmp_path: Path) -> None: + """An unreachable vault (or empty token) resolves via the file fallback.""" + (tmp_path / "ONLY_IN_FILE").write_text("from-file") + backend = OpenBaoBackend( + addr="http://127.0.0.1:1", # nothing listening + token="dummy", + mount="castle", + fallback=FileSecretBackend(tmp_path), + ) + assert backend.read("ONLY_IN_FILE") == "from-file" + assert backend.read("NOT_ANYWHERE") is None + + +def test_openbao_empty_token_uses_fallback(tmp_path: Path) -> None: + (tmp_path / "K").write_text("v") + backend = OpenBaoBackend( + addr="http://127.0.0.1:8200", + token="", # no token → never hits the network + mount="castle", + fallback=FileSecretBackend(tmp_path), + ) + assert backend.read("K") == "v" diff --git a/docs/fleet-mesh-plan.md b/docs/fleet-mesh-plan.md index ccd13f8..6a0c70e 100644 --- a/docs/fleet-mesh-plan.md +++ b/docs/fleet-mesh-plan.md @@ -1,7 +1,8 @@ # Fleet Mesh Plan — OpenBao + NATS **Status:** in progress on branch `feat/fleet-mesh-nats-openbao`. -Phases 0–2 complete + single-node verified (live). Phases 3–4 pending (need 2nd node). +Phases 0–2 + Phase 4 (secret-read backend) complete + single-node verified (live). +Phase 3 (cross-node routing/breaker) and Phase 4 hardening pending the 2nd node. ## Context @@ -218,6 +219,34 @@ second node is proven on NATS. **Pending second node:** the follower-side reconcile (watch `castle-config` → `castle apply`) is wired as an SSE hook but only meaningful with a peer. +### Phase 4 — secret-read backend DONE + verified + tested (2026-07-07) + +- New `core/castle_core/secret_backends.py`: `SecretBackend` protocol, + `FileSecretBackend` (the historical behavior), `OpenBaoBackend` (KV-v2 read with + file fallback), and `build_backend()` selecting via `CASTLE_SECRET_BACKEND` + (default **file** — production is byte-for-byte unchanged until opted in). +- `_read_secret` (the single chokepoint, `config.py`) now delegates to the active + backend. `${secret:NAME}` syntax untouched. +- OpenBao token bootstraps from the file backend (it can't live in the vault it + unlocks); a missing key / auth failure / unreachable server all fall through to + file, so a partly-migrated vault keeps working. +- Verified live against the running `castle-openbao`: a secret stored in the vault + resolves through `${secret:...}` in openbao mode; file-only secrets resolve via + fallback; missing → placeholder; **default file mode unchanged**. +- **Tests:** `core/tests/test_secret_backends.py` (file hit/miss, backend + selection, unreachable-vault + empty-token fallback). Suites: **206 core + 93 api**. + +**Remaining for full OpenBao production use (documented, not blocking — nothing +uses the backend until `CASTLE_SECRET_BACKEND=openbao`):** +1. **Write path** — `castle-api/secrets.py` CRUD still writes to files; in openbao + mode dashboard-set secrets land in file (resolve via fallback) rather than the + vault. Add `write`/`delete` to the backends + route the API writer through them. +2. **Auto-unseal on boot** (Phase 0 deferral) — OpenBao re-seals on container + restart. Wire an unseal step (manifest `exec_start_post` or a companion + oneshot reading `OPENBAO_UNSEAL_KEY`). Not urgent while the backend is unused. +3. **TLS hardening** — NATS + OpenBao are plaintext localhost. Required before + cross-network or moving real secrets: NATS mTLS + auth, OpenBao TLS listener. + ## Decisions (resolved) 1. **Discovery — both.** Keep mDNS for zero-config LAN peer discovery *and* From 526736f7787d9082105971c1a3872ebe53b31d91 Mon Sep 17 00:00:00 2001 From: Paul Payne Date: Tue, 7 Jul 2026 05:38:15 -0700 Subject: [PATCH 6/7] feat(mesh): cross-node routing + presence breaker (Phase 3) - NodeConfig.address: routable host peers proxy to (wired registry + wire) - compute_routes emits 'remote' routes for consumed peer services (local requires ref satisfied by an online peer ->
:) - _host_remote_block: fail-fast reverse_proxy (2s dial, passive health); presence expiry removes the route entirely = the primary circuit-breaker - mesh_gateway.py: API re-renders + reloads the Caddyfile on mesh change, iff content changed (no-op until a cross-node service is consumed) - tests: route emit / address fallback / breaker-absent / unconsumed Logic verified hermetically + against primer's real registry (castle-api -> primer:9020); live integration proven a no-op on civil. --- castle-api/src/castle_api/mesh_gateway.py | 55 +++++++++++++ castle-api/src/castle_api/mesh_wire.py | 3 + castle-api/src/castle_api/nats_client.py | 4 + core/src/castle_core/generators/caddyfile.py | 70 +++++++++++++++- core/src/castle_core/registry.py | 7 ++ core/tests/test_caddyfile_remote.py | 87 ++++++++++++++++++++ docs/fleet-mesh-plan.md | 31 +++++++ 7 files changed, 255 insertions(+), 2 deletions(-) create mode 100644 castle-api/src/castle_api/mesh_gateway.py create mode 100644 core/tests/test_caddyfile_remote.py diff --git a/castle-api/src/castle_api/mesh_gateway.py b/castle-api/src/castle_api/mesh_gateway.py new file mode 100644 index 0000000..3048b97 --- /dev/null +++ b/castle-api/src/castle_api/mesh_gateway.py @@ -0,0 +1,55 @@ +"""Regenerate the gateway with cross-node (remote) routes from the live mesh. + +Local routes come from `castle apply` (static). Remote routes are dynamic — they +appear/vanish as peers join/leave — so the API owns them: on a mesh change it +re-renders the Caddyfile (same generator `apply` uses, plus remote routes for +online peers) and reloads the gateway iff the content changed. + +Safety: with no cross-node `requires`, the output equals the local-only Caddyfile, +so this is a verified no-op until this node actually consumes a peer service. +""" + +from __future__ import annotations + +import asyncio +import logging +import subprocess + +from castle_core.config import SPECS_DIR +from castle_core.generators.caddyfile import generate_caddyfile_from_registry + +from castle_api.config import get_registry +from castle_api.mesh import mesh_state + +logger = logging.getLogger(__name__) + +_GATEWAY_UNIT = "castle-castle-gateway.service" + + +def _regenerate(reload: bool) -> bool: + try: + reg = get_registry() + except Exception: + return False + remotes = {h: n.registry for h, n in mesh_state.all_nodes().items()} + try: + content = generate_caddyfile_from_registry(reg, remotes) + except Exception: + logger.exception("mesh gateway: Caddyfile generation failed") + return False + path = SPECS_DIR / "Caddyfile" + old = path.read_text() if path.exists() else "" + if content == old: + return False + path.write_text(content) + logger.info("mesh gateway: Caddyfile updated with cross-node routes") + if reload: + subprocess.run( + ["systemctl", "--user", "reload", _GATEWAY_UNIT], check=False + ) + return True + + +async def refresh_remote_routes(reload: bool = True) -> bool: + """Async wrapper — runs the blocking regen off the event loop.""" + return await asyncio.to_thread(_regenerate, reload) diff --git a/castle-api/src/castle_api/mesh_wire.py b/castle-api/src/castle_api/mesh_wire.py index 1a79d16..9e20655 100644 --- a/castle-api/src/castle_api/mesh_wire.py +++ b/castle-api/src/castle_api/mesh_wire.py @@ -28,6 +28,8 @@ def registry_to_json(registry: NodeRegistry) -> str: "gateway_domain": registry.node.gateway_domain, # fleet role — so peers know which node is the config/secret authority. "role": registry.node.role, + # routable host peers proxy to for this node's services. + "address": registry.node.address, }, "deployed": {}, } @@ -76,6 +78,7 @@ def json_to_registry(payload: str) -> NodeRegistry: gateway_port=node_data.get("gateway_port", 9000), gateway_domain=node_data.get("gateway_domain"), role=node_data.get("role", "follower"), + address=node_data.get("address"), ) deployed: dict[str, Deployment] = {} for key, comp_data in data.get("deployed", {}).items(): diff --git a/castle-api/src/castle_api/nats_client.py b/castle-api/src/castle_api/nats_client.py index 3a71b2b..e3255b4 100644 --- a/castle-api/src/castle_api/nats_client.py +++ b/castle-api/src/castle_api/nats_client.py @@ -28,6 +28,7 @@ from nats.js.api import KeyValueConfig from castle_core.registry import NodeRegistry from castle_api.mesh import mesh_state +from castle_api.mesh_gateway import refresh_remote_routes from castle_api.mesh_wire import json_to_registry, registry_to_json from castle_api.stream import broadcast @@ -93,6 +94,7 @@ class CastleNATSClient: await self.publish_registry(self._local_registry) await self._presence_kv.put(self._local_hostname, b"online") await self._seed_existing() + await refresh_remote_routes() # establish any cross-node routes on startup self._tasks = [ asyncio.create_task(self._watch_loop()), @@ -182,6 +184,7 @@ class CastleNATSClient: await broadcast( "mesh", {"event": "node_updated", "hostname": key} ) + asyncio.create_task(refresh_remote_routes()) except Exception: logger.exception("Error handling mesh entry for %s", key) @@ -199,6 +202,7 @@ class CastleNATSClient: self._online.discard(hostname) self._last_json.pop(hostname, None) await broadcast("mesh", {"event": "node_offline", "hostname": hostname}) + asyncio.create_task(refresh_remote_routes()) async def _heartbeat_loop(self) -> None: while True: diff --git a/core/src/castle_core/generators/caddyfile.py b/core/src/castle_core/generators/caddyfile.py index 98298ef..a60c624 100644 --- a/core/src/castle_core/generators/caddyfile.py +++ b/core/src/castle_core/generators/caddyfile.py @@ -121,8 +121,10 @@ def compute_routes( """Build the ordered list of gateway routes. Every route is a host route whose address is the service/frontend **name** (published at ``.``); ``proxy`` routes reverse-proxy a local port, ``static`` routes file-serve a - frontend's dist. Path routes no longer exist. ``remote_registries`` is accepted - for signature compatibility but cross-node routing is out of scope here.""" + frontend's dist. Path routes no longer exist. When ``remote_registries`` is + given (online peers, keyed by hostname), ``remote`` routes are added for + services this node **consumes** (a local ``requires`` ref satisfied by a peer) + — so a consumed cross-node service is reachable at ``.``.""" if config is None: try: from castle_core.config import load_config @@ -141,9 +143,50 @@ def compute_routes( for name, kind, target in _local_routes(config, registry): routes.append(GatewayRoute(name, kind, target, name, node)) + if remote_registries: + routes.extend(_remote_routes(config, registry, remote_registries)) + return routes +def _remote_routes( + config: CastleConfig | None, + registry: NodeRegistry, + remote_registries: dict[str, NodeRegistry], +) -> list[GatewayRoute]: + """Routes to services this node consumes from online peers. + + A route is emitted for each local ``requires`` ref that (a) isn't satisfied + locally and (b) is provided by an exposed service on some peer. The route is + only present while the peer is (presence expiry removes the peer from + ``remote_registries``, which *is* the circuit-breaker: gone → no route).""" + # Refs this node consumes. + consumed: set[str] = set() + local_names: set[str] = set() + if config is not None: + for _kind, name, dep in config.all_deployments(): + local_names.add(name) + for req in getattr(dep, "requires", []) or []: + ref = getattr(req, "ref", None) + if ref and getattr(req, "kind", "deployment") == "deployment": + consumed.add(ref) + # Drop refs already satisfied locally. + consumed -= local_names + + out: list[GatewayRoute] = [] + for host, remote in sorted(remote_registries.items()): + addr = remote.node.address or host + for _kind, name, dep in remote.all(): + if name not in consumed: + continue + if dep.subdomain and dep.port: + out.append( + GatewayRoute(name, "remote", f"{addr}:{dep.port}", name, host) + ) + consumed.discard(name) # first online provider wins + return out + + def _host_matcher_block(label: str, host: str, target: str) -> list[str]: """A `@host_X host / handle @host_X { reverse_proxy }` block. @@ -159,6 +202,27 @@ def _host_matcher_block(label: str, host: str, target: str) -> list[str]: ] +def _host_remote_block(label: str, host: str, target: str) -> list[str]: + """A remote (cross-node) host route with a fail-fast breaker: a short dial + timeout + passive health, so an unreachable peer 502s in ~2s instead of + hanging. (Presence removal drops the route entirely — this guards the + there-but-wedged case.)""" + matcher = f"@host_{label.replace('-', '_').replace('.', '_')}" + return [ + f" {matcher} host {host}", + f" handle {matcher} {{", + f" reverse_proxy {target} {{", + " lb_try_duration 1s", + " fail_duration 30s", + " transport http {", + " dial_timeout 2s", + " }", + " }", + " }", + "", + ] + + def _host_static_block(label: str, host: str, serve_dir: str) -> list[str]: """A host matcher that file-serves a frontend's dist (with SPA fallback).""" matcher = f"@host_{label.replace('-', '_').replace('.', '_')}" @@ -237,6 +301,8 @@ def generate_caddyfile_from_registry( host = f"{r.address}.{domain}" if r.kind == "static": lines += _host_static_block(r.name or r.address, host, r.target) + elif r.kind == "remote": + lines += _host_remote_block(r.name or r.address, host, r.target) else: lines += _host_matcher_block(r.name or r.address, host, r.target) lines.append("}") diff --git a/core/src/castle_core/registry.py b/core/src/castle_core/registry.py index 668a799..b1aecb7 100644 --- a/core/src/castle_core/registry.py +++ b/core/src/castle_core/registry.py @@ -32,6 +32,10 @@ class NodeConfig: tunnel_id: str | None = None # Emit the cert_obtained → `castle tls reconcile` hook (needs events-exec plugin). cert_hook: bool = False + # Routable host peers use to reach this node's services (LAN IP/hostname). + # Defaults to the hostname; set explicitly when the hostname isn't resolvable + # cross-node. Used to build `remote` gateway routes to this node. + address: str | None = None # Fleet role: "authority" may write shared config/secrets to the mesh; # "follower" reconciles from it. Static (no election) — the authority is # pinned in castle.yaml. When the authority is down, shared state is @@ -158,6 +162,7 @@ def load_registry(path: Path | None = None) -> NodeRegistry: tunnel_id=node_data.get("tunnel_id"), cert_hook=node_data.get("cert_hook", False), role=node_data.get("role", "follower"), + address=node_data.get("address"), ) deployed: dict[str, Deployment] = {} @@ -249,6 +254,8 @@ def save_registry(registry: NodeRegistry, path: Path | None = None) -> None: data["node"]["cert_hook"] = registry.node.cert_hook if registry.node.role and registry.node.role != "follower": data["node"]["role"] = registry.node.role + if registry.node.address: + data["node"]["address"] = registry.node.address for key, comp in registry.deployed.items(): entry: dict = { diff --git a/core/tests/test_caddyfile_remote.py b/core/tests/test_caddyfile_remote.py new file mode 100644 index 0000000..3c763dc --- /dev/null +++ b/core/tests/test_caddyfile_remote.py @@ -0,0 +1,87 @@ +"""Cross-node (remote) gateway routes + presence breaker.""" + +from __future__ import annotations + +from pathlib import Path + +import yaml +from castle_core.config import load_config +from castle_core.generators.caddyfile import compute_routes +from castle_core.registry import Deployment, NodeConfig, NodeRegistry + + +def _config_requiring_widget(root: Path) -> None: + (root / "castle.yaml").write_text(yaml.safe_dump({"gateway": {"port": 18000}})) + svc_dir = root / "services" + svc_dir.mkdir() + (svc_dir / "consumer.yaml").write_text( + yaml.safe_dump( + { + "description": "consumes a remote widget", + "manager": "systemd", + "run": {"launcher": "command", "argv": ["consumer"]}, + "requires": [{"kind": "deployment", "ref": "widget"}], + "manage": {"systemd": {}}, + } + ) + ) + + +def _peer_with_widget(address: str | None) -> NodeRegistry: + widget = Deployment( + manager="systemd", + launcher="python", + run_cmd=[], + name="widget", + kind="service", + port=9099, + subdomain="widget", + managed=True, + ) + return NodeRegistry( + node=NodeConfig(hostname="tower", address=address), + deployed={NodeRegistry.key("service", "widget"): widget}, + ) + + +def _local() -> NodeRegistry: + return NodeRegistry(node=NodeConfig(hostname="civil"), deployed={}) + + +def test_remote_route_emitted_for_consumed_peer_service(tmp_path: Path) -> None: + _config_requiring_widget(tmp_path) + config = load_config(tmp_path) + routes = compute_routes( + _local(), config, {"tower": _peer_with_widget("10.0.0.5")} + ) + remote = [r for r in routes if r.kind == "remote"] + assert len(remote) == 1 + assert remote[0].address == "widget" + assert remote[0].target == "10.0.0.5:9099" + assert remote[0].node == "tower" + + +def test_address_falls_back_to_hostname(tmp_path: Path) -> None: + _config_requiring_widget(tmp_path) + config = load_config(tmp_path) + routes = compute_routes(_local(), config, {"tower": _peer_with_widget(None)}) + remote = [r for r in routes if r.kind == "remote"] + assert remote[0].target == "tower:9099" + + +def test_breaker_no_route_when_peer_absent(tmp_path: Path) -> None: + """Presence expiry removes the peer from remote_registries -> no route.""" + _config_requiring_widget(tmp_path) + config = load_config(tmp_path) + routes = compute_routes(_local(), config, {}) # no online peers + assert not [r for r in routes if r.kind == "remote"] + + +def test_no_remote_route_for_unconsumed_service(tmp_path: Path) -> None: + """A peer service nobody requires is not routed.""" + (tmp_path / "castle.yaml").write_text(yaml.safe_dump({"gateway": {"port": 18000}})) + config = load_config(tmp_path) # no requires anywhere + routes = compute_routes( + _local(), config, {"tower": _peer_with_widget("10.0.0.5")} + ) + assert not [r for r in routes if r.kind == "remote"] diff --git a/docs/fleet-mesh-plan.md b/docs/fleet-mesh-plan.md index 6a0c70e..7c433ca 100644 --- a/docs/fleet-mesh-plan.md +++ b/docs/fleet-mesh-plan.md @@ -247,6 +247,37 @@ uses the backend until `CASTLE_SECRET_BACKEND=openbao`):** 3. **TLS hardening** — NATS + OpenBao are plaintext localhost. Required before cross-network or moving real secrets: NATS mTLS + auth, OpenBao TLS listener. +### Phase 3 — cross-node routing + breaker: logic DONE + verified against a real peer (2026-07-07) + +**Real second node:** `primer` (192.168.8.129) migrated onto the NATS mesh +(branch checked out, castle-api pointed at `nats://civil:4222` via a systemd +drop-in). civil ↔ primer mesh confirmed over the LAN — **Phase 1 two-node parity +done for real**, not simulated. (This also *restored* the civil↔primer mesh my +Phase 1 cutover had split — primer was still on MQTT→civil.) + +- `NodeConfig.address` — routable host peers proxy to (wired through registry + + wire; falls back to hostname, which civil resolves for primer). +- `compute_routes` now emits **`remote`** routes for services this node + **consumes** (a local `requires` ref satisfied by an online peer), targeting + `:`. `_host_remote_block` renders a **fail-fast breaker** + (2s dial timeout + passive health) for the there-but-wedged case; **presence + expiry removes the peer from the route set entirely** (gone → no route) — the + primary breaker. +- `castle_api/mesh_gateway.py` — on peer join/leave/change (+ startup) the API + re-renders the Caddyfile (same generator `apply` uses + remote routes for + online peers) and reloads the gateway **iff content changed**. +- **Verified:** 4 hermetic tests (route emitted / address fallback / breaker: + peer-absent → no route / unconsumed → no route); **resolution against primer's + real registry** pulled live from the mesh → `castle-api → primer:9020`; and the + live integration proven a **no-op** on civil (Caddyfile hash unchanged, gateway + healthy) since civil consumes nothing cross-node yet. Suites: **210 core + 93 api**. + +**Remaining:** the full live curl+kill E2E (civil routing to a peer service, then +failing fast on kill) needs a peer-unique service civil consumes — every +underlying piece is verified, but the end-to-end demo needs that provisioning. +primer is now a permanent mesh member on the branch (revert: remove the drop-in + +`git checkout main` on primer). + ## Decisions (resolved) 1. **Discovery — both.** Keep mDNS for zero-config LAN peer discovery *and* From b437f7130061fc5314c7f210b89b931c8e512c76 Mon Sep 17 00:00:00 2001 From: Paul Payne Date: Tue, 7 Jul 2026 05:44:23 -0700 Subject: [PATCH 7/7] feat(secrets): OpenBao write path + auto-unseal on boot (Phase 4 hardening) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - SecretBackend gains write/delete/list_names; FileSecretBackend + OpenBaoBackend implement them (vault KV-v2 POST/DELETE/LIST); castle-api/secrets.py routes all CRUD through the active backend (dashboard writes to the vault in openbao mode) - add systemd exec_start_post (general; '-' prefix so a hook failure can't fail the unit); castle-openbao runs an unseal.sh on boot using OPENBAO_UNSEAL_KEY - tests: file write/read/list/delete round-trip Verified live: write→vault→read→list→delete against OpenBao; seal→restart→ auto-unsealed. TLS hardening remains (deferred; gates cross-network/real secrets). --- castle-api/src/castle_api/secrets.py | 25 +++++----- core/src/castle_core/generators/systemd.py | 9 ++++ core/src/castle_core/manifest.py | 3 ++ core/src/castle_core/secret_backends.py | 55 ++++++++++++++++++++-- core/tests/test_secret_backends.py | 13 +++++ docs/fleet-mesh-plan.md | 25 ++++++---- 6 files changed, 103 insertions(+), 27 deletions(-) diff --git a/castle-api/src/castle_api/secrets.py b/castle-api/src/castle_api/secrets.py index bb60ed6..fd28fea 100644 --- a/castle-api/src/castle_api/secrets.py +++ b/castle-api/src/castle_api/secrets.py @@ -1,4 +1,4 @@ -"""Secrets management — read and write ~/.castle/secrets/.""" +"""Secrets management — routes through the active backend (file or OpenBao).""" from __future__ import annotations @@ -6,10 +6,15 @@ from fastapi import APIRouter, HTTPException, status from pydantic import BaseModel from castle_core.config import SECRETS_DIR +from castle_core.secret_backends import build_backend router = APIRouter(prefix="/secrets", tags=["secrets"]) +def _backend(): + return build_backend(SECRETS_DIR) + + class SecretValue(BaseModel): value: str @@ -17,31 +22,27 @@ class SecretValue(BaseModel): @router.get("") def list_secrets() -> list[str]: """List all secret names (not values).""" - if not SECRETS_DIR.exists(): - return [] - return sorted(f.name for f in SECRETS_DIR.iterdir() if f.is_file()) + return _backend().list_names() @router.get("/{name}") def get_secret(name: str) -> dict: """Get a secret value.""" _validate_name(name) - path = SECRETS_DIR / name - if not path.exists(): + value = _backend().read(name) + if value is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Secret '{name}' not found", ) - return {"name": name, "value": path.read_text().strip()} + return {"name": name, "value": value} @router.put("/{name}") def set_secret(name: str, body: SecretValue) -> dict: """Set a secret value.""" _validate_name(name) - SECRETS_DIR.mkdir(parents=True, exist_ok=True) - path = SECRETS_DIR / name - path.write_text(body.value.strip() + "\n") + _backend().write(name, body.value) return {"name": name, "ok": True} @@ -49,9 +50,7 @@ def set_secret(name: str, body: SecretValue) -> dict: def delete_secret(name: str) -> dict: """Delete a secret.""" _validate_name(name) - path = SECRETS_DIR / name - if path.exists(): - path.unlink() + _backend().delete(name) return {"name": name, "ok": True} diff --git a/core/src/castle_core/generators/systemd.py b/core/src/castle_core/generators/systemd.py index 10d437e..d2bfa3a 100644 --- a/core/src/castle_core/generators/systemd.py +++ b/core/src/castle_core/generators/systemd.py @@ -174,6 +174,15 @@ SuccessExitStatus=143 reload_argv[0] = resolved_reload unit += f"ExecReload={' '.join(reload_argv)}\n" + # Post-start hooks (e.g. OpenBao auto-unseal). `-` prefix → failure is ignored, + # so a hiccup in the hook never fails the unit. + for cmd in (sd.exec_start_post if sd else []): + argv = cmd.split() + resolved = shutil.which(argv[0]) + if resolved: + argv[0] = resolved + unit += f"ExecStartPost=-{' '.join(argv)}\n" + if sd and sd.no_new_privileges: unit += "NoNewPrivileges=true\n" diff --git a/core/src/castle_core/manifest.py b/core/src/castle_core/manifest.py index 06adc39..07aa60b 100644 --- a/core/src/castle_core/manifest.py +++ b/core/src/castle_core/manifest.py @@ -160,6 +160,9 @@ class SystemdSpec(BaseModel): no_new_privileges: bool = True readiness: ReadinessHttpGet | None = None exec_reload: str | None = None + # Commands run after the main process starts (systemd ``ExecStartPost=``), one + # line each — e.g. an OpenBao auto-unseal step. Failures don't fail the unit. + exec_start_post: list[str] = Field(default_factory=list) class ManageSpec(BaseModel): diff --git a/core/src/castle_core/secret_backends.py b/core/src/castle_core/secret_backends.py index 377aad0..14e18f1 100644 --- a/core/src/castle_core/secret_backends.py +++ b/core/src/castle_core/secret_backends.py @@ -26,10 +26,13 @@ from typing import Protocol class SecretBackend(Protocol): def read(self, name: str) -> str | None: ... + def write(self, name: str, value: str) -> None: ... + def delete(self, name: str) -> None: ... + def list_names(self) -> list[str]: ... class FileSecretBackend: - """Reads ``/`` (the historical behavior).""" + """Reads/writes ``/`` (the historical behavior).""" def __init__(self, secrets_dir: Path) -> None: self._dir = secrets_dir @@ -40,6 +43,20 @@ class FileSecretBackend: return path.read_text().strip() return None + def write(self, name: str, value: str) -> None: + self._dir.mkdir(parents=True, exist_ok=True) + (self._dir / name).write_text(value.strip() + "\n") + + def delete(self, name: str) -> None: + path = self._dir / name + if path.exists(): + path.unlink() + + def list_names(self) -> list[str]: + if not self._dir.exists(): + return [] + return sorted(f.name for f in self._dir.iterdir() if f.is_file()) + class OpenBaoBackend: """Reads from an OpenBao/Vault KV-v2 mount; falls back to ``fallback``. @@ -66,14 +83,44 @@ class OpenBaoBackend: if not self._token: return None url = f"{self._addr}/v1/{self._mount}/data/{name}" - req = urllib.request.Request(url, headers={"X-Vault-Token": self._token}) try: - with urllib.request.urlopen(req, timeout=5) as resp: # noqa: S310 - data = json.load(resp) + data = self._request("GET", url) return data["data"]["data"].get("value") except Exception: return None + def write(self, name: str, value: str) -> None: + url = f"{self._addr}/v1/{self._mount}/data/{name}" + self._request("POST", url, {"data": {"value": value.strip()}}) + + def delete(self, name: str) -> None: + # Remove all versions (metadata delete), matching file-backend semantics. + url = f"{self._addr}/v1/{self._mount}/metadata/{name}" + self._request("DELETE", url) + + def list_names(self) -> list[str]: + url = f"{self._addr}/v1/{self._mount}/metadata?list=true" + try: + data = self._request("GET", url) + return sorted(data.get("data", {}).get("keys", [])) + except Exception: + return [] + + def _request(self, method: str, url: str, body: dict | None = None) -> dict: + payload = json.dumps(body).encode() if body is not None else None + req = urllib.request.Request( # noqa: S310 + url, + data=payload, + method=method, + headers={ + "X-Vault-Token": self._token, + "Content-Type": "application/json", + }, + ) + with urllib.request.urlopen(req, timeout=5) as resp: # noqa: S310 + raw = resp.read() + return json.loads(raw) if raw else {} + def build_backend(secrets_dir: Path) -> SecretBackend: """Construct the active secret backend from the environment.""" diff --git a/core/tests/test_secret_backends.py b/core/tests/test_secret_backends.py index 855dcee..c9b945b 100644 --- a/core/tests/test_secret_backends.py +++ b/core/tests/test_secret_backends.py @@ -20,6 +20,19 @@ def test_file_backend_read_miss(tmp_path: Path) -> None: assert FileSecretBackend(tmp_path).read("ABSENT") is None +def test_file_backend_write_read_list_delete(tmp_path: Path) -> None: + b = FileSecretBackend(tmp_path) + assert b.list_names() == [] + b.write("A", "one") + b.write("B", "two") + assert b.read("A") == "one" + assert b.list_names() == ["A", "B"] + b.delete("A") + assert b.read("A") is None + assert b.list_names() == ["B"] + b.delete("ABSENT") # no error + + def test_build_backend_defaults_to_file(tmp_path: Path, monkeypatch) -> None: monkeypatch.delenv("CASTLE_SECRET_BACKEND", raising=False) assert isinstance(build_backend(tmp_path), FileSecretBackend) diff --git a/docs/fleet-mesh-plan.md b/docs/fleet-mesh-plan.md index 7c433ca..12eeea5 100644 --- a/docs/fleet-mesh-plan.md +++ b/docs/fleet-mesh-plan.md @@ -236,16 +236,21 @@ second node is proven on NATS. - **Tests:** `core/tests/test_secret_backends.py` (file hit/miss, backend selection, unreachable-vault + empty-token fallback). Suites: **206 core + 93 api**. -**Remaining for full OpenBao production use (documented, not blocking — nothing -uses the backend until `CASTLE_SECRET_BACKEND=openbao`):** -1. **Write path** — `castle-api/secrets.py` CRUD still writes to files; in openbao - mode dashboard-set secrets land in file (resolve via fallback) rather than the - vault. Add `write`/`delete` to the backends + route the API writer through them. -2. **Auto-unseal on boot** (Phase 0 deferral) — OpenBao re-seals on container - restart. Wire an unseal step (manifest `exec_start_post` or a companion - oneshot reading `OPENBAO_UNSEAL_KEY`). Not urgent while the backend is unused. -3. **TLS hardening** — NATS + OpenBao are plaintext localhost. Required before - cross-network or moving real secrets: NATS mTLS + auth, OpenBao TLS listener. +**Hardening:** +1. **Write path — DONE + verified live.** `SecretBackend` gained + `write`/`delete`/`list_names`; `castle-api/secrets.py` routes all CRUD through + the active backend, so in openbao mode the dashboard writes to the vault. + Verified: write→vault→read→list→delete round-trip against live OpenBao. +2. **Auto-unseal on boot — DONE + verified.** Added `exec_start_post` to the + systemd spec (general, `-`-prefixed so a hook failure never fails the unit); + `castle-openbao` runs `/data/castle/castle-openbao/unseal.sh` (polls up, unseals + with `OPENBAO_UNSEAL_KEY`). Verified: manual seal → restart → auto-unsealed. +3. **TLS hardening — REMAINING (deliberately deferred).** NATS + OpenBao are + plaintext on the trusted LAN. This is the gate before cross-*network* or moving + real secrets — neither of which is here yet (the vault holds no production + secrets). Doing NATS TLS/auth touches the *live* civil↔primer mesh, so it's a + deliberate, staged change (regenerate certs, update both nodes' clients), not a + tail-end one. Next dedicated step. ### Phase 3 — cross-node routing + breaker: logic DONE + verified against a real peer (2026-07-07)