Reisacher-Forderung/app/logging_setup.py

"""Safe structured logging shared by HTTP, database and CLI entrypoints."""

import hashlib
import json
import logging
import os
import re
import sys
import threading
import time
import traceback
from contextvars import ContextVar
from datetime import datetime, timedelta, timezone
from pathlib import Path
from uuid import uuid4
from zoneinfo import ZoneInfo

from concurrent_log_handler import ConcurrentTimedRotatingFileHandler

request_context = ContextVar("request_logging_context", default=None)
db_operation = ContextVar("database_logging_operation", default=None)
db_query = ContextVar("database_logging_query", default=None)
BERLIN = ZoneInfo("Europe/Berlin")
PROJECT_ROOT = Path(__file__).resolve().parent.parent
_lock = threading.RLock()
_configuration = None
_SAFE_FIELDS = (
    "event",
    "operation",
    "query_name",
    "query_fingerprint",
    "sql",
    "sql_truncated",
    "sqlstate",
    "db_code",
    "db_message",
    "db_diagnostics",
    "status",
    "duration_ms",
    "outcome",
    "size_bytes",
    "count",
    "reason",
    "error_type",
)


def clean(value):
    return str(value).replace("\r", "\\r").replace("\n", "\\n").replace("\x00", "\\0")[:8192]


def sanitize_sql(statement):
    """Lex T-SQL literals/comments without modifying quoted identifiers or bind names."""
    source, output, i = str(statement), [], 0
    while i < len(source):
        char = source[i]
        if source.startswith("--", i):
            end = source.find("\n", i + 2)
            i = len(source) if end < 0 else end
            output.append(" ")
        elif source.startswith("/*", i):
            depth = 1
            i += 2
            while i < len(source) and depth:
                if source.startswith("/*", i):
                    depth += 1
                    i += 2
                elif source.startswith("*/", i):
                    depth -= 1
                    i += 2
                else:
                    i += 1
            output.append(" ")
        elif char == "'" or (
            char in "nN" and source[i : i + 2].endswith("'") and (i == 0 or not source[i - 1].isalnum())
        ):
            if char != "'":
                i += 1
            i += 1
            while i < len(source):
                if source[i : i + 2] == "''":
                    i += 2
                elif source[i] == "'":
                    i += 1
                    break
                else:
                    i += 1
            output.append("?")
        elif char in '["':
            end_char = "]" if char == "[" else '"'
            start = i
            i += 1
            while i < len(source):
                if source[i : i + 2] == end_char * 2:
                    i += 2
                elif source[i] == end_char:
                    i += 1
                    break
                else:
                    i += 1
            output.append(source[start:i] if char == "[" else "?")
        elif (char.isdecimal() or char == "." and i + 1 < len(source) and source[i + 1].isdecimal()) and (
            i == 0 or not (source[i - 1].isalnum() or source[i - 1] in "_:@$")
        ):
            match = re.match(r"(?:0[xX][0-9a-fA-F]+|(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?)", source[i:])
            output.append("?")
            i += len(match[0])
        else:
            output.append(char)
            i += 1
    return " ".join("".join(output).split())


def exception_roots(error):
    seen = set()

    def visit(exc):
        if id(exc) in seen:
            return []
        seen.add(id(exc))
        if isinstance(exc, BaseExceptionGroup):
            return [root for child in exc.exceptions for root in visit(child)]
        original = getattr(exc, "orig", None) or exc.__cause__
        return visit(original) if isinstance(original, BaseException) else [exc]

    return visit(error)


def frames(tb=None):
    extracted = traceback.extract_tb(tb) if tb else traceback.extract_stack()[:-1]
    return [dict(file=Path(item.filename).name, line=item.lineno, function=item.name) for item in extracted[-45:]]


