haproxy-openmanager/backend/routers/request_logs.py

"""v1.11.0 — the read/administration API for the unified request/response log.

Endpoints (declaration order matters — see below):

    GET    /api/request-logs/settings   requestlog.manage
    PUT    /api/request-logs/settings   requestlog.manage
    GET    /api/request-logs/stats      requestlog.read
    POST   /api/request-logs/purge      requestlog.manage
    GET    /api/request-logs            requestlog.read
    GET    /api/request-logs/{log_id}   requestlog.read

`/{log_id}` is a single-segment path, so FastAPI — which matches in declaration
order — would shadow `/settings`, `/stats` and `/purge` if it came first. The
literals are therefore declared before it. (This is the mirror image of the
trap in routers/settings.py, where `GET /{category}` sits at the top of the
file and swallows every literal route added after it.)

Settings are stored in `system_settings` under the `requestlog` category, so
`GET /api/settings/requestlog` still reads them, but writes go through THIS
router: the generic `PUT /api/settings/{category}` stringifies values with
`str(value)`, which turns `True` into `'True'` — not valid JSON, and the
`::jsonb` cast then fails.
"""
import json
import logging
from datetime import datetime
from typing import Any, Dict, List, Optional

from fastapi import APIRouter, Header, HTTPException, Query
from pydantic import BaseModel, Field, field_validator

from auth_middleware import check_user_permission, get_current_user_from_token
from database.connection import get_database_connection, close_database_connection
from utils.request_log_settings import (
    DEFAULT_CONFIG,
    DEFAULT_EXCLUDE_PATHS,
    MAX_EXCLUDE_PATHS,
    MAX_EXCLUDE_PATH_LENGTH,
    SETTINGS_CATEGORY,
    config_from_mapping,
    get_config,
    load_settings_rows,
    refresh_config,
    set_config,
)
from utils.request_log_sink import TARGET_INBOUND_AGENT, request_log_sink

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/api/request-logs", tags=["Request Logs"])

# Columns returned by the list endpoint. Bodies and headers are detail-only:
# a 200-row page carrying two 8 KB JSONB blobs per row is a 3 MB response.
_LIST_COLUMNS = """
    id, request_id, direction, target, method, url, path, status_code,
    status_class, duration_ms, user_id, username, host(client_ip) AS client_ip,
    error, request_body_bytes, response_body_bytes, truncated, created_at
"""

_JSONB_COLUMNS = ("query_params", "request_headers", "request_body",
                  "response_headers", "response_body")


class RequestLogSettings(BaseModel):
    """Operator-tunable capture + retention policy."""

    enabled: bool = True
    capture_inbound: bool = True
    capture_outbound: bool = True
    capture_bodies: bool = True
    capture_get: bool = True
    capture_agent_success: bool = False
    max_body_bytes: int = Field(8192, ge=0, le=262144)
    sample_rate: float = Field(1.0, ge=0.0, le=1.0)
    exclude_paths: List[str] = Field(
        default_factory=lambda: list(DEFAULT_EXCLUDE_PATHS),
        max_length=MAX_EXCLUDE_PATHS,
    )
    success_retention_days: int = Field(7, ge=1, le=365)
    error_retention_days: int = Field(30, ge=1, le=365)
    max_rows: int = Field(500000, ge=1000, le=50_000_000)
    prune_interval_minutes: int = Field(60, ge=5, le=1440)

    @field_validator("exclude_paths")
    @classmethod
    def _validate_paths(cls, value: List[str]) -> List[str]:
        for entry in value:
            if not entry.startswith("/"):
                raise ValueError("exclude_paths entries must start with '/'")
            if len(entry) > MAX_EXCLUDE_PATH_LENGTH:
                raise ValueError(
                    f"exclude_paths entries must be <= {MAX_EXCLUDE_PATH_LENGTH} characters"
                )
        return value


async def _require(authorization: Optional[str], action: str) -> Dict[str, Any]:
    """Authenticate, then enforce `requestlog.<action>`.

    `current_user=` is passed through so the admin bypass in
    check_user_permission short-circuits without a second DB round-trip.
    """
    current_user = await get_current_user_from_token(authorization)
    allowed = await check_user_permission(
        current_user["id"], "requestlog", action, current_user=current_user
    )
    if not allowed:
        raise HTTPException(
            status_code=403,
            detail=f"Insufficient permissions: requestlog.{action} required",
        )
    return current_user


async def _can_manage(current_user: Dict[str, Any]) -> bool:
    return await check_user_permission(
        current_user["id"], "requestlog", "manage", current_user=current_user
    )


