Files
haproxy-openmanager/backend/utils/request_log_settings.py
T
mustafa.ulukaya ef26860df9 feat(logging): unified request/response log with configurable retention (v1.11.0)
Until now the only record of what happened was `user_activity_logs`, which
stores non-GET 2xx operations with no bodies. When something failed you could
see that a counter went up, never what was sent or what came back.

This adds one queryable timeline covering both directions:

- inbound: every API call, including GETs and including 4xx/5xx, with the
  user, client IP, status, duration and — redacted, size-capped — the request
  and response bodies.
- outbound: every HTTP call the backend makes, tagged with who it went to
  (ACME/Let's Encrypt, Cloudflare, GoDaddy, HAProxy stats, agents, the ACME
  diagnostics probe).

Outbound rows inherit the inbound request's id, so one operator action and the
CA/DNS calls it triggered read as a single trace: opening a failed "Request
Certificate" shows the exact POST /acme/new-order and the CA's 429 underneath.

Implementation notes:

- Capture is a pure-ASGI middleware that TEES the request and response streams
  rather than draining them. `await request.body()` inside a BaseHTTPMiddleware
  would consume the receive channel and break the raw-body agent heartbeat
  handler. Registered last so it is outermost: it then sees the final
  client-visible response and seeds correlation_id_context before the error
  handler reads it.
- Rows are written by a batching background writer with a bounded queue, so the
  request path never awaits the database and a saturated logger drops rows
  visibly (surfaced on the page) instead of blocking. Redaction runs on the
  writer, off the request coroutine.
- Secrets never land: headers are an allowlist with Authorization/Cookie kept
  only as a presence marker; body keys and value shapes are redacted
  (passwords, tokens, api_token, API keys, private-key PEMs, JWTs); the ACME
  JWS request body is never stored, because a stored protected+signature pair
  is a replayable credential — a summary is logged instead; DNS-provider errors
  record only the exception type; the ACME HTTP-01 challenge endpoint is
  excluded so key_authorization is never captured.
- Retention is operator-configurable in Settings -> Request Log: separate day
  counts for successful and failed rows (7 / 30) plus a hard row cap (500k),
  whichever is reached first. Pruned in batches under a Postgres advisory lock,
  with the day counts bound as parameters, never interpolated.
- New permissions requestlog.read / requestlog.manage. super_admin and
  security_admin get both, operator gets read, viewer gets neither.

Schema: one new table (request_logs) plus its settings seed, SCHEMA_VERSION
10 -> 11, auto-migrated. No existing table altered, no agent or rendered-config
change. Kill switches: REQUEST_LOG_ENABLED=false (middleware never registered)
or the `enabled` toggle in Settings.

Tests: 245 new (7 backend files + 1 frontend), full suite 1655 backend +
17 frontend passing.
2026-08-11 02:36:03 +03:00

257 lines
9.2 KiB
Python

