210 lines
11 KiB
Python
210 lines
11 KiB
Python
import os
|
|
import asyncio
|
|
from contextlib import asynccontextmanager
|
|
from fastapi import FastAPI, Depends
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from app.internal.logger import logger
|
|
from app.internal.mqtt_handler import mqtt_handler
|
|
from app.internal.node_sweeper import sweeper_loop
|
|
from app.internal.summarizer import summarizer_loop
|
|
from app.internal.vocabulary_learner import vocabulary_induction_loop
|
|
from app.internal.recorrelation_sweep import recorrelation_loop
|
|
from app.internal import ai_health
|
|
from app.internal import discord_streams
|
|
from app.config import settings
|
|
from app.internal.auth import (
|
|
require_firebase_token,
|
|
require_service_or_firebase_token,
|
|
require_node_service_or_firebase_token,
|
|
)
|
|
from app.routers import nodes, systems, calls, upload, tokens, incidents, alerts, admin, trips, places, links, users
|
|
from app.routers import enrollment, media, org, waitlist, telemetry, replay, traffic, cameras
|
|
from app.routers import outages, weather
|
|
from app.routers import discord as discord_router
|
|
from app.routers import discord_link as discord_link_router
|
|
from app.internal import dynsec
|
|
from app.internal import firestore as fstore
|
|
from app.routers import discord_web
|
|
from app.routers import discord_checks
|
|
from app.routers import discord_oauth
|
|
|
|
|
|
async def _clear_legacy_token_flags():
|
|
"""Tokens written by the pre-lease code carry ``in_use`` with no ``lease``.
|
|
|
|
Nothing reads ``in_use`` for decisions any more (discord_leases.py decides
|
|
on ``lease``), but the /tokens page shows it, so a legacy "in use" with no
|
|
lease behind it would be a permanent false label. A token that HAS a lease
|
|
is left alone: voice connections now belong to the edge and survive a C2
|
|
restart, so releasing leases at startup (what this used to do) would hand
|
|
a still-logged-in token to a second node.
|
|
"""
|
|
def _find():
|
|
from app.internal.firestore import db
|
|
return [d for d in db.collection("bot_tokens").where("in_use", "==", True).stream()]
|
|
|
|
results = await asyncio.to_thread(_find)
|
|
cleared = 0
|
|
for doc in results:
|
|
if (doc.to_dict() or {}).get("lease"):
|
|
continue
|
|
await fstore.doc_update("bot_tokens", doc.id, {
|
|
"lease": None, "in_use": False, "assigned_node_id": None, "assigned_at": None,
|
|
})
|
|
cleared += 1
|
|
if cleared:
|
|
logger.info(f"Cleared legacy in_use flag on {cleared} token(s) with no lease.")
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
logger.info("DRB C2 Core starting.")
|
|
await _clear_legacy_token_flags()
|
|
|
|
# dynsec bootstrap + reconcile — must happen before mqtt_handler.connect()
|
|
# so that by the time the app is serving requests, c2-core's own dynsec
|
|
# client/roles exist and every already-approved node's dynsec client
|
|
# matches Firestore (see app/internal/dynsec.py "TWO-SOURCES-OF-TRUTH").
|
|
# Non-fatal by design: if the broker or MQTT_DYNSEC_ADMIN_PASS isn't
|
|
# reachable/configured yet (e.g. first-ever deploy, mosquitto still
|
|
# starting), log loudly and keep booting rather than crash-looping
|
|
# c2-core itself — mqtt_handler.connect() below has its own retry loop
|
|
# and node approval/reissue endpoints fail loudly on their own if dynsec
|
|
# calls fail later, so nothing here is silently swallowed forever.
|
|
try:
|
|
await dynsec.ensure_roles_and_c2core_grant()
|
|
await dynsec.reconcile_all()
|
|
except dynsec.DynsecError as e:
|
|
logger.error(f"dynsec bootstrap/reconcile failed — node approval/reissue will fail until this is resolved: {e}")
|
|
|
|
await mqtt_handler.connect()
|
|
sweeper_task = asyncio.create_task(sweeper_loop())
|
|
summarizer_task = asyncio.create_task(summarizer_loop())
|
|
induction_task = asyncio.create_task(vocabulary_induction_loop())
|
|
recorrelation_task = asyncio.create_task(recorrelation_loop())
|
|
discord_task = asyncio.create_task(discord_streams.reconcile_loop())
|
|
|
|
yield # --- app running ---
|
|
|
|
logger.info("DRB C2 Core shutting down.")
|
|
sweeper_task.cancel()
|
|
summarizer_task.cancel()
|
|
induction_task.cancel()
|
|
recorrelation_task.cancel()
|
|
discord_task.cancel()
|
|
await mqtt_handler.disconnect()
|
|
|
|
|
|
app = FastAPI(title="DRB C2 Core", lifespan=lifespan)
|
|
|
|
# The browser needs CORS to reach this API at all: the frontend's Archive page
|
|
# calls GET /calls/search with Authorization + Content-Type headers, which
|
|
# forces a preflight. Without this middleware the OPTIONS gets a bare 405 and
|
|
# the fetch fails (#110). allow_origins is an explicit list -- never "*" in a
|
|
# deployment -- so name every host the frontend is served from in CORS_ORIGINS.
|
|
#
|
|
# allow_credentials stays False on purpose: auth here is a Bearer header, not a
|
|
# cookie, so credentialed CORS is never needed, and keeping it False is what
|
|
# lets an explicit-origin allowlist work without Starlette's "*"-only
|
|
# restriction. "*" + credentials is the dangerous pair (Starlette reflects the
|
|
# caller's Origin back WITH Access-Control-Allow-Credentials: true); this code
|
|
# cannot produce it because credentials are hard-off.
|
|
def cors_allows_credentials(origins: list[str]) -> bool:
|
|
"""Always False -- credentialed CORS is never enabled here (Bearer auth,
|
|
not cookies). Kept as a named predicate so a future edit that wants to
|
|
turn credentials on has to go through here and confront the "*" case.
|
|
A wildcard entry would additionally be refused a credentialed response."""
|
|
return False
|
|
|
|
|
|
_cors_is_wildcard = "*" in settings.cors_origins
|
|
if _cors_is_wildcard:
|
|
logger.error(
|
|
"CORS_ORIGINS contains '*'. That is fine for local dev but is almost "
|
|
"certainly a misconfigured deployment -- set CORS_ORIGINS to your "
|
|
"frontend origin(s), e.g. [\"https://drb.cusano.net\"]."
|
|
)
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=settings.cors_origins,
|
|
allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
|
|
allow_headers=["authorization", "content-type"],
|
|
allow_credentials=False,
|
|
)
|
|
|
|
app.include_router(nodes.router, dependencies=[Depends(require_service_or_firebase_token)])
|
|
# systems is the one router edge nodes read directly (system_cacher.py builds
|
|
# the OP25 config from it), so its gate also accepts a per-node api_key. The
|
|
# write routes inside carry their own require_admin_token, so nodes get read
|
|
# access only.
|
|
app.include_router(systems.router, dependencies=[Depends(require_node_service_or_firebase_token)])
|
|
app.include_router(telemetry.router, dependencies=[Depends(require_node_service_or_firebase_token)])
|
|
app.include_router(calls.router, dependencies=[Depends(require_service_or_firebase_token)])
|
|
app.include_router(tokens.router, dependencies=[Depends(require_service_or_firebase_token)])
|
|
app.include_router(incidents.router, dependencies=[Depends(require_service_or_firebase_token)])
|
|
app.include_router(alerts.router, dependencies=[Depends(require_service_or_firebase_token)])
|
|
app.include_router(trips.router, dependencies=[Depends(require_service_or_firebase_token)])
|
|
app.include_router(places.router, dependencies=[Depends(require_service_or_firebase_token)])
|
|
app.include_router(traffic.router, dependencies=[Depends(require_service_or_firebase_token)])
|
|
app.include_router(cameras.router, dependencies=[Depends(require_service_or_firebase_token)])
|
|
app.include_router(cameras.admin_router) # auth: admin only, per endpoint (needs the principal for audit)
|
|
app.include_router(outages.router, dependencies=[Depends(require_service_or_firebase_token)])
|
|
app.include_router(outages.admin_router) # auth: admin only, per endpoint (needs the principal for audit)
|
|
app.include_router(weather.router, dependencies=[Depends(require_service_or_firebase_token)]) # auth: any signed-in user/service, like traffic
|
|
app.include_router(upload.router) # auth is per-node, handled inline
|
|
app.include_router(admin.router) # auth is per-endpoint (read: firebase, write: admin)
|
|
app.include_router(replay.router) # auth: admin only (every route spends or reads a replay run)
|
|
app.include_router(users.router) # auth: admin only
|
|
app.include_router(links.router) # auth is per-endpoint (generate: firebase, resolve: service key)
|
|
app.include_router(enrollment.router) # public; auth is the enrollment/pickup-secret tokens, checked inline
|
|
app.include_router(org.router) # auth is per-endpoint (read: firebase, write: org owner)
|
|
app.include_router(waitlist.router) # public — no auth, source-IP rate limited inline
|
|
# public by necessity — an <audio src> can't send a bearer token, so the
|
|
# short-lived HMAC in the URL is the credential. Checked inline in media.py.
|
|
app.include_router(media.router)
|
|
# auth is per-endpoint: directory push takes the service key; the node-page
|
|
# picker takes a Firebase token scoped to the node's org (server-26#213).
|
|
app.include_router(discord_web.router)
|
|
# auth is per-endpoint: bot routes take the service key (or a platform admin);
|
|
# the token-fetch route takes the calling node's own API key.
|
|
app.include_router(discord_router.router)
|
|
app.include_router(discord_link_router.router) # auth per-endpoint (org owner / bot service key) — server-26#213
|
|
app.include_router(discord_checks.router) # auth per-endpoint (org member / service key or admin) — server-26#225
|
|
app.include_router(discord_oauth.router) # auth per-endpoint (org owner / signed-in starter); 404 while OAuth is off — #225 S6
|
|
# NOTE: there used to be an app.routers.mqtt_auth router here (an HTTP
|
|
# backend for the mosquitto-go-auth plugin). That plugin's upstream project
|
|
# is archived (no CVE patches) and was rejected for an internet-facing
|
|
# broker — see MQTT-PUBLIC-AUTH-PLAN.md. MQTT auth is now mosquitto's own
|
|
# built-in dynamic-security plugin (app/internal/dynsec.py talks to it over
|
|
# MQTT control topics, not HTTP), so there is nothing at /internal/mqtt/*
|
|
# anymore. Caddy's Caddyfile.j2 still 404s /internal/* on api.<domain> as
|
|
# defence in depth even though nothing calls it today — cheap insurance
|
|
# against a future /internal/* route being added and forgotten there.
|
|
|
|
|
|
# Read straight from the environment rather than through Settings: this is a
|
|
# build stamp baked in by the Dockerfile, not configuration anyone sets or
|
|
# tunes, and keeping it out of Settings avoids implying it can be changed.
|
|
_GIT_SHA = os.getenv("GIT_SHA", "unknown")
|
|
|
|
|
|
@app.get("/health")
|
|
async def health():
|
|
return {
|
|
"ok": True,
|
|
"mqtt_connected": mqtt_handler.is_connected,
|
|
# CI asserts this equals the commit it just deployed. Without it a
|
|
# deploy can "succeed" while the previous container is still serving.
|
|
"git_sha": _GIT_SHA,
|
|
}
|
|
|
|
|
|
# Deliberately unauthenticated, same as /health above: the CI deploy step
|
|
# curls /health with no credentials, and this is diagnostic state (which AI
|
|
# tier is degraded and why), not a secret — no API keys or tokens appear in
|
|
# it. Keeping it auth-free means an external uptime check can watch it too.
|
|
@app.get("/health/ai")
|
|
async def health_ai():
|
|
return {"tiers": ai_health.snapshot()}
|