def _parse_jsonb(value: Any) -> Any:
    """asyncpg has no JSONB codec on this pool, so JSONB comes back as raw
    text. This router is a new contract, so it parses server-side and returns
    real JSON rather than pushing a JSON.parse() into the UI."""
    if isinstance(value, str):
        try:
            return json.loads(value)
        except (json.JSONDecodeError, ValueError):
            return value
    return value


def _row_to_dict(row) -> Dict[str, Any]:
    out = dict(row)
    for key in _JSONB_COLUMNS:
        if key in out:
            out[key] = _parse_jsonb(out[key])
    created = out.get("created_at")
    if isinstance(created, datetime):
        out["created_at"] = created.isoformat()
    return out


# ---------------------------------------------------------------------------
# Literal paths FIRST — see the module docstring.
# ---------------------------------------------------------------------------


@router.get("/settings")
async def get_request_log_settings(authorization: Optional[str] = Header(None)):
    """Current capture + retention policy, plus the shipped defaults so the UI
    can offer a 'reset' without hardcoding them."""
    await _require(authorization, "manage")

    conn = None
    try:
        conn = await get_database_connection()
        values = await load_settings_rows(conn)
        config = config_from_mapping(values) if values else DEFAULT_CONFIG
        return {
            "settings": config.as_dict(),
            "defaults": DEFAULT_CONFIG.as_dict(),
            "category": SETTINGS_CATEGORY,
        }
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Error fetching request log settings: {e}")
        raise HTTPException(status_code=500, detail="Failed to fetch request log settings")
    finally:
        if conn is not None:
            await close_database_connection(conn)


@router.put("/settings")
async def update_request_log_settings(
    body: RequestLogSettings,
    authorization: Optional[str] = Header(None),
):
    """Persist the policy and apply it immediately.

    `refresh_config()` at the end is what makes an operator's change take
    effect on the very next request instead of up to 30 seconds later, when
    the writer loop would otherwise pick it up.
    """
    current_user = await _require(authorization, "manage")

    conn = None
    try:
        conn = await get_database_connection()
        updated = []
        for suffix, value in body.model_dump().items():
            await conn.execute(
                """
                INSERT INTO system_settings (key, value, category, updated_at, updated_by)
                VALUES ($1, $2::jsonb, $3, $4, $5)
                ON CONFLICT (key) DO UPDATE SET
                    value = EXCLUDED.value,
                    updated_at = EXCLUDED.updated_at,
                    updated_by = EXCLUDED.updated_by
                """,
                f"{SETTINGS_CATEGORY}.{suffix}",
                json.dumps(value),
                SETTINGS_CATEGORY,
                datetime.utcnow(),
                current_user.get("id"),
            )
            updated.append(suffix)

        # Apply in-process right away, then re-read so this worker's snapshot
        # is exactly what is on disk.
        set_config(config_from_mapping(body.model_dump()))
        await refresh_config()

        logger.info(
            f"Request log settings updated by {current_user.get('username')}: {len(updated)} keys"
        )
        return {"message": f"Updated {len(updated)} settings", "settings": get_config().as_dict()}
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Error updating request log settings: {e}")
        raise HTTPException(status_code=500, detail="Failed to update request log settings")
    finally:
        if conn is not None:
            await close_database_connection(conn)


