sec(api): mode-gate and eager-load JWT secret in lifespan
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.
This commit is contained in:
122
tests/web/test_api_startup_guards.py
Normal file
122
tests/web/test_api_startup_guards.py
Normal file
@@ -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"
|
||||
)
|
||||
Reference in New Issue
Block a user