ctms/backend/app/core/request_context.py

"""Request-scoped metadata used by server-side audit writers."""

from __future__ import annotations

import ipaddress
import re
import uuid
from contextvars import ContextVar, Token
from dataclasses import dataclass
from typing import Any

from app.core.config import settings


@dataclass(frozen=True)
class RequestAuditContext:
    request_id: str | None = None
    client_ip: str | None = None
    user_agent: str | None = None
    client_type: str | None = None
    client_version: str | None = None
    client_platform: str | None = None
    build_channel: str | None = None
    build_commit: str | None = None
    request_headers: dict[str, str] | None = None
    request_snapshot: dict[str, Any] | None = None


_request_audit_context: ContextVar[RequestAuditContext | None] = ContextVar(
    "request_audit_context",
    default=None,
)

_SENSITIVE_TEXT_PATTERN = re.compile(
    r"(?i)(bearer\s+)[A-Za-z0-9._~+/=-]+|((?:access_)?token|authorization|password|passwd|secret|credential|api[-_]?key|session)=([^&\s]+)"
)
_SENSITIVE_HEADER_NAME_PATTERN = re.compile(
    r"(?i)(authorization|cookie|set-cookie|token|password|passwd|secret|credential|api[-_]?key|session)"
)
_SAFE_REQUEST_HEADER_NAMES = frozenset(
    {
        "accept",
        "accept-language",
        "content-type",
        "host",
        "origin",
        "user-agent",
        "x-request-id",
        "x-correlation-id",
    }
)
_SAFE_REQUEST_HEADER_PREFIXES = ("x-ctms-",)
_SENSITIVE_PARAMETER_NAME_PATTERN = re.compile(
    r"(?i)(token|authorization|password|passwd|secret|credential|api[-_]?key|session|"
    r"subject|participant|patient|user_?name|full_?name|email|phone|mobile|id_?card|"
    r"identity|certificate|contact|address)"
)
_KNOWN_CLIENT_TYPES = frozenset({"web", "desktop"})
_CTMS_CLIENT_SOURCE_TO_TYPE = {
    "ctms-web": "web",
    "ctms-desktop": "desktop",
}


def _clean_audit_value(value: str | None, max_length: int) -> str | None:
    if value is None:
        return None
    cleaned = _SENSITIVE_TEXT_PATTERN.sub(lambda m: f"{m.group(1) or m.group(2) + '='}[redacted]", value.strip())
    if not cleaned:
        return None
    return cleaned[:max_length]


def _clean_header_value(value: str | None, max_length: int) -> str | None:
    return _clean_audit_value(value, max_length)


def resolve_ctms_client_type(headers: Any) -> str | None:
    source = _clean_header_value(headers.get("x-ctms-client-source"), 32)
    if source:
        mapped_type = _CTMS_CLIENT_SOURCE_TO_TYPE.get(source.strip().lower())
        if mapped_type:
            return mapped_type

    client_type = _clean_header_value(headers.get("x-ctms-client-type"), 16)
    if not client_type:
        return None
    normalized = client_type.strip().lower()
    return normalized if normalized in _KNOWN_CLIENT_TYPES else normalized[:16]


def build_sanitized_request_headers(request: Any) -> dict[str, str] | None:
    captured: dict[str, str] = {}
    for raw_name, raw_value in request.headers.items():
        name = str(raw_name).strip().lower()
        if not name:
            continue
        if _SENSITIVE_HEADER_NAME_PATTERN.search(name):
            captured[name] = "[redacted]"
            continue
        if name not in _SAFE_REQUEST_HEADER_NAMES and not name.startswith(_SAFE_REQUEST_HEADER_PREFIXES):
            continue
        value = _clean_header_value(str(raw_value), 500)
        if value:
            captured[name] = value
    return captured or None


def _sanitize_named_value(raw_name: Any, raw_value: Any, max_length: int = 500) -> dict[str, str] | None:
    name = _clean_audit_value(str(raw_name), 120)
    if not name:
        return None
    if _SENSITIVE_PARAMETER_NAME_PATTERN.search(name):
        return {"name": name, "value": "[redacted]"}
    value = _clean_audit_value(str(raw_value), max_length)
    if value is None:
        return {"name": name, "value": ""}
    return {"name": name, "value": value}


def _query_param_items(request: Any) -> list[dict[str, str]] | None:
    query_params = getattr(request, "query_params", None)
    if not query_params:
        return None
    if hasattr(query_params, "multi_items"):
        raw_items = query_params.multi_items()
    else:
        raw_items = query_params.items()
    items = [
        item
        for item in (_sanitize_named_value(name, value) for name, value in raw_items)
        if item is not None
    ][:50]
    return items or None