@router.get("/stats")
async def get_request_log_stats(
    authorization: Optional[str] = Header(None),
    hours: int = Query(24, ge=1, le=720),
):
    """Volume and error breakdown over a window, plus table-level totals and
    this worker's sink counters (so a saturated queue is visible)."""
    await _require(authorization, "read")

    conn = None
    try:
        conn = await get_database_connection()

        by_direction = await conn.fetch(
            """
            SELECT direction,
                   COUNT(*)                                                    AS total,
                   COUNT(*) FILTER (WHERE status_class = 0 OR status_class >= 4) AS errors,
                   COALESCE(ROUND(AVG(duration_ms))::int, 0)                   AS avg_duration_ms,
                   COALESCE(MAX(duration_ms), 0)                               AS max_duration_ms
            FROM request_logs
            WHERE created_at > NOW() - ($1 || ' hours')::INTERVAL
            GROUP BY direction
            """,
            str(hours),
        )

        by_status = await conn.fetch(
            """
            SELECT status_class, COUNT(*) AS total
            FROM request_logs
            WHERE created_at > NOW() - ($1 || ' hours')::INTERVAL
            GROUP BY status_class
            ORDER BY status_class
            """,
            str(hours),
        )

        by_target = await conn.fetch(
            """
            SELECT target,
                   COUNT(*)                                                    AS total,
                   COUNT(*) FILTER (WHERE status_class = 0 OR status_class >= 4) AS errors
            FROM request_logs
            WHERE target IS NOT NULL
              AND created_at > NOW() - ($1 || ' hours')::INTERVAL
            GROUP BY target
            ORDER BY total DESC
            LIMIT 20
            """,
            str(hours),
        )

        totals = await conn.fetchrow(
            "SELECT COUNT(*) AS total_rows, MIN(created_at) AS oldest_at, "
            "MAX(created_at) AS newest_at FROM request_logs"
        )

        return {
            "window_hours": hours,
            "by_direction": [dict(r) for r in by_direction],
            "by_status_class": [dict(r) for r in by_status],
            "by_target": [dict(r) for r in by_target],
            "total_rows": (totals or {}).get("total_rows", 0),
            "oldest_at": totals["oldest_at"].isoformat() if totals and totals["oldest_at"] else None,
            "newest_at": totals["newest_at"].isoformat() if totals and totals["newest_at"] else None,
            # THIS WORKER only. The sink is a module global, so with
            # UVICORN_WORKERS > 1 each process keeps its own queue and its own
            # counters, and whichever worker happens to serve this request is
            # the one being reported. Labelled rather than aggregated: there is
            # no cross-process channel here, and a number that looks fleet-wide
            # but is not would understate drops by exactly the worker count.
            "sink": {**request_log_sink.stats, "scope": "this worker only"},
            "retention": {
                "success_retention_days": get_config().success_retention_days,
                "error_retention_days": get_config().error_retention_days,
                "max_rows": get_config().max_rows,
            },
        }
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Error fetching request log stats: {e}")
        raise HTTPException(status_code=500, detail="Failed to fetch request log stats")
    finally:
        if conn is not None:
            await close_database_connection(conn)


@router.post("/purge")
async def purge_request_logs(authorization: Optional[str] = Header(None)):
    """Run a retention pass now, ignoring the watermark.

    This applies the CONFIGURED retention — it is not a 'delete everything'
    button. It exists so an operator who has just lowered the retention does
    not have to wait for the next scheduled pass to reclaim the space.
    """
    current_user = await _require(authorization, "manage")
    from utils.request_log_prune import prune_request_logs_if_due

    counts = await prune_request_logs_if_due(force=True)
    logger.info(f"Manual request log purge by {current_user.get('username')}: {counts}")
    return {
        "message": "Retention pass completed",
        "removed": {
            "success": counts.get("success", 0),
            "error": counts.get("error", 0),
            "overflow": counts.get("overflow", 0),
        },
        "ran": bool(counts.get("ran")),
    }


# ---------------------------------------------------------------------------
# List, then the catch-all detail route LAST.
# ---------------------------------------------------------------------------


