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.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
from app.internal import dynsec
from app.internal import firestore as fstore


async def _release_orphaned_tokens():
    """Release all in-use tokens on startup — voice connections don't survive server restarts."""
    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)
    for doc in results:
        await fstore.doc_update("bot_tokens", doc.id, {
            "in_use": False,
            "assigned_node_id": None,
            "assigned_at": None,
        })
    if results:
        logger.info(f"Released {len(results)} orphaned token(s) on startup.")


@asynccontextmanager
async def lifespan(app: FastAPI):
    logger.info("DRB C2 Core starting.")
    await _release_orphaned_tokens()

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

    yield  # --- app running ---

    logger.info("DRB C2 Core shutting down.")
    sweeper_task.cancel()
    summarizer_task.cancel()
    induction_task.cancel()
    recorrelation_task.cancel()
    await mqtt_handler.disconnect()


app = FastAPI(title="DRB C2 Core", lifespan=lifespan)

# "*" plus allow_credentials=True is not the permissive-but-harmless setting it
# looks like. Starlette does not refuse the combination -- it reflects the
# caller's Origin back and still sends Access-Control-Allow-Credentials: true,
# so the effective policy becomes "any origin, with credentials", the opposite
# of what a wildcard normally means. Rather than trust every deployment to
# remember to override CORS_ORIGINS, make the dangerous pair unrepresentable.
def cors_allows_credentials(origins: list[str]) -> bool:
    """False when any entry is a wildcard. Extracted so it can be tested
    without re-importing this module, which drags in every router."""
    return "*" not in origins


_cors_is_wildcard = not cors_allows_credentials(settings.cors_origins)
if _cors_is_wildcard:
    logger.error(
        "CORS_ORIGINS is '*', so credentialed cross-origin requests are being "
        "DISABLED to avoid reflecting every caller's origin back with "
        "Access-Control-Allow-Credentials. Set CORS_ORIGINS to your frontend "
        "origin(s) in production, e.g. [\"https://app.example.com\"]."
    )

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.cors_origins,
    allow_methods=["*"],
    allow_headers=["*"],
    allow_credentials=not _cors_is_wildcard,
)

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(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(upload.router)    # auth is per-node, handled inline
app.include_router(admin.router)     # auth is per-endpoint (read: firebase, write: admin)
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)
# 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()}