server-26/drb-c2-core/app/main.py

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


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(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
# 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()}