def _sanitized_query_string(items: list[dict[str, str]] | None) -> str | None:
    if not items:
        return None
    value = "&".join(f"{item['name']}={item['value']}" for item in items)
    return value[:2000] or None


def _request_client_snapshot(request: Any) -> dict[str, str | int | None] | None:
    client = getattr(request, "client", None)
    if not client:
        return None
    host = _clean_audit_value(getattr(client, "host", None), 120)
    port = getattr(client, "port", None)
    if host is None and port is None:
        return None
    return {"host": host, "port": port}


def build_request_snapshot(request: Any, *, request_id: str | None = None) -> dict[str, Any]:
    headers = getattr(request, "headers", {})
    scope = getattr(request, "scope", {}) or {}
    url = getattr(request, "url", None)
    query_params = _query_param_items(request)
    path = getattr(url, "path", None) or scope.get("path")
    method = getattr(request, "method", None) or scope.get("method")
    snapshot: dict[str, Any] = {
        "request_id": request_id,
        "method": _clean_audit_value(str(method).upper() if method else None, 12),
        "path": _clean_audit_value(path, 500),
        "query_string": _sanitized_query_string(query_params),
        "query_params": query_params,
        "headers": build_sanitized_request_headers(request),
        "http_version": _clean_audit_value(scope.get("http_version"), 16),
        "scheme": _clean_audit_value(getattr(url, "scheme", None) or scope.get("scheme"), 16),
        "client": _request_client_snapshot(request),
        "content": {
            "type": _clean_audit_value(headers.get("content-type"), 200),
            "length": _clean_audit_value(headers.get("content-length"), 32),
        },
        "body": {
            "captured": False,
            "reason": "body_not_captured_by_audit_policy",
        },
    }
    return {key: value for key, value in snapshot.items() if value not in (None, {}, [])}


def _parse_ip(value: str | None):
    try:
        return ipaddress.ip_address((value or "").strip())
    except ValueError:
        return None


def _trusted_proxy_networks():
    networks = []
    for value in settings.TRUSTED_PROXY_CIDRS.split(","):
        candidate = value.strip()
        if not candidate:
            continue
        try:
            networks.append(ipaddress.ip_network(candidate, strict=False))
        except ValueError:
            continue
    return tuple(networks)


def _is_trusted_proxy(value: str | None) -> bool:
    address = _parse_ip(value)
    return bool(address and any(address in network for network in _trusted_proxy_networks()))


def resolve_client_ip(request: Any) -> str | None:
    peer_ip = request.client.host if request.client else None
    if not _is_trusted_proxy(peer_ip):
        return _clean_header_value(peer_ip, 45)

    forwarded = request.headers.get("x-forwarded-for")
    if forwarded:
        chain = [part.strip() for part in forwarded.split(",") if _parse_ip(part)]
        chain.append(str(peer_ip))
        for candidate in reversed(chain):
            if not _is_trusted_proxy(candidate):
                return _clean_header_value(candidate, 45)

    real_ip = request.headers.get("x-real-ip")
    if _parse_ip(real_ip):
        return _clean_header_value(real_ip.strip(), 45)
    return _clean_header_value(peer_ip, 45)


def build_request_audit_context(request: Any) -> RequestAuditContext:
    headers = request.headers
    request_id = str(uuid.uuid4())
    return RequestAuditContext(
        request_id=request_id,
        client_ip=resolve_client_ip(request),
        user_agent=_clean_header_value(headers.get("user-agent"), 500),
        client_type=resolve_ctms_client_type(headers),
        client_version=_clean_header_value(headers.get("x-ctms-client-version"), 32),
        client_platform=_clean_header_value(headers.get("x-ctms-client-platform"), 16),
        build_channel=_clean_header_value(headers.get("x-ctms-build-channel"), 16),
        build_commit=_clean_header_value(headers.get("x-ctms-build-commit"), 64),
        request_headers=build_sanitized_request_headers(request),
        request_snapshot=build_request_snapshot(request, request_id=request_id),
    )


def set_request_audit_context(context: RequestAuditContext) -> Token[RequestAuditContext | None]:
    return _request_audit_context.set(context)


def reset_request_audit_context(token: Token[RequestAuditContext | None]) -> None:
    _request_audit_context.reset(token)


def get_request_audit_context() -> RequestAuditContext | None:
    return _request_audit_context.get()