@router.get("")
async def list_request_logs(
    authorization: Optional[str] = Header(None),
    direction: Optional[str] = Query(None, pattern="^(inbound|outbound)$"),
    status_class: Optional[int] = Query(None, ge=0, le=5),
    method: Optional[str] = Query(None, max_length=10),
    target: Optional[str] = Query(None, max_length=32),
    user_id: Optional[int] = Query(None, ge=1),
    path_prefix: Optional[str] = Query(None, max_length=200),
    q: Optional[str] = Query(None, max_length=200),
    request_id: Optional[str] = Query(None, max_length=64),
    errors_only: bool = Query(False),
    since: Optional[datetime] = Query(None),
    until: Optional[datetime] = Query(None),
    min_duration_ms: Optional[int] = Query(None, ge=0),
    limit: int = Query(50, ge=1, le=500),
    offset: int = Query(0, ge=0),
):
    """Filtered, server-paginated list. Bodies are not included — use the
    detail endpoint for those."""
    current_user = await _require(authorization, "read")
    can_manage = await _can_manage(current_user)

    where: List[str] = []
    params: List[Any] = []

    def add(clause_template: str, value: Any) -> None:
        params.append(value)
        where.append(clause_template.format(n=len(params)))

    if direction:
        add("direction = ${n}", direction)
    if status_class is not None:
        add("status_class = ${n}", status_class)
    if method:
        add("method = ${n}", method.upper())
    if target:
        add("target = ${n}", target)
    if user_id is not None:
        add("user_id = ${n}", user_id)
    if path_prefix:
        add("path LIKE ${n} || '%'", path_prefix)
    if q:
        # Substring search has no index to lean on; it is the deliberately slow
        # filter and should be combined with a time window.
        add("url ILIKE '%' || ${n} || '%'", q)
    if request_id:
        add("request_id = ${n}", request_id)
    if errors_only:
        where.append("(status_class = 0 OR status_class >= 4)")
    if since:
        add("created_at >= ${n}", since)
    if until:
        add("created_at <= ${n}", until)
    if min_duration_ms is not None:
        add("duration_ms >= ${n}", min_duration_ms)

    # Self-scoping. Captured bodies are a broader disclosure surface than the
    # existing activity log, so a caller holding only `requestlog.read` sees
    # their OWN inbound requests, plus the fleet's. `requestlog.manage` (and the
    # is_admin bypass inside it) lifts the restriction.
    #
    # The agent clause is not a widening for its own sake, it is what makes the
    # `operator` grant do what the migration says it is for: "operators debug
    # failing applies and ACME orders, so they get read access to the request
    # log". An apply fails on the NODE, and the node reports that back over its
    # own API key - so the row carrying the diagnosis is an agent row with
    # `user_id IS NULL`, which own-rows-only scoping hid from exactly the role
    # the grant was written for. Scoped on `target`, not on `user_id IS NULL`:
    # anonymous inbound traffic (failed logins and their usernames, unauthorised
    # probes) is NOT agent traffic and stays admin-only.
    if not can_manage:
        params.append(current_user["id"])
        own = f"user_id = ${len(params)}"
        params.append(TARGET_INBOUND_AGENT)
        where.append(
            f"(direction = 'inbound' AND ({own} OR target = ${len(params)}))"
        )

    where_sql = (" WHERE " + " AND ".join(where)) if where else ""

    conn = None
    try:
        conn = await get_database_connection()

        rows = await conn.fetch(
            f"SELECT {_LIST_COLUMNS} FROM request_logs{where_sql} "
            f"ORDER BY id DESC LIMIT ${len(params) + 1} OFFSET ${len(params) + 2}",
            *params, limit, offset,
        )

        # Bounded count: an unfiltered COUNT(*) over a multi-million-row table
        # is a sequential scan on every page change. Cap it and tell the client
        # the number is a floor.
        count_cap = 10001
        counted = await conn.fetchval(
            f"SELECT COUNT(*) FROM (SELECT 1 FROM request_logs{where_sql} LIMIT {count_cap}) t",
            *params,
        )
        total = int(counted or 0)

        return {
            "logs": [_row_to_dict(r) for r in rows],
            "total": total,
            "total_is_estimate": total >= count_cap,
            "limit": limit,
            "offset": offset,
            "scoped_to_self": not can_manage,
        }
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Error listing request logs: {e}")
        raise HTTPException(status_code=500, detail="Failed to list request logs")
    finally:
        if conn is not None:
            await close_database_connection(conn)


@router.get("/{log_id}")
async def get_request_log(log_id: int, authorization: Optional[str] = Header(None)):
    """One exchange in full, plus every other row sharing its `request_id`.

    That `related` list is the point of the feature: one inbound API call and
    the ACME / DNS / agent calls it triggered read as a single trace.
    """
    current_user = await _require(authorization, "read")
    can_manage = await _can_manage(current_user)

    conn = None
    try:
        conn = await get_database_connection()
        row = await conn.fetchrow(
            "SELECT *, host(client_ip) AS client_ip_text FROM request_logs WHERE id = $1",
            log_id,
        )
        if not row:
            raise HTTPException(status_code=404, detail="Request log entry not found")

        record = _row_to_dict(row)
        record["client_ip"] = record.pop("client_ip_text", None)

        if not can_manage and not (
            record.get("direction") == "inbound"
            and (
                record.get("user_id") == current_user["id"]
                or record.get("target") == TARGET_INBOUND_AGENT
            )
        ):
            # Same self-scoping rule as the list endpoint. 404 rather than 403
            # so the endpoint does not confirm that a given id exists.
            raise HTTPException(status_code=404, detail="Request log entry not found")

        related = await conn.fetch(
            f"SELECT {_LIST_COLUMNS} FROM request_logs "
            "WHERE request_id = $1 AND id <> $2 ORDER BY id ASC LIMIT 100",
            record["request_id"], log_id,
        )

        return {"log": record, "related": [_row_to_dict(r) for r in related]}
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Error fetching request log {log_id}: {e}")
        raise HTTPException(status_code=500, detail="Failed to fetch request log entry")
    finally:
        if conn is not None:
            await close_database_connection(conn)