Cheng Zhou
74 lines
2.5 KiB
Python
74 lines
2.5 KiB
Python
"""Request-scoped metadata used by server-side audit writers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from contextvars import ContextVar, Token
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RequestAuditContext:
|
|
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_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)=([^&\s]+)"
|
|
)
|
|
|
|
|
|
def _clean_header_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 resolve_client_ip(request: Any) -> str | None:
|
|
forwarded = request.headers.get("x-forwarded-for")
|
|
if forwarded:
|
|
return _clean_header_value(forwarded.split(",")[0].strip(), 45)
|
|
real_ip = request.headers.get("x-real-ip")
|
|
if real_ip:
|
|
return _clean_header_value(real_ip.strip(), 45)
|
|
return _clean_header_value(request.client.host if request.client else None, 45)
|
|
|
|
|
|
def build_request_audit_context(request: Any) -> RequestAuditContext:
|
|
headers = request.headers
|
|
return RequestAuditContext(
|
|
client_ip=resolve_client_ip(request),
|
|
user_agent=_clean_header_value(headers.get("user-agent"), 500),
|
|
client_type=_clean_header_value(headers.get("x-ctms-client-type"), 16),
|
|
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),
|
|
)
|
|
|
|
|
|
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()
|