From 5415e9845838689be9e042f590e62c8b885c2dda Mon Sep 17 00:00:00 2001 From: anti Date: Mon, 27 Apr 2026 21:26:03 -0400 Subject: [PATCH] sec(api): mode-gate and eager-load JWT secret in lifespan MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refuse to start decnet.web.api when DECNET_MODE=agent (unless the operator explicitly opts into dual-role with DECNET_DISALLOW_MASTER= false). The Typer CLI already hides master-only commands on agents, but a misconfigured systemd unit or a direct uvicorn invocation would bypass that — now the lifespan itself refuses, before any worker, DB or bus comes up. Resolve DECNET_JWT_SECRET eagerly at startup so a missing or known- bad value fails at boot rather than on the first auth-gated request. The lazy-load shape stays useful for non-master CLIs. --- decnet/web/api.py | 19 +++++ tests/web/test_api_startup_guards.py | 122 +++++++++++++++++++++++++++ 2 files changed, 141 insertions(+) create mode 100644 tests/web/test_api_startup_guards.py diff --git a/decnet/web/api.py b/decnet/web/api.py index 82a40e06..d446871c 100644 --- a/decnet/web/api.py +++ b/decnet/web/api.py @@ -75,6 +75,25 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: # Raises ValueError with an actionable message; uvicorn surfaces it. validate_public_binding() + # Defence-in-depth on top of the CLI mode gating. Typer hides master-only + # commands when DECNET_MODE=agent, but a misconfigured systemd unit or + # a direct `python -m uvicorn decnet.web.api:app` call would bypass that. + # This raises before any worker / DB / bus comes up. + _mode = os.environ.get("DECNET_MODE", "master").lower() + _disallow = os.environ.get("DECNET_DISALLOW_MASTER", "true").lower() == "true" + if _mode == "agent" and _disallow: + raise RuntimeError( + "decnet.web.api refuses to start with DECNET_MODE=agent. " + "The master API is master-only; agents run `decnet agent` instead. " + "If this host genuinely plays both roles, set DECNET_DISALLOW_MASTER=false." + ) + + # Resolve DECNET_JWT_SECRET eagerly so a missing/insecure secret fails + # at boot rather than on the first request that hits an auth-gated + # endpoint. The lazy-load shape stays useful for non-master CLIs. + from decnet import env as _env + _ = _env.DECNET_JWT_SECRET # raises ValueError on missing/bad + log.info("API startup initialising database") for attempt in range(1, 6): try: diff --git a/tests/web/test_api_startup_guards.py b/tests/web/test_api_startup_guards.py new file mode 100644 index 00000000..f6f9d17f --- /dev/null +++ b/tests/web/test_api_startup_guards.py @@ -0,0 +1,122 @@ +"""Master API startup guards: mode gating + eager JWT load. + +The lifespan is what enforces these. We invoke it directly with a fresh +FastAPI app instance rather than spinning up a TestClient — TestClient +fixtures elsewhere set DECNET_JWT_SECRET globally and would mask the +"missing secret fails at boot" assertion. +""" +from __future__ import annotations + +import asyncio +import importlib +import sys + +import pytest +from fastapi import FastAPI + + +def _reload_api(monkeypatch: pytest.MonkeyPatch): + for mod in list(sys.modules): + if mod == "decnet.env" or mod == "decnet.web.api" or mod.startswith("decnet.env."): + sys.modules.pop(mod) + return importlib.import_module("decnet.web.api") + + +def _strip_pytest_vars(monkeypatch: pytest.MonkeyPatch) -> None: + import os + for k in list(os.environ): + if k.startswith("PYTEST"): + monkeypatch.delenv(k, raising=False) + + +async def _run_lifespan_startup(api_mod) -> None: + """Run the lifespan up to (but not past) yield, then unwind cleanly.""" + app = FastAPI() + cm = api_mod.lifespan(app) + await cm.__aenter__() + try: + return + finally: + try: + await cm.__aexit__(None, None, None) + except Exception: + pass + + +def test_master_api_refuses_to_start_in_agent_mode( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("DECNET_MODE", "agent") + monkeypatch.setenv("DECNET_DISALLOW_MASTER", "true") + monkeypatch.setenv("DECNET_JWT_SECRET", "x" * 32) + api = _reload_api(monkeypatch) + with pytest.raises(RuntimeError, match="master-only"): + asyncio.get_event_loop_policy().new_event_loop().run_until_complete( + _run_lifespan_startup(api) + ) + + +def test_master_api_starts_when_dual_role_enabled( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """DECNET_DISALLOW_MASTER=false is the documented escape hatch for + dev hosts that play both sides — must not trip the gate.""" + monkeypatch.setenv("DECNET_MODE", "agent") + monkeypatch.setenv("DECNET_DISALLOW_MASTER", "false") + monkeypatch.setenv("DECNET_JWT_SECRET", "x" * 32) + api = _reload_api(monkeypatch) + # Reaching the DB init phase means the gate passed; we don't need to + # actually finish startup. Cancel via a synthetic exception that the + # lifespan doesn't catch. + # Reaching repo.initialize means the gate passed. We don't actually + # need DB to come up — short-circuit and assert no master-only raise. + seen: list[str] = [] + + async def _spy(*_a, **_kw): + seen.append("init_called") + + monkeypatch.setattr(api.repo, "initialize", _spy) + asyncio.get_event_loop_policy().new_event_loop().run_until_complete( + _run_lifespan_startup(api) + ) + assert seen == ["init_called"], "DB init should have been reached, gate must be inert" + + +def test_master_api_eager_loads_jwt_secret_at_startup( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The lifespan must touch DECNET_JWT_SECRET so a missing/insecure + value fails at boot rather than on the first auth-gated request. + + We can't realistically exercise the raise-on-missing path in this + repo: dev hosts have a populated .env.local that dotenv auto-loads, + and conftest seeds a JWT secret globally. The actual raise behaviour + is covered by tests/web/test_env_lazy_jwt.py — here we just assert + the lifespan calls into the env module's lazy resolver. + """ + monkeypatch.setenv("DECNET_MODE", "master") + monkeypatch.setenv("DECNET_JWT_SECRET", "y" * 32) + monkeypatch.setenv("DECNET_API_HOST", "127.0.0.1") + monkeypatch.delenv("DECNET_CORS_ORIGINS", raising=False) + api = _reload_api(monkeypatch) + import decnet.env as env_mod + + seen: list[str] = [] + real_getattr = env_mod.__getattr__ + + def _spy(name: str) -> str: + seen.append(name) + return real_getattr(name) + + monkeypatch.setattr(env_mod, "__getattr__", _spy, raising=False) + + async def _noop_init(*_a, **_kw) -> None: + return None + + monkeypatch.setattr(api.repo, "initialize", _noop_init) + asyncio.get_event_loop_policy().new_event_loop().run_until_complete( + _run_lifespan_startup(api) + ) + assert "DECNET_JWT_SECRET" in seen, ( + "lifespan must access env.DECNET_JWT_SECRET at startup" + )