admin / Syanpse-Vanguard

"""Agent ingest API (separate port, mTLS-guarded).

Hot path: decompress -> push raw NDJSON to the broker -> ack immediately. No
parsing here — normalizer workers do that off the hot path so p99 stays low and
agents can trim their disk buffer as soon as we ack `last_seq`."""
from __future__ import annotations

import gzip
import hashlib
import logging
from contextlib import asynccontextmanager

import zstandard
from fastapi import (APIRouter, Depends, FastAPI, Header, HTTPException, Request,
                    status)

from sv.bootstrap import shutdown, startup
from sv.config import settings
from sv.ingest import registry
from sv.ingest.mtls import client_cn, client_ip
from sv.models.agent import (AgentRegisterRequest, AgentRegisterResponse,
                             EventBundle, IngestAck)

log = logging.getLogger("sv.ingest")
_ZSTD = zstandard.ZstdDecompressor()

# Routes live on a router so they can be mounted either standalone (this module's
# `app`) or into the consolidated single-process app (sv.main).
router = APIRouter()


@asynccontextmanager
async def lifespan(app: FastAPI):
    from sv.metrics import EpsMeter
    engine, broker = await startup(with_broker=True)
    app.state.engine, app.state.broker = engine, broker
    app.state.eps_meter = EpsMeter()
    try:
        yield
    finally:
        await shutdown(engine, broker)


# --- identity ----------------------------------------------------------------
async def authenticated_agent(
    request: Request,
    authorization: str | None = Header(default=None),
) -> str:
    """Resolve the calling agent's id. mTLS mode: bind cert CN -> agent. Dev mode
    (mTLS disabled): bearer ingest token -> agent."""
    engine = request.app.state.engine
    if settings.mtls_enabled:
        cn = client_cn(request)
        if not cn:
            raise HTTPException(status.HTTP_401_UNAUTHORIZED, "client certificate required")
        agent = await registry.find_by_cert_cn(engine, cn)
        if not agent or agent["revoked"]:
            raise HTTPException(status.HTTP_401_UNAUTHORIZED, "unknown or revoked agent cert")
        return agent["id"]

    # dev-mode bearer fallback
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "bearer ingest token required")
    from sv.storage.base import Query
    token_hash = hashlib.sha256(authorization[7:].encode()).hexdigest()
    row = await engine.fetchone(Query(
        "SELECT id, revoked FROM agents WHERE ingest_token_hash = ?", [token_hash]))
    if not row or row["revoked"]:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid ingest token")
    return row["id"]


# --- endpoints ---------------------------------------------------------------
@router.post("/v1/agents/register", response_model=AgentRegisterResponse)
async def register(
    req: AgentRegisterRequest,
    request: Request,
    authorization: str | None = Header(default=None),
) -> AgentRegisterResponse:
    """One-time bootstrap. Requires the out-of-band enrollment token as
    `Authorization: Bearer <SV_ENROLLMENT_TOKEN>`."""
    if authorization != f"Bearer {settings.enrollment_token}":
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid enrollment token")

    engine = request.app.state.engine
    # In a full deployment the server-CA signs req.csr_pem and returns the cert.
    # The signing plumbing lives in ops/ (see certs/README); here we bind the CN
    # that the CSR/cert will carry so mTLS calls resolve to this agent.
    cert_cn = f"agent-{req.hostname}" if settings.mtls_enabled else None
    agent_id, ingest_token = await registry.enroll(engine, req, cert_cn,
                                                   ip=client_ip(request))

    ca_chain = None
    if settings.mtls_enabled:
        ca_chain = _read_optional(f"{settings.cert_dir}/ca.crt")

    return AgentRegisterResponse(
        agent_id=agent_id,
        client_cert_pem=None,      # populated once CSR signing is wired in ops
        ca_chain_pem=ca_chain,
        ingest_token=ingest_token,
        heartbeat_seconds=60,
    )


@router.post("/v1/ingest/bulk", response_model=IngestAck)
async def ingest_bulk(
    request: Request,
    x_sv_bundle: str = Header(..., alias="X-SV-Bundle"),
    agent_id: str = Depends(authenticated_agent),
) -> IngestAck:
    bundle = EventBundle.model_validate_json(x_sv_bundle)
    if str(bundle.agent_id) != agent_id:
        raise HTTPException(status.HTTP_403_FORBIDDEN, "bundle/identity agent mismatch")

    body = await request.body()
    if bundle.codec == "zstd":
        raw = _ZSTD.decompress(body)
    elif bundle.codec == "gzip":
        raw = gzip.decompress(body)
    else:
        raw = body

    lines = [ln for ln in raw.split(b"\n") if ln.strip()]
    broker = request.app.state.broker
    await broker.push_raw(agent_id, bundle.seq, b"\n".join(lines))
    await registry.touch_seq(request.app.state.engine, agent_id, bundle.seq,
                             ip=client_ip(request))

    # Feed the events-per-second meter (server-side receive rate).
    meter = getattr(request.app.state, "eps_meter", None)
    if meter is not None:
        meter.record(len(lines))

    log.debug("ingest agent=%s seq=%s lines=%d", agent_id, bundle.seq, len(lines))
    return IngestAck(accepted=len(lines), last_seq=bundle.seq)


@router.post("/v1/ingest/events")
async def ingest_events(
    request: Request,
    authorization: str | None = Header(default=None),
) -> dict:
    """Key-authenticated push ingest for external Synapse-suite apps (IronNode).

    Auth: Authorization: Bearer <ingress API key> (managed in the admin UI).
    Body: {"events": [ {ts, host, source_type?, message, severity?, category?,
           action?, mitre_technique?, user_name?, host_ip?, event_id?, fields?} ]}
    Each event is pushed through the normal broker -> normalizer -> storage path.
    """
    from sv import ingress

    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "ingress API key required")
    engine = request.app.state.engine
    if not await ingress.verify(engine, authorization[7:]):
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid or disabled ingress API key")

    payload = await request.json()
    events = payload.get("events") if isinstance(payload, dict) else None
    if not isinstance(events, list):
        raise HTTPException(status.HTTP_400_BAD_REQUEST, "body must be {\"events\": [...]}")
    if len(events) > 5000:
        raise HTTPException(status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, "max 5000 events per request")

    import json as _json
    lines = []
    for ev in events:
        if not isinstance(ev, dict):
            continue
        ev.setdefault("source_type", "ironnode")   # force the ironnode parser path
        ev["source_type"] = "ironnode"
        lines.append(_json.dumps(ev, default=str).encode())

    if lines:
        # seq 0: external push has no per-agent replay sequence.
        await request.app.state.broker.push_raw("ingress", 0, b"\n".join(lines))
        meter = getattr(request.app.state, "eps_meter", None)
        if meter is not None:
            meter.record(len(lines))

    log.debug("ingress events accepted=%d", len(lines))
    return {"accepted": len(lines)}


@router.get("/healthz")
async def healthz():
    return {"status": "ok"}


def _read_optional(path: str) -> str | None:
    try:
        with open(path, "r", encoding="utf-8") as fh:
            return fh.read()
    except OSError:
        return None


app = FastAPI(title="Synapse-Vanguard Agent Ingest", docs_url=None,
              redoc_url=None, lifespan=lifespan)
app.include_router(router)