"""v1.11.0 — operator-tunable settings for the request/response log.
The middleware runs on EVERY request, so the hot path must not touch the
database. `get_config()` returns a module-global immutable snapshot with no
`await`; `refresh_config()` reloads it from `system_settings` and is called
* once at startup, right after migrations,
* every `_TTL_SECONDS` from the sink's writer loop (off the request path),
* synchronously at the end of `PUT /api/request-logs/settings`, so an
operator's change takes effect immediately instead of up to 30s later.
asyncpg has no JSONB codec registered on this pool (see
database/connection.py), so every value comes back as a raw JSON *string* and
needs the `isinstance(v, str)` + `json.loads` guard used elsewhere in this
codebase (services/acme_service.py, utils/activity_log.py).
"""
import json
import logging
import time
from dataclasses import dataclass, replace
from typing import Any, Dict, Optional, Tuple
from database.connection import get_database_connection, close_database_connection
logger = logging.getLogger("haproxy_openmanager.request_log")
SETTINGS_CATEGORY = "requestlog"
# Kept in sync with the seed in database/migrations.ensure_request_log_settings().
# backend/tests/test_request_log_settings.py asserts the two agree, so a change
# here without a change there fails the suite rather than drifting silently.
DEFAULT_EXCLUDE_PATHS = (
"/api/request-logs",
"/api/health",
"/api/docs",
"/api/redoc",
"/api/openapi.json",
"/.well-known/acme-challenge",
"/api/agents/heartbeat",
"/static",
"/favicon.ico",
)
@dataclass(frozen=True)
class RequestLogConfig:
enabled: bool = True
capture_inbound: bool = True
capture_outbound: bool = True
capture_bodies: bool = True
capture_get: bool = True
max_body_bytes: int = 8192
sample_rate: float = 1.0
exclude_paths: Tuple[str, ...] = DEFAULT_EXCLUDE_PATHS
success_retention_days: int = 7
error_retention_days: int = 30
max_rows: int = 500000
prune_interval_minutes: int = 60
def as_dict(self) -> Dict[str, Any]:
return {
"enabled": self.enabled,
"capture_inbound": self.capture_inbound,
"capture_outbound": self.capture_outbound,
"capture_bodies": self.capture_bodies,
"capture_get": self.capture_get,
"max_body_bytes": self.max_body_bytes,
"sample_rate": self.sample_rate,
"exclude_paths": list(self.exclude_paths),
"success_retention_days": self.success_retention_days,
"error_retention_days": self.error_retention_days,
"max_rows": self.max_rows,
"prune_interval_minutes": self.prune_interval_minutes,
}
DEFAULT_CONFIG = RequestLogConfig()
_CACHE: RequestLogConfig = DEFAULT_CONFIG
_CACHE_AT: float = 0.0
_TTL_SECONDS: float = 30.0
# Bounds, mirrored by the Pydantic model in routers/request_logs.py. Kept here
# too because refresh_config() reads whatever is in the table, which may have
# been written by an older build or by hand.
_BOUNDS = {
"max_body_bytes": (0, 262144),
"success_retention_days": (1, 365),
"error_retention_days": (1, 365),
"max_rows": (1000, 50_000_000),
"prune_interval_minutes": (5, 1440),
}
MAX_EXCLUDE_PATHS = 64
MAX_EXCLUDE_PATH_LENGTH = 200
def get_config() -> RequestLogConfig:
"""Hot-path read: no await, no DB, no lock. Returns the last snapshot."""
return _CACHE
def set_config(config: RequestLogConfig) -> None:
"""Replace the snapshot directly. Used by the settings PUT handler (which
already has the validated values) and by tests."""
global _CACHE, _CACHE_AT
_CACHE = config
_CACHE_AT = time.monotonic()
def _clamp_int(raw: Any, field: str, fallback: int) -> int:
try:
value = int(raw)
except (TypeError, ValueError):
return fallback
low, high = _BOUNDS[field]
return max(low, min(high, value))
def _clamp_float(raw: Any, fallback: float, low: float, high: float) -> float:
try:
value = float(raw)
except (TypeError, ValueError):
return fallback
return max(low, min(high, value))
def _as_bool(raw: Any, fallback: bool) -> bool:
if isinstance(raw, bool):
return raw
if isinstance(raw, (int, float)):
return bool(raw)
if isinstance(raw, str):
lowered = raw.strip().lower()
if lowered in ("true", "1", "yes", "on"):
return True
if lowered in ("false", "0", "no", "off"):
return False
return fallback
def normalize_exclude_paths(raw: Any, fallback: Tuple[str, ...]) -> Tuple[str, ...]:
"""Coerce whatever is stored into a bounded tuple of path prefixes."""
if not isinstance(raw, (list, tuple)):
return fallback
out = []
for item in raw:
if not isinstance(item, str):
continue
candidate = item.strip()
if not candidate.startswith("/") or len(candidate) > MAX_EXCLUDE_PATH_LENGTH:
continue
out.append(candidate)
if len(out) >= MAX_EXCLUDE_PATHS:
break
return tuple(out) if out else fallback
def config_from_mapping(values: Dict[str, Any], base: Optional[RequestLogConfig] = None) -> RequestLogConfig:
"""Build a config from a plain suffix→value mapping, clamping every field.
Unknown keys are ignored and missing keys keep the value from `base`
(default: the shipped defaults), so a partially-seeded table still yields a
complete, usable config.
"""
base = base or DEFAULT_CONFIG
return replace(
base,
enabled=_as_bool(values.get("enabled", base.enabled), base.enabled),
capture_inbound=_as_bool(values.get("capture_inbound", base.capture_inbound), base.capture_inbound),
capture_outbound=_as_bool(values.get("capture_outbound", base.capture_outbound), base.capture_outbound),
capture_bodies=_as_bool(values.get("capture_bodies", base.capture_bodies), base.capture_bodies),
capture_get=_as_bool(values.get("capture_get", base.capture_get), base.capture_get),
max_body_bytes=_clamp_int(values.get("max_body_bytes", base.max_body_bytes), "max_body_bytes", base.max_body_bytes),
sample_rate=_clamp_float(values.get("sample_rate", base.sample_rate), base.sample_rate, 0.0, 1.0),
exclude_paths=normalize_exclude_paths(values.get("exclude_paths"), base.exclude_paths),
success_retention_days=_clamp_int(
values.get("success_retention_days", base.success_retention_days),
"success_retention_days", base.success_retention_days,
),
error_retention_days=_clamp_int(
values.get("error_retention_days", base.error_retention_days),
"error_retention_days", base.error_retention_days,
),
max_rows=_clamp_int(values.get("max_rows", base.max_rows), "max_rows", base.max_rows),
prune_interval_minutes=_clamp_int(
values.get("prune_interval_minutes", base.prune_interval_minutes),
"prune_interval_minutes", base.prune_interval_minutes,
),
)
def _decode_setting_value(raw: Any) -> Any:
"""JSONB comes back as a raw string on this pool — parse it, but keep the
original text if it is not valid JSON (an operator may have hand-written
`7` or `seven`)."""
if isinstance(raw, str):
try:
return json.loads(raw)
except (json.JSONDecodeError, ValueError):
return raw
return raw
async def load_settings_rows(conn) -> Dict[str, Any]:
"""Read the `requestlog.*` rows into a suffix→value mapping."""
rows = await conn.fetch(
"SELECT key, value FROM system_settings WHERE category = $1",
SETTINGS_CATEGORY,
)
values: Dict[str, Any] = {}
for row in rows:
key = row["key"]
suffix = key.split(".", 1)[1] if "." in key else key
values[suffix] = _decode_setting_value(row["value"])
return values
async def refresh_config(force: bool = True) -> RequestLogConfig:
"""Reload the snapshot from the database.
Never raises and never leaves a half-built config behind: on any failure
the previous snapshot is kept, so a transient DB blip cannot silently turn
logging off (or on).
"""
global _CACHE_AT
if not force and (time.monotonic() - _CACHE_AT) < _TTL_SECONDS:
return _CACHE
conn = None
try:
conn = await get_database_connection()
values = await load_settings_rows(conn)
if values:
set_config(config_from_mapping(values))
else:
# Table not seeded yet (fresh install mid-migration) — keep the
# in-code defaults but stamp the timestamp so we don't re-query
# every tick.
_CACHE_AT = time.monotonic()
return _CACHE
except Exception as exc:
logger.debug(f"refresh_config: keeping previous snapshot ({exc})")
_CACHE_AT = time.monotonic()
return _CACHE
finally:
if conn is not None:
try:
await close_database_connection(conn)
except Exception:
pass
async def maybe_refresh_config() -> RequestLogConfig:
"""TTL-gated refresh, called from the sink's writer loop."""
return await refresh_config(force=False)