Coverage for node / src / stigmem_node / main.py: 84%
228 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Stigmem reference node — FastAPI application factory and entrypoint."""
3from __future__ import annotations
5import asyncio
6import logging
7import signal
8import ssl
9from collections.abc import AsyncGenerator, Awaitable, Callable
10from contextlib import asynccontextmanager, suppress
11from pathlib import Path
12from typing import Annotated, Any, cast
14import uvicorn
15from fastapi import Depends, FastAPI, HTTPException, Request, Response, status
16from fastapi.middleware.cors import CORSMiddleware
17from fastapi.responses import FileResponse, JSONResponse
19from .auth import Identity, resolve_identity, resolve_identity_optional
20from .body_limit import BodySizeLimitMiddleware
21from .db import apply_migrations
22from .net_util import node_url_is_loopback
23from .rate_limit import RateLimitMiddleware
24from .routes.admin_audit import router as admin_audit_router
25from .routes.agent_keys import router as agent_keys_router
26from .routes.aliases import router as aliases_router
27from .routes.audit import router as audit_router
28from .routes.auth import router as auth_router
29from .routes.cards import router as cards_router
30from .routes.cid_admin import router as cid_admin_router
31from .routes.decay import router as decay_router
32from .routes.facts import router as facts_router
33from .routes.federation import router as federation_router
34from .routes.gardens import router as gardens_router
35from .routes.graph import router as graph_router
36from .routes.identity import router as identity_router
37from .routes.intents import router as intents_router
38from .routes.lint import router as lint_router
39from .routes.mcp import router as mcp_router
40from .routes.quarantine import router as quarantine_router
41from .routes.recall import router as recall_router
42from .routes.resolver import router as resolver_router
43from .routes.subscriptions import router as subscriptions_router
44from .routes.synthesize import router as synthesize_router
45from .routes.tombstones import router as tombstones_router
46from .routes.wellknown import router as wellknown_router
47from .settings import settings
49_STATIC_DIR = Path(__file__).parent / "static"
51logger = logging.getLogger("stigmem")
53_DEV_LOCALHOST_CORS_REGEX = r"^https?://(localhost|127\.0\.0\.1)(:\d+)?$"
56def _enforce_federation_transport_security() -> None:
57 """Require explicit opt-in for federation without mTLS."""
58 federation_active = settings.federation_enabled or settings.federation_push_enabled
59 if not federation_active or settings.mtls_enabled:
60 return
62 if not settings.federation_insecure:
63 raise RuntimeError(
64 "Federation requires mTLS by default. Configure STIGMEM_TLS_CERT_PATH, "
65 "STIGMEM_TLS_KEY_PATH, and STIGMEM_TLS_CA_BUNDLE, or set "
66 "STIGMEM_FEDERATION_INSECURE=1 only for local/dev/test federation."
67 )
69 if not _node_url_is_loopback(settings.node_url) and not (
70 settings.local_dev_allow_insecure_non_loopback
71 ):
72 raise RuntimeError(
73 "STIGMEM_FEDERATION_INSECURE=1 is only permitted when node_url is "
74 f"bound to 127.0.0.1 or localhost. Got node_url={settings.node_url!r}. "
75 "Configure mTLS for any non-loopback deployment, or set "
76 "STIGMEM_LOCAL_DEV_ALLOW_INSECURE_NON_LOOPBACK=1 only for local "
77 "Docker/dev networks."
78 )
80 logger.warning(
81 "SECURITY WARNING: federation is running without mTLS because "
82 "STIGMEM_FEDERATION_INSECURE=1 is set. This is only allowed because "
83 "node_url is a loopback address or "
84 "STIGMEM_LOCAL_DEV_ALLOW_INSECURE_NON_LOOPBACK=1 is set. Use this only "
85 "for local/dev/test."
86 )
89def _enforce_auth_required_in_production() -> None:
90 """Refuse to run unauthenticated outside loopback."""
91 if settings.auth_required:
92 return
93 if not _node_url_is_loopback(settings.node_url) and not (
94 settings.local_dev_allow_insecure_non_loopback
95 ):
96 raise RuntimeError(
97 "STIGMEM_AUTH_REQUIRED=false is only permitted when node_url is "
98 f"bound to 127.0.0.1 or localhost. Got node_url={settings.node_url!r}. "
99 "Anonymous identity has read/write/federate permissions; never expose "
100 "this configuration to a network. Set "
101 "STIGMEM_LOCAL_DEV_ALLOW_INSECURE_NON_LOOPBACK=1 only for local "
102 "Docker/dev networks."
103 )
104 logger.warning(
105 "SECURITY WARNING: STIGMEM_AUTH_REQUIRED=false. Anonymous identity has "
106 "full read/write/federate permissions. This is only allowed because "
107 "node_url is a loopback address or "
108 "STIGMEM_LOCAL_DEV_ALLOW_INSECURE_NON_LOOPBACK=1 is set."
109 )
112def _enforce_rate_limit_kill_switch_ack() -> None:
113 """Refuse boot when quota is fully disabled without an explicit acknowledgement."""
114 if settings.rate_limit_write_per_hour != 0 or settings.rate_limit_read_per_hour != 0:
115 return
116 if not settings.rate_limit_disabled_ack:
117 raise RuntimeError(
118 "STIGMEM_RATE_LIMIT_WRITE_PER_HOUR=0 and "
119 "STIGMEM_RATE_LIMIT_READ_PER_HOUR=0 fully disable quota enforcement. "
120 "To proceed, set STIGMEM_RATE_LIMIT_DISABLED_ACK=1 to acknowledge "
121 "that this node accepts unbounded read and write traffic."
122 )
124 logger.warning(
125 "SECURITY WARNING: quota enforcement is fully disabled "
126 "(write=0, read=0) with explicit operator acknowledgment via "
127 "STIGMEM_RATE_LIMIT_DISABLED_ACK=1."
128 )
131def _warn_if_cors_dev_localhost_enabled() -> None:
132 """Log the expanded development CORS posture at startup."""
133 if settings.cors_dev_localhost:
134 logger.warning(
135 "SECURITY WARNING: STIGMEM_CORS_DEV_LOCALHOST=1 enables browser "
136 "access from localhost and loopback origins. Use this only for "
137 "local development."
138 )
141def _warn_if_backend_immutability_unenforced() -> None:
142 """Warn when the storage backend lacks DB-level facts-immutability triggers.
144 P-DESTROY-1 (honesty): the append-only no-UPDATE/no-DELETE triggers
145 (ADR-016 L2) are implemented for SQLite only. On libsql/Postgres the facts
146 table is mutable by anyone with direct database write access, so tamper
147 resistance on those backends rests on the L3+ CID / hash-chain layers, not
148 L2 triggers. State that loudly rather than imply uniform immutability.
149 """
150 backend = getattr(settings, "storage_backend", "sqlite")
151 if backend == "sqlite":
152 return
153 logger.warning(
154 "SECURITY WARNING: storage_backend=%r does NOT enforce database-level "
155 "facts immutability. The append-only no-UPDATE/no-DELETE triggers "
156 "(ADR-016 L2) exist for SQLite only; on %s the facts table is mutable "
157 "by anyone with direct DB write access. Tamper-evidence on this backend "
158 "relies on the L3+ CID / hash-chain layers, not L2 triggers.",
159 backend,
160 backend,
161 )
164def _warn_if_legacy_sha256_acceptance_unbounded() -> None:
165 """Warn when legacy SHA-256 API-key hashes are accepted with no cutoff (F-ID-4).
167 Unsalted SHA-256 is weaker than the Argon2id default. The acceptance window
168 is a v0.9.x migration affordance; leaving it open indefinitely
169 (STIGMEM_LEGACY_SHA256_ACCEPT_UNTIL unset) is a standing weakness.
170 """
171 if settings.legacy_sha256_accept_until is not None:
172 return
173 logger.warning(
174 "SECURITY WARNING: legacy SHA-256 API-key hashes are accepted with no "
175 "cutoff (STIGMEM_LEGACY_SHA256_ACCEPT_UNTIL is unset). Unsalted SHA-256 "
176 "is weaker than the Argon2id default; set a cutoff date to bound the "
177 "migration window and force rotation of any keys not yet rehashed."
178 )
181def _node_url_is_loopback(node_url: str) -> bool:
182 """Return True iff node_url's host is a loopback address."""
183 return node_url_is_loopback(node_url)
186def create_app() -> FastAPI:
187 @asynccontextmanager
188 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
189 from .plugins import get_registry, register_discovered_plugins
191 if settings.trust_mode == "strict" and not settings.node_private_key: 191 ↛ 192line 191 didn't jump to line 192 because the condition on line 191 was never true
192 raise RuntimeError("STIGMEM_NODE_PRIVATE_KEY must be set when trust_mode=strict")
193 _enforce_federation_transport_security()
194 _enforce_auth_required_in_production()
195 _enforce_rate_limit_kill_switch_ack()
196 _warn_if_cors_dev_localhost_enabled()
197 _warn_if_backend_immutability_unenforced()
198 _warn_if_legacy_sha256_acceptance_unbounded()
200 discovered_plugins = register_discovered_plugins(freeze=False)
201 _include_plugin_routers(app, discovered_plugins)
202 apply_migrations()
203 from .memory_garden_acl_gate import warn_if_memory_garden_acl_filtering_disabled
205 warn_if_memory_garden_acl_filtering_disabled(logger)
206 get_registry().freeze()
208 if settings.otel_enabled: 208 ↛ 209line 208 didn't jump to line 209 because the condition on line 208 was never true
209 from .observability.tracing import init_tracing
211 init_tracing(
212 service_name=settings.otel_service_name,
213 otlp_endpoint=settings.otel_exporter_otlp_endpoint,
214 )
216 pull_task: asyncio.Task[None] | None = None
217 if settings.federation_enabled:
218 from .federation.federation_pull import pull_loop_task
219 from .federation.peer_token import init_federation_keys
221 init_federation_keys()
222 pull_task = asyncio.create_task(pull_loop_task())
223 logger.info("Federation enabled — pull %ds", settings.federation_pull_interval_s)
225 from .subscription_delivery import sweep_loop as _sub_sweep_loop
227 sweep_task: asyncio.Task[None] = asyncio.create_task(_sub_sweep_loop())
228 logger.info(
229 "Stigmem subscription sweep enabled — interval %ds",
230 settings.subscription_delivery_sweep_s,
231 )
233 logger.info(
234 "Stigmem node ready — db=%s auth=%s federation=%s",
235 settings.db_path,
236 settings.auth_required,
237 "enabled" if settings.federation_enabled else "disabled",
238 )
239 yield
241 sweep_task.cancel()
242 with suppress(asyncio.CancelledError):
243 _unused_result = await cast("asyncio.Task[object]", sweep_task)
245 if pull_task is not None:
246 pull_task.cancel()
247 with suppress(asyncio.CancelledError):
248 _unused_result = await cast("asyncio.Task[object]", pull_task)
250 app = FastAPI(
251 title="Stigmem Reference Node",
252 version="0.9.0a12",
253 description=(
254 "Reference node implementing the Stigmem v0.9.0a12 HTTP API — facts, federation, "
255 "gardens, recall, subscriptions, audit, identity, content-addressed fact IDs. "
256 "Experimental cross-cutting behavior remains opt-in where plugin-gated; tombstone "
257 "admin and federation route contracts are mounted but access-gated."
258 ),
259 license_info={"name": "Apache-2.0", "url": "https://www.apache.org/licenses/LICENSE-2.0"},
260 lifespan=lifespan,
261 )
263 app.add_middleware(RateLimitMiddleware)
264 app.add_middleware(BodySizeLimitMiddleware)
265 _cors_regex = settings.cors_allowed_origin_regex
266 if settings.cors_dev_localhost:
267 _cors_regex = _DEV_LOCALHOST_CORS_REGEX
268 if settings.cors_allowed_origins or _cors_regex:
269 app.add_middleware(
270 CORSMiddleware,
271 allow_origins=settings.cors_allowed_origins,
272 allow_origin_regex=_cors_regex,
273 allow_credentials=settings.cors_allow_credentials,
274 allow_methods=["GET", "POST", "PATCH", "DELETE", "OPTIONS"],
275 allow_headers=["*"],
276 expose_headers=["ETag"],
277 max_age=600,
278 )
280 @app.middleware("http")
281 async def unsigned_plugin_override_warning(
282 _request: Request,
283 call_next: Callable[[Request], Awaitable[Response]],
284 ) -> Response:
285 """Warn every request while development unsigned-plugin override is active."""
286 from .plugins import get_registry
288 unsigned_plugins = get_registry().development_unsigned_plugins()
289 if unsigned_plugins: 289 ↛ 290line 289 didn't jump to line 290 because the condition on line 289 was never true
290 logger.warning(
291 "SECURITY WARNING: unsigned plugins active via "
292 "STIGMEM_PLUGIN_SIGNING_REQUIRED=false: %s",
293 ", ".join(unsigned_plugins),
294 )
295 return await call_next(_request)
297 if settings.mtls_enabled:
299 @app.middleware("http")
300 async def mtls_plaintext_guard(
301 request: Request,
302 call_next: Callable[[Request], Awaitable[Response]],
303 ) -> Response:
304 """Reject plaintext federation requests when mTLS is configured (§22.1)."""
305 if request.method == "OPTIONS" and not request.url.path.startswith("/v1/federation"):
306 return await call_next(request)
307 if request.url.path.startswith("/v1/federation") and request.url.scheme != "https":
308 return JSONResponse(
309 {
310 "error": "mTLS required",
311 "detail": "Federation transport requires mutual TLS (spec §22.1). "
312 "Connect via HTTPS with a valid node certificate.",
313 },
314 status_code=421,
315 )
316 return await call_next(request)
318 app.include_router(admin_audit_router)
319 app.include_router(cid_admin_router)
320 app.include_router(auth_router)
321 app.include_router(agent_keys_router)
322 app.include_router(audit_router)
323 app.include_router(facts_router)
324 app.include_router(gardens_router)
325 app.include_router(graph_router)
326 app.include_router(identity_router)
327 app.include_router(intents_router)
328 app.include_router(federation_router)
329 app.include_router(quarantine_router)
330 app.include_router(lint_router)
331 app.include_router(synthesize_router)
332 app.include_router(decay_router)
333 app.include_router(aliases_router)
334 app.include_router(resolver_router)
335 app.include_router(cards_router)
336 app.include_router(recall_router)
337 app.include_router(subscriptions_router)
338 app.include_router(tombstones_router)
339 app.include_router(mcp_router)
340 app.include_router(wellknown_router)
342 @app.get("/healthz", tags=["ops"])
343 def health() -> dict[str, str]:
344 return {"status": "ok"}
346 @app.get("/v1/doctor", tags=["ops"])
347 def doctor() -> dict[str, str]:
348 """Return coarse node health and operator posture.
350 This endpoint is unauthenticated in v0.9.0a12. The garden ACL posture
351 field is accepted as ops-endpoint disclosure and intentionally avoids
352 garden names, membership rows, tenant identifiers, or policy subjects.
353 """
354 from .memory_garden_acl_gate import memory_garden_acl_filtering_state
356 return {
357 "status": "ok",
358 "memory_garden_acl_filtering": memory_garden_acl_filtering_state(),
359 }
361 @app.get("/metrics", include_in_schema=False, tags=["ops"])
362 def prometheus_metrics(
363 identity: Annotated[Identity | None, Depends(resolve_identity_optional)],
364 ) -> Response:
365 # M11 / F-AVAIL-1: /metrics leaks per-principal labels + enables cardinality
366 # DoS, so require the admin capability by default. Set
367 # STIGMEM_METRICS_REQUIRE_AUTH=false only where /metrics is otherwise
368 # access-controlled (e.g. a private scrape interface or sidecar proxy).
369 if settings.metrics_require_auth:
370 if identity is None:
371 raise HTTPException(
372 status_code=status.HTTP_401_UNAUTHORIZED,
373 detail="Authorization required to scrape /metrics",
374 headers={"WWW-Authenticate": "Bearer"},
375 )
376 if not identity.is_admin():
377 raise HTTPException(
378 status_code=status.HTTP_403_FORBIDDEN,
379 detail="admin capability required to scrape /metrics",
380 )
382 from .observability.metrics import make_metrics_response
384 resp = make_metrics_response()
385 if resp is None: 385 ↛ 389line 385 didn't jump to line 389 because the condition on line 385 was always true
386 from fastapi.responses import PlainTextResponse
388 return PlainTextResponse("# prometheus_client not installed\n", status_code=200)
389 return resp
391 @app.get("/v1/me", tags=["auth"])
392 def whoami(identity: Annotated[Identity, Depends(resolve_identity)]) -> dict[str, Any]:
393 return {
394 "entity_uri": identity.entity_uri,
395 "permissions": sorted(identity.permissions),
396 "oidc_sub": identity.oidc_sub,
397 "tenant_id": identity.tenant_id,
398 }
400 @app.get("/ui", include_in_schema=False)
401 def ui_index() -> FileResponse:
402 return FileResponse(_STATIC_DIR / "index.html", media_type="text/html")
404 return app
407def _include_plugin_routers(app: FastAPI, discovered_plugins: tuple[Any, ...]) -> None:
408 """Include routers declared by installed plugins once per app instance."""
409 if getattr(app.state, "stigmem_plugin_routes_included", False):
410 return
411 for plugin in discovered_plugins:
412 for router in plugin.manifest.routes:
413 app.include_router(router)
414 app.state.stigmem_plugin_routes_included = True
417app = create_app()
420def run() -> None:
421 if not settings.mtls_enabled:
422 uvicorn.run(
423 "stigmem_node.main:app",
424 host=settings.host,
425 port=settings.port,
426 log_level=settings.log_level,
427 reload=False,
428 )
429 return
431 from .federation.tls import cert_watcher_task, reload_tls_cert
433 # Let uvicorn build the SSL context from cert/key files, then enforce TLS 1.3
434 # floor and mTLS client-cert requirement on the resulting context object.
435 config = uvicorn.Config(
436 "stigmem_node.main:app",
437 host=settings.host,
438 port=settings.port,
439 log_level=settings.log_level,
440 reload=False,
441 ssl_certfile=settings.tls_cert_path,
442 ssl_keyfile=settings.tls_key_path,
443 ssl_ca_certs=settings.tls_ca_bundle or None,
444 ssl_cert_reqs=ssl.CERT_REQUIRED,
445 )
446 config.load()
448 if config.ssl:
449 config.ssl.minimum_version = ssl.TLSVersion.TLSv1_3
450 ssl_ctx = config.ssl
452 async def _serve_with_cert_watcher() -> None:
453 loop = asyncio.get_running_loop()
454 if ssl_ctx is not None:
455 loop.add_signal_handler(
456 signal.SIGHUP,
457 lambda: reload_tls_cert(ssl_ctx),
458 )
460 server = uvicorn.Server(config)
461 watcher_task: asyncio.Task[None] | None = None
462 if ssl_ctx is not None:
463 watcher_task = asyncio.create_task(cert_watcher_task(ssl_ctx))
465 try:
466 await server.serve()
467 finally:
468 if watcher_task is not None:
469 watcher_task.cancel()
470 with suppress(asyncio.CancelledError):
471 _unused_result = await cast("asyncio.Task[object]", watcher_task)
473 asyncio.run(_serve_with_cert_watcher())
476if __name__ == "__main__": 476 ↛ 477line 476 didn't jump to line 477 because the condition on line 476 was never true
477 run()