class ContextFilter(logging.Filter):
    def filter(self, record):
        if record.name.startswith("uvicorn.") and record.levelno < logging.INFO:
            return False
        if (
            not record.name.startswith(("app.", "uvicorn.", "__main__", "filter_export"))
            and record.levelno < logging.WARNING
        ):
            return False
        if not hasattr(record, "event_id"):
            record.event_id = uuid4().hex
        context = request_context.get()
        request_id = context["request_id"] if context else None
        if record.exc_info and record.exc_info[1]:
            roots = exception_roots(record.exc_info[1])
            fresh = [
                exc
                for exc in roots
                if getattr(exc, "_gc_log_event", None) in (None, record.event_id)
                or request_id is not None
                and getattr(exc, "_gc_log_request", None) != request_id
            ]
            if not fresh:
                return False
            for exc in fresh:
                exc._gc_log_event = record.event_id
                exc._gc_log_request = request_id
            record.safe_exceptions = [
                dict(
                    type=type(exc).__module__ + "." + type(exc).__name__,
                    frames=getattr(exc, "_gc_log_frames", None) or frames(exc.__traceback__),
                )
                for exc in fresh
            ]
        record.request_fields = {}
        if context:
            scope = context["scope"]
            principal = scope.get("state", {}).get("principal")
            record.request_fields = dict(
                request_id=context["request_id"],
                user=getattr(principal, "username", None),
                method=scope["method"],
                path=scope.get("path", ""),
            )
        return True


def record_data(record):
    data = dict(
        timestamp=datetime.fromtimestamp(record.created, BERLIN).isoformat(timespec="milliseconds"),
        level=record.levelname,
        logger=record.name,
        event_id=record.event_id,
        event=getattr(record, "event", "application"),
        message=(
            clean(record.getMessage())
            if record.name.startswith(("app.", "uvicorn.", "__main__", "filter_export"))
            else "External logger event"
        ),
    )
    data.update(
        {key: clean(value) if isinstance(value, str) else value for key, value in record.request_fields.items()}
    )
    data.update({key: getattr(record, key) for key in _SAFE_FIELDS if hasattr(record, key)})
    if hasattr(record, "safe_exceptions"):
        data["exceptions"] = record.safe_exceptions
    return data


class JsonFormatter(logging.Formatter):
    def format(self, record):
        return json.dumps(
            record_data(record), ensure_ascii=True, separators=(",", ":"), default=lambda _: "[unsupported]"
        )


class ConsoleFormatter(logging.Formatter):
    def format(self, record):
        data = record_data(record)
        suffix = {key: value for key, value in data.items() if key not in ("timestamp", "level", "logger", "message")}
        suffix_text = json.dumps(suffix, ensure_ascii=True, separators=(",", ":")) if suffix else ""
        return f"{data['timestamp']} {data['level']} {data['logger']}: {data['message']} {suffix_text}".rstrip()


class SafeConsoleHandler(logging.StreamHandler):
    def emit(self, record):
        super().emit(record)
        record._gc_console_emitted = True

    def handleError(self, record):
        pass  # A broken console must not change application behavior or expose raw LogRecords.


class ApplicationFileHandler(ConcurrentTimedRotatingFileHandler):
    def __init__(self, filename, retention_days=30):
        self.retention_days = retention_days
        self.last_fallback = float("-inf")
        super().__init__(
            filename=filename,
            when="midnight",
            interval=1,
            utc=True,
            backupCount=retention_days,
            encoding="utf-8",
            maxBytes=0,
        )
        self.cleanup_archives()

    def cleanup_archives(self):
        cutoff = datetime.now(timezone.utc).date() - timedelta(days=self.retention_days)
        path = Path(self.baseFilename)
        for candidate in path.parent.glob(path.name + ".*"):
            suffix = candidate.name[len(path.name) + 1 :]
            if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", suffix):
                continue
            try:
                if datetime.strptime(suffix, "%Y-%m-%d").date() < cutoff:
                    candidate.unlink(missing_ok=True)
            except (OSError, ValueError):
                self.fallback(None, "log_cleanup_failed")

    def doRollover(self):
        try:
            super().doRollover()
        except Exception:
            self.fallback(None, "log_rotation_failed")
            raise
        self.cleanup_archives()

    def fallback(self, record, event="log_file_unavailable"):
        handler = SafeConsoleHandler(sys.stderr)
        handler.setFormatter(ConsoleFormatter())
        if time.monotonic() - self.last_fallback >= 60:
            self.last_fallback = time.monotonic()
            notice = logging.LogRecord(
                "app.logging", logging.ERROR, __file__, 0, "Logdatei nicht verfügbar; Konsolenfallback aktiv", (), None
            )
            notice.event = event
            ContextFilter().filter(notice)
            handler.handle(notice)
        if record is not None and not getattr(record, "_gc_console_emitted", False):
            handler.handle(record)

    def handleError(self, record):
        self.fallback(record)


def configure_logging(config=None, *, console_stream=None):
    """Replace only our own handlers; repeated startup in one process is a no-op."""
    global _configuration
    if config is None:
        from .config import settings

        config = settings
    directory = Path(config.LOG_DIR)
    if not directory.is_absolute():
        directory = PROJECT_ROOT / directory
    signature = (
        os.getpid(),
        str(directory),
        config.LOG_LEVEL,
        config.LOG_TO_FILE,
        config.LOG_TO_CONSOLE,
        config.LOG_RETENTION_DAYS,
        config.LOG_SQL_STATEMENT,
        id(console_stream),
    )
    with _lock:
        root = logging.getLogger()
        if _configuration == signature and any(getattr(handler, "_gc_handler", False) for handler in root.handlers):
            return
        for handler in list(root.handlers):
            if getattr(handler, "_gc_handler", False):
                root.removeHandler(handler)
                handler.close()
        handlers, file_failure = [], False
        if config.LOG_TO_CONSOLE:
            console = SafeConsoleHandler(console_stream or sys.stderr)
            console.setFormatter(ConsoleFormatter())
            handlers.append(console)
        if config.LOG_TO_FILE:
            try:
                directory.mkdir(parents=True, exist_ok=True)
                file_handler = ApplicationFileHandler(str(directory / "application.log"), config.LOG_RETENTION_DAYS)
                file_handler.setFormatter(JsonFormatter())
                handlers.append(file_handler)
            except Exception:
                file_failure = True
                if not handlers:
                    console = SafeConsoleHandler(console_stream or sys.stderr)
                    console.setFormatter(ConsoleFormatter())
                    handlers.append(console)
        if not handlers:
            handlers.append(logging.NullHandler())
        for handler in handlers:
            handler._gc_handler = True
            handler.addFilter(ContextFilter())
            root.addHandler(handler)
        root.setLevel(config.LOG_LEVEL)
        for name in ("uvicorn", "uvicorn.error", "uvicorn.access"):
            logger = logging.getLogger(name)
            logger.handlers.clear()
            logger.propagate = True
            logger.setLevel(logging.INFO)
        logging.getLogger("uvicorn.access").disabled = True
        for name in ("sqlalchemy", "urllib3", "httpx", "httpcore", "ldap3", "docuware"):
            logging.getLogger(name).setLevel(logging.WARNING)
        _configuration = signature
        if file_failure:
            logging.getLogger("app.logging").error(
                "Logverzeichnis nicht verfügbar; Konsolenfallback aktiv", extra={"event": "log_file_unavailable"}
            )


def database_diagnostics(error, statement=None, parameters=None):
    """Read ODBC diagnostics from the driver, never from SQLAlchemy's formatted exception."""
    if type(error).__module__ != "pyodbc":
        return []
    state = (
        error.args[0]
        if error.args and isinstance(error.args[0], str) and re.fullmatch(r"[A-Z0-9]{5}", error.args[0])
        else None
    )
    secrets = set()

    def collect(value):
        if isinstance(value, str) and value:
            secrets.add(value)
        elif isinstance(value, bytes):
            collect(value.decode("utf-8", errors="replace"))
        elif isinstance(value, dict):
            for child in value.values():
                collect(child)
        elif isinstance(value, (tuple, list)):
            for child in value:
                collect(child)

    collect(parameters)
    if statement is not None:
        for match in re.finditer(r"N?'((?:''|[^'])*)'", str(statement), re.IGNORECASE):
            collect(match[1].replace("''", "'"))

    diagnostics = []
    for raw in error.args[1:] if state else error.args:
        if not isinstance(raw, str):
            continue
        # pyodbc concatenates SQLGetDiagRec records, appending native codes and the ODBC function.
        parts = list(re.finditer(r"(.*?)\s+\((-?\d+)\)(?:\s+\(SQL\w+\))?(?=;\s*|$)", raw, re.DOTALL))
        records = [(part[1].lstrip("; \t\r\n"), int(part[2])) for part in parts] or [(raw.strip(), None)]
        for description, code in records:
            own_state = re.match(r"\[([A-Z0-9]{5})\]", description)
            description = re.sub(r"^(?:\[[^\]\r\n]*\]\s*)+", "", description)
            # Error 207 identifies schema metadata, not a parameter value. Only exempt
            # the canonical English/German message, never arbitrary text with code 207.
            column_message = (
                description
                if code == 207
                and re.fullmatch(
                    r"""(?:Invalid column name|Ungültiger Spaltenname)\s+(?:'(?:''|[^'])*'|"(?:""|[^"])*"|„[^“]*“|“[^”]*”|‚[^‘]*‘|‘[^’]*’|«[^»]*»)\s*\.?""",
                    description.strip(),
                    re.IGNORECASE,
                )
                else None
            )
            # Values may also be echoed without quotes (notably duplicate key tuples).
            for secret in sorted(secrets, key=len, reverse=True):
                description = description.replace(secret, "?").replace(secret.replace("'", "''"), "?")
            description = re.sub(r"\([^)]*(?:\)|$)", "(?)", description)
            description = re.sub(r"(?i)\b(password|pwd|token|secret)\s*[:=]\s*[^;\s]+", r"\1=?", description)
            description = clean(column_message if column_message is not None else sanitize_sql(description))[:4096]
            diagnostics.append(dict(sqlstate=own_state[1] if own_state else state, code=code, message=description))
    return diagnostics[:20]


def log_query_error(error, statement=None, query_name=None, operation=None, parameters=None):
    from .config import settings

    try:
        sql = sanitize_sql(statement) if statement is not None else ""
    except Exception:
        sql = "[statement unavailable]"
    metadata = dict(
        event="database_error",
        operation=operation or db_operation.get() or "execute",
        query_name=query_name or query_origin(),
        query_fingerprint=hashlib.sha256(sql.encode()).hexdigest(),
    )
    if settings.LOG_SQL_STATEMENT and sql:
        metadata.update(sql=sql[:8192], sql_truncated=len(sql) > 8192)
    roots = exception_roots(error)
    diagnostics = []
    for root in roots:
        if not getattr(root, "_gc_log_frames", None):
            root._gc_log_frames = frames()
        first = root.args[0] if root.args else None
        if isinstance(first, str) and re.fullmatch(r"[A-Z0-9]{5}", first):
            metadata["sqlstate"] = first
        code = getattr(root, "sqlite_errorcode", None)
        if isinstance(code, int):
            metadata["db_code"] = code
        try:
            diagnostics.extend(
                database_diagnostics(
                    root, statement, parameters if parameters is not None else getattr(error, "params", None)
                )
            )
        except Exception:
            pass  # A malformed driver diagnostic must not replace the original database error.
    if diagnostics:
        metadata["db_diagnostics"] = diagnostics[:20]
        metadata["db_message"] = diagnostics[0]["message"]
        if diagnostics[0]["code"] is not None:
            metadata["db_code"] = diagnostics[0]["code"]
    logging.getLogger("app.db").error(
        "Datenbankoperation fehlgeschlagen", exc_info=(type(error), error, error.__traceback__), extra=metadata
    )


def query_origin():
    for item in reversed(traceback.extract_stack()[:-1]):
        path = Path(item.filename)
        if PROJECT_ROOT / "app" in path.parents and path.name not in ("logging_setup.py", "db.py"):
            return f"{path.name}:{item.name}"
    return "database"


def run_cli(callback):
    configure_logging()
    try:
        callback()
    except Exception:
        logging.getLogger("app.cli").exception("Hilfsprogramm fehlgeschlagen", extra={"event": "cli_failed"})
        raise SystemExit(1) from None