feat(secrets): OpenBao write path + auto-unseal on boot (Phase 4 hardening)

- 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

@@ -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}
 
 

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"
 

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):

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 ``<secrets_dir>/<name>`` (the historical behavior)."""
+    """Reads/writes ``<secrets_dir>/<name>`` (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."""

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)

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)