tracer-cloud--opensre/gateway/webapp.py

"""The gateway's single FastAPI app: health probes, alert intake, investigations.

Every HTTP endpoint OpenSRE serves lives here, on one port — ``/`` ``/health``
``/ok`` (health probes), ``/healthz`` (liveness), ``POST /alerts`` (external
alert pushes into the process-wide :class:`AlertInbox`), and ``POST /investigate``
(run an investigation synchronously and return the RCA report). Hosted by the
gateway daemon and the interactive shell via :mod:`gateway.web_server`, or
standalone via ``uvicorn gateway.webapp:app``.
"""

from __future__ import annotations

import hmac
import json
import logging
import os
from datetime import UTC, datetime
from typing import Any

from fastapi import FastAPI, Request, Response, status
from fastapi.responses import JSONResponse
from pydantic import BaseModel, ValidationError

from config.config import LLMSettings, get_environment
from config.platform_bootstrap import ensure_project_platform_package
from config.version import get_opensre_version
from core.domain.alerts.inbox import (
    AlertInbox,
    IncomingAlert,
    get_current_inbox,
    set_current_inbox,
)

ensure_project_platform_package()

from platform.observability.errors.sentry import capture_exception, init_sentry  # noqa: E402
from tools.investigation.capability import (  # noqa: E402
    resolve_investigation_context,
    run_investigation_payload,
)

init_sentry(entrypoint="webapp")

logger = logging.getLogger(__name__)

# Cap on POST body size accepted from any caller (authed or not). Realistic
# alert payloads top out around 50 KB, so 1 MiB is ~20× headroom.
MAX_ALERT_BODY_BYTES = 1 * 1024 * 1024

_LOOPBACK_HOSTS = frozenset({"127.0.0.1", "::1", "localhost"})


class HealthResponse(BaseModel):
    ok: bool
    version: str
    llm_configured: bool
    env: str


app = FastAPI()


def get_health_response() -> HealthResponse:
    try:
        LLMSettings.from_env()
        llm_configured = True
    except ValidationError:
        llm_configured = False

    return HealthResponse(
        ok=llm_configured,
        version=get_opensre_version(),
        llm_configured=llm_configured,
        env=get_environment().value,
    )


@app.get("/", response_model=HealthResponse)
@app.get("/health", response_model=HealthResponse)
@app.get("/ok", response_model=HealthResponse)
def health(response: Response) -> HealthResponse:
    health_response = get_health_response()
    response.status_code = (
        status.HTTP_200_OK if health_response.ok else status.HTTP_503_SERVICE_UNAVAILABLE
    )
    return health_response


@app.get("/healthz")
def healthz() -> dict[str, str]:
    return {"status": "ok"}


def _alert_inbox() -> AlertInbox:
    """The process-wide inbox; hosts may install their own via set_current_inbox."""
    inbox = get_current_inbox()
    if inbox is None:
        inbox = AlertInbox()
        set_current_inbox(inbox)
    return inbox


def _gateway_auth_error(request: Request) -> JSONResponse | None:
    """Bearer-token auth when configured; otherwise loopback callers only.

    Shared by every mutating gateway route (``/alerts``, ``/investigate``) since
    they sit behind the same trust boundary: local callers or a configured token.
    """
    token = os.environ.get("OPENSRE_ALERT_LISTENER_TOKEN")
    if token:
        supplied = request.headers.get("authorization", "")
        if hmac.compare_digest(supplied, f"Bearer {token}"):
            return None
        return JSONResponse({"error": "unauthorized"}, status_code=401)
    client_host = request.client.host if request.client else ""
    if client_host in _LOOPBACK_HOSTS:
        return None
    return JSONResponse(
        {"error": "set OPENSRE_ALERT_LISTENER_TOKEN to accept non-loopback callers"},
        status_code=403,
    )


@app.post("/alerts")
async def receive_alert(request: Request) -> JSONResponse:
    if (auth_error := _gateway_auth_error(request)) is not None:
        return auth_error

    try:
        declared_length = int(request.headers.get("content-length", 0))
    except ValueError:
        return JSONResponse({"error": "invalid Content-Length"}, status_code=400)
    if declared_length < 0:
        return JSONResponse({"error": "invalid Content-Length"}, status_code=400)
    if declared_length > MAX_ALERT_BODY_BYTES:
        return JSONResponse({"error": "payload too large"}, status_code=413)

    body = await request.body()
    if len(body) > MAX_ALERT_BODY_BYTES:
        return JSONResponse({"error": "payload too large"}, status_code=413)

    try:
        data = json.loads(body)
    except ValueError:
        return JSONResponse({"error": "invalid json"}, status_code=400)

    try:
        if not isinstance(data, dict):
            raise TypeError("alert payload must be a JSON object")
        if data.get("received_at") is None:
            data["received_at"] = datetime.now(UTC)
        alert = IncomingAlert.model_validate(data)
    except (TypeError, ValidationError, ValueError) as exc:
        return JSONResponse({"error": str(exc)}, status_code=400)

    inbox = _alert_inbox()
    accepted = inbox.put(alert)
    payload: dict[str, Any] = {"queued": True, "queue_depth": inbox.qsize}
    if not accepted:
        payload["dropped"] = inbox.dropped
        payload["warning"] = "inbox full, oldest alert dropped"
    return JSONResponse(payload, status_code=202)


class InvestigateRequest(BaseModel):
    raw_alert: dict[str, Any]
    alert_name: str | None = None
    pipeline_name: str | None = None
    severity: str | None = None


class InvestigateResponse(BaseModel):
    report: str
    problem_md: str
    root_cause: str
    is_noise: bool = False
    validity_score: float = 0.0
    tool_calls: list[dict[str, Any]] | None = None


@app.post("/investigate", response_model=InvestigateResponse)
def investigate(req: InvestigateRequest, request: Request) -> InvestigateResponse | JSONResponse:
    """Run an investigation synchronously and return the RCA report.

    Lets external systems (CI pipelines, custom webhooks, chat integrations
    without a native tool) trigger the same investigation pipeline the CLI and
    interactive shell use, over HTTP. FastAPI runs this sync handler in a
    threadpool, so a long investigation does not block ``/health`` or ``/alerts``.
    """
    if (auth_error := _gateway_auth_error(request)) is not None:
        return auth_error

    investigation_metadata = resolve_investigation_context(
        raw_alert=req.raw_alert,
        alert_name=req.alert_name,
        pipeline_name=req.pipeline_name,
        severity=req.severity,
    )
    try:
        result = run_investigation_payload(
            raw_alert=req.raw_alert,
            investigation_metadata=investigation_metadata,
        )
        return InvestigateResponse(**result)
    except Exception as exc:
        # Full detail (which may include internal paths, stack context, or
        # upstream error bodies) goes to logs/Sentry only. The HTTP response
        # carries just the exception type so it stays actionable without
        # exposing internals to the caller (CodeQL: information exposure
        # through an exception).
        logger.exception("Investigation failed")
        capture_exception(exc, context="gateway.webapp.investigate")
        return JSONResponse(
            {"error": f"investigation failed: {type(exc).__name__}"},
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
        )