learningcircuit--local-deep-research/.pre-commit-hooks/check-sensitive-logging.py

#!/usr/bin/env python3
"""
Pre-commit hook to detect potential logging of sensitive data.
Prevents passwords, tokens, and other sensitive information from being logged.

Note: this hook is part of the reason we do NOT enforce ``raise X from e``
universally (see ADR-0003). Exception chains preserved via ``from e`` can
leak PII through error-tracking services and downstream handlers. Code that
wraps user-facing exceptions should omit ``from e`` to break the chain.
"""

import ast
import sys
from pathlib import Path
from typing import List, Optional


SENSITIVE_VARS = {
    "password",
    "user_password",
    "pwd",
    "secret",
    "api_key",
    "private_key",
    "auth",
    "credential",
    "csrf_token",  # CSRF tokens should not be logged
    "bearer_token",
    "access_token",
    "refresh_token",
    "auth_token",
    "session_id",  # Can be used to retrieve passwords
    "encryption_key",
    "secret_key",
    "auth_header",
}

SENSITIVE_DICT_KEYS = {
    "shared_research_context",  # Contains user_password for SQLCipher
    "settings_snapshot",  # Contains API keys and other sensitive data
    "kwargs",  # Often contains user_password and gets passed through many functions
    "credentials",  # Obviously contains authentication data
    "auth_data",  # Authentication related data
    "auth_context",  # Authentication context
    "user_data",  # May contain user passwords
    "login_data",  # Login form data including passwords
    "form_data",  # Form submissions may contain passwords
    "post_data",  # POST request data
    "payload",  # API request payloads
    "body",  # Request body data
    "session",  # Flask session data (when used as dict)
    "flask_session",  # Explicit Flask session
    "g.user_password",  # Flask g object with password
    "connection_string",  # Database connection strings with passwords
    "conn_str",
    "db_url",
    "database_url",
    "conn_params",
    "db_config",
    "environ",  # os.environ contains sensitive vars
}

# These are false positives - specific variable names that are safe to log
FALSE_POSITIVE_VARS = {
    # Token counting related - these are LLM tokens, not auth tokens
    "prompt_tokens",
    "completion_tokens",
    "token_research_count",
    "tokens_truncated",
    "estimated_prompt_tokens",
    "estimated_completion_tokens",
    "max_tokens",
    "new_max_tokens",
    # Configuration paths and names (not the actual config contents)
    "config_path",
    "config_name",
    "config_hash",
    "context_window_size",
    # Database paths (not credentials)
    "auth_db_path",
    # Safe/redacted versions
    "safe_settings",  # This is already redacted
}

# Specific file + variable combinations that are allowed
# Each entry must have an explanation of why it's safe
ALLOWED_LOGGING = [
    {
        "file_pattern": "settings/logger.py",
        "variable": "safe_settings",
        "reason": "This variable contains already redacted settings with sensitive values replaced by ***REDACTED***",
    },
    {
        "file_pattern": "settings/logger.py",
        "variable": "settings",
        "reason": "This module is responsible for safe logging and applies redaction before logging when log_level is 'debug_unsafe'",
    },
    {
        "file_pattern": "tests/",
        "variable": "csrf_token",
        "reason": "Test files need to log CSRF tokens for debugging authentication issues during test development",
    },
    {
        "file_pattern": "web/services/socket_service.py",
        "variable": "kwargs",
        "reason": "These are logger wrapper methods where kwargs are logger configuration options (like exc_info, extra), not actual data being logged",
    },
    {
        "file_pattern": "search_engine_openlibrary.py",
        "variable": "author_key",
        "reason": "OpenLibrary author identifier (e.g. /authors/OL123A), not a cryptographic key",
    },
    {
        "file_pattern": "web/queue/processor_v2.py",
        "variable": "user_session",
        "reason": "Queue routing key (username:session_id format), not session contents",
    },
]


# Directories where ``logger.exception()`` routes through the diagnose-gated
# ``security.secure_logging`` wrapper: the message is production-visible at
# ERROR level (only the traceback is gated), so interpolating the raw
# exception variable leaks error details there just like any other level
# (#4183). Elsewhere, plain loguru ``.exception()`` output remains dev-only.
SECURE_LOGGING_DIRS = (
    "src/local_deep_research/llm/providers/",
    "src/local_deep_research/embeddings/providers/",
    "src/local_deep_research/web_search_engines/",
)

# The only sanctioned logger import inside SECURE_LOGGING_DIRS. Matched
# exactly (module + level), never by suffix, so lookalike modules such as
# ``notsecurity.secure_logging`` cannot spoof it.
WRAPPER_MODULE_RELATIVE = "security.secure_logging"
WRAPPER_MODULE_ABSOLUTE = "local_deep_research.security.secure_logging"
WRAPPER_IMPORT_HINT = (
    "use 'from ...security.secure_logging import logger' "
    "(relative depth per file)"
)

# Attribute names reserved inside SECURE_LOGGING_DIRS: routes from the
# SecureLogger wrapper (or any object) back to raw loguru handles.
#   .opt / .catch      — wrapper delegates these to raw loguru ungated
#   .patch             — patchers can mutate record["exception"] and attach
#                        an ungated traceback
#   .logger            — raw singleton re-exports (log_utils.logger,
#                        local_deep_research.logger) and future self.logger
#   ._logger           — SecureLogger's wrapped handle (__slots__)
#   ._loguru_logger    — raw singleton inside secure_logging
#   .__getattr__       — logger.__getattr__("opt") delegates to raw loguru
BANNED_LOGGER_ATTRS = {
    "opt",
    "catch",
    "patch",
    "logger",
    "_logger",
    "_loguru_logger",
    "__getattr__",
}

# Traceback-formatting attribute names: even when the ``traceback`` module
# import ban is dodged (dynamic import, re-export), these calls produce raw
# traceback text that must never reach production logs.
BANNED_TRACEBACK_ATTRS = {
    "format_exc",
    "format_exception",
    "format_exception_only",
    "format_tb",
    "print_exc",
    "print_exception",
    "print_tb",
}

# Module names whose dynamic import (import_module / __import__) is banned in
# SECURE_LOGGING_DIRS — same modules the static import bans cover.
BANNED_DYNAMIC_IMPORTS = {"loguru", "traceback"}

# Wrapper-preserving chain methods safe for secure dirs: SecureLogger.bind()
# re-wraps without letting callers mutate the raw loguru record. Messages are
# still production-visible and must pass the exception-variable check.
WRAPPER_CHAIN_METHODS = {"bind"}


class SensitiveLoggingChecker(ast.NodeVisitor):
    """AST visitor to detect sensitive data in logging statements."""

    def __init__(self, filename: str):
        self.filename = filename
        self.errors: List[str] = []
        # Track exception variable names from enclosing except handlers
        self._except_var_stack: List[Optional[str]] = []
        # Scope-local names that refer to the sys module. Used to catch
        # ``import sys as s; s.exc_info()`` without leaking local aliases into
        # unrelated functions.
        self._sys_name_scopes = [{"sys"}]
        self._shadowed_name_scopes = [set()]
        self._name_scope_kinds = ["module"]
        # Single normalized dir check reused by every dir-scoped rule so
        # Windows paths cannot diverge between checks.
        self.in_secure_dir = any(
            d in filename.replace("\\", "/") for d in SECURE_LOGGING_DIRS
        )

    def visit_ExceptHandler(self, node: ast.ExceptHandler) -> None:
        """Track exception variable names inside except blocks."""
        if self.in_secure_dir and node.name == "logger":
            self.errors.append(
                f"{self.filename}:{node.lineno}: "
                f"'except ... as logger' shadows the secure logger — "
                f"rename the exception variable"
            )
        previous_shadowed_names = set(self._shadowed_name_scopes[-1])
        if self.in_secure_dir and node.name:
            self._mark_name_shadow(node.name)
        self._except_var_stack.append(node.name)
        self.generic_visit(node)
        self._except_var_stack.pop()
        self._shadowed_name_scopes[-1] = previous_shadowed_names

    def _push_name_scope(self, shadowed_names=(), kind: str = "block") -> None:
        self._sys_name_scopes.append(set())
        self._shadowed_name_scopes.append(set(shadowed_names))
        self._name_scope_kinds.append(kind)

    def _pop_name_scope(self) -> None:
        self._sys_name_scopes.pop()
        self._shadowed_name_scopes.pop()
        self._name_scope_kinds.pop()

    def _name_refers_to_sys(self, name: str) -> bool:
        nearest_function_scope = max(
            (
                idx
                for idx, kind in enumerate(self._name_scope_kinds)
                if kind == "function"
            ),
            default=-1,
        )
        for idx in range(len(self._sys_name_scopes) - 1, -1, -1):
            if (
                self._name_scope_kinds[idx] == "class"
                and nearest_function_scope > idx
            ):
                continue
            sys_names = self._sys_name_scopes[idx]
            shadowed_names = self._shadowed_name_scopes[idx]
            if name in shadowed_names:
                return False
            if name in sys_names:
                return True
        return False

    def _mark_name_shadow(self, name: str) -> None:
        self._sys_name_scopes[-1].discard(name)
        self._shadowed_name_scopes[-1].add(name)

    def _mark_target_shadow(self, target: ast.AST) -> None:
        if isinstance(target, ast.Name):
            self._mark_name_shadow(target.id)
        elif isinstance(target, (ast.Tuple, ast.List)):
            for elt in target.elts:
                self._mark_target_shadow(elt)
        elif isinstance(target, ast.Starred):
            self._mark_target_shadow(target.value)

    @staticmethod
    def _is_literal_dynamic_import(node: ast.AST, module: str) -> bool:
        if not isinstance(node, ast.Call) or not node.args:
            return False
        func = node.func
        is_dynamic_import = (
            (isinstance(func, ast.Attribute) and func.attr == "import_module")
            or (isinstance(func, ast.Name) and func.id == "import_module")
            or (isinstance(func, ast.Name) and func.id == "__import__")
        )
        return (
            is_dynamic_import
            and isinstance(node.args[0], ast.Constant)
            and node.args[0].value == module
        )

    def _expr_is_sys_module(self, node: ast.AST) -> bool:
        return (
            isinstance(node, ast.Name) and self._name_refers_to_sys(node.id)
        ) or self._is_literal_dynamic_import(node, "sys")

    # ------------------------------------------------------------------
    # Wrapper-import enforcement (SECURE_LOGGING_DIRS only, #4183 step 2)
    # ------------------------------------------------------------------

    def visit_Import(self, node: ast.Import) -> None:
        """Ban raw-logger import routes in secure-logging dirs."""
        if self.in_secure_dir:
            for alias in node.names:
                root = alias.name.split(".")[0]
                leaf = alias.name.split(".")[-1]
                bound = alias.asname or root
                if alias.name == "sys" or (
                    root == "sys" and alias.asname is None
                ):
                    # `import sys`, `import sys as s`, and submodule forms
                    # like `import sys.monitoring` (which bind the root
                    # name `sys`) all make the bound name refer to sys.
                    self._sys_name_scopes[-1].add(bound)
                    self._shadowed_name_scopes[-1].discard(bound)
                else:
                    self._mark_name_shadow(bound)
                if root == "loguru":
                    self.errors.append(
                        f"{self.filename}:{node.lineno}: "
                        f"raw loguru import is banned in secure-logging dirs — "
                        f"{WRAPPER_IMPORT_HINT}"
                    )
                elif leaf == "secure_logging":
                    self.errors.append(
                        f"{self.filename}:{node.lineno}: "
                        f"importing the secure_logging module exposes raw "
                        f"handles (secure_logging._loguru_logger) — "
                        f"{WRAPPER_IMPORT_HINT}"
                    )
                elif leaf == "log_utils":
                    self.errors.append(
                        f"{self.filename}:{node.lineno}: "
                        f"importing utilities.log_utils exposes the raw "
                        f"loguru logger (log_utils.logger) — "
                        f"{WRAPPER_IMPORT_HINT}"
                    )
                elif root == "local_deep_research" and alias.asname is None:
                    self.errors.append(
                        f"{self.filename}:{node.lineno}: "
                        f"absolute 'import local_deep_research...' binds the "
                        f"root package, whose .logger attribute is the raw "
                        f"loguru logger — use relative imports; "
                        f"{WRAPPER_IMPORT_HINT}"
                    )
                elif root == "traceback":
                    self.errors.append(
                        f"{self.filename}:{node.lineno}: "
                        f"'import traceback' is banned in secure-logging dirs — "
                        f"tracebacks must never be interpolated into log "
                        f"messages; use logger.exception() (diagnose-gated) "
                        f"with a scrubbed safe_msg"
                    )
                if bound == "logger":
                    self.errors.append(
                        f"{self.filename}:{node.lineno}: "
                        f"import binds the local name 'logger' — only the "
                        f"secure_logging wrapper may bind 'logger' here; "
                        f"{WRAPPER_IMPORT_HINT}"
                    )
        self.generic_visit(node)

    def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
        """Allow only the exact secure_logging wrapper import of 'logger'."""
        if self.in_secure_dir:
            self._check_import_from_secure_dir(node)
        self.generic_visit(node)

    def _check_import_from_secure_dir(self, node: ast.ImportFrom) -> None:
        module = node.module or ""
        is_wrapper_module = (
            node.level > 0 and module == WRAPPER_MODULE_RELATIVE
        ) or (node.level == 0 and module == WRAPPER_MODULE_ABSOLUTE)
        for alias in node.names:
            if alias.name != "*":
                self._mark_name_shadow(alias.asname or alias.name)

        if module.split(".")[0] == "loguru":
            self.errors.append(
                f"{self.filename}:{node.lineno}: "
                f"raw loguru import is banned in secure-logging dirs — "
                f"{WRAPPER_IMPORT_HINT}"
            )
            return

        if node.level == 0 and module == "traceback":
            self.errors.append(
                f"{self.filename}:{node.lineno}: "
                f"'from traceback import ...' is banned in secure-logging "
                f"dirs — tracebacks must never be interpolated into log "
                f"messages; use logger.exception() (diagnose-gated) with a "
                f"scrubbed safe_msg"
            )
            return

        if node.level == 0 and module == "sys":
            for alias in node.names:
                if alias.name in ("*", "exc_info"):
                    self.errors.append(
                        f"{self.filename}:{node.lineno}: "
                        f"'from sys import exc_info' is banned in "
                        f"secure-logging dirs — tracebacks must never be "
                        f"interpolated into log messages; use "
                        f"logger.exception() (diagnose-gated) with a "
                        f"scrubbed safe_msg"
                    )

        if is_wrapper_module:
            for alias in node.names:
                if alias.name != "logger":
                    self.errors.append(
                        f"{self.filename}:{node.lineno}: "
                        f"importing '{alias.name}' from secure_logging is "
                        f"banned — only the unaliased 'logger' proxy may be "
                        f"imported in secure-logging dirs"
                    )
                elif alias.asname not in (None, "logger"):
                    self.errors.append(
                        f"{self.filename}:{node.lineno}: "
                        f"aliasing the secure logger to "
                        f"'{alias.asname}' is banned — the local name must "
                        f"remain 'logger'"
                    )
            return

        # Module-import routes to raw handles
        if (node.level > 0 and module == "utilities.log_utils") or (
            node.level == 0
            and module == "local_deep_research.utilities.log_utils"
        ):
            self.errors.append(
                f"{self.filename}:{node.lineno}: "
                f"importing from utilities.log_utils is banned in "
                f"secure-logging dirs — it re-exports the raw loguru "
                f"logger; {WRAPPER_IMPORT_HINT}"
            )
            return

        for alias in node.names:
            if alias.name == "logger":
                self.errors.append(
                    f"{self.filename}:{node.lineno}: "
                    f"importing 'logger' from a module other than "
                    f"security.secure_logging is banned in secure-logging "
                    f"dirs — {WRAPPER_IMPORT_HINT}"
                )
            elif alias.name == "secure_logging" and (
                (node.level > 0 and module == "security")
                or (
                    node.level == 0 and module == "local_deep_research.security"
                )
            ):
                self.errors.append(
                    f"{self.filename}:{node.lineno}: "
                    f"importing the secure_logging module exposes raw "
                    f"handles (secure_logging._loguru_logger) — "
                    f"{WRAPPER_IMPORT_HINT}"
                )
            elif alias.name == "log_utils" and (
                (node.level > 0 and module == "utilities")
                or (
                    node.level == 0
                    and module == "local_deep_research.utilities"
                )
            ):
                self.errors.append(
                    f"{self.filename}:{node.lineno}: "
                    f"importing utilities.log_utils exposes the raw loguru "
                    f"logger (log_utils.logger) — {WRAPPER_IMPORT_HINT}"
                )
            elif alias.asname == "logger":
                self.errors.append(
                    f"{self.filename}:{node.lineno}: "
                    f"import binds the local name 'logger' — only the "
                    f"secure_logging wrapper may bind 'logger' here; "
                    f"{WRAPPER_IMPORT_HINT}"
                )

    # ------------------------------------------------------------------
    # Logger rebinding / shadowing bans (SECURE_LOGGING_DIRS only)
    # ------------------------------------------------------------------

    def _flag_logger_binding(self, target: ast.AST, lineno: int) -> None:
        """Flag any assignment target that binds the name 'logger'."""
        if isinstance(target, ast.Name) and target.id == "logger":
            self.errors.append(
                f"{self.filename}:{lineno}: "
                f"rebinding the name 'logger' is banned in secure-logging "
                f"dirs — only the secure_logging wrapper import may bind "
                f"'logger'"
            )
        elif isinstance(target, (ast.Tuple, ast.List)):
            for elt in target.elts:
                self._flag_logger_binding(elt, lineno)
        elif isinstance(target, ast.Starred):
            self._flag_logger_binding(target.value, lineno)

    def _check_logger_derivation(self, value: ast.AST, lineno: int) -> None:
        """Flag assignments that derive a new local name from a logger.

        Covers ``log = logger``, ``log = logger.bind(...)``,
        ``x = log_utils.logger``, ``log = pkg.logger.bind(...)``.
        """
        expr = value
        while (
            isinstance(expr, ast.Call)
            and isinstance(expr.func, ast.Attribute)
            and expr.func.attr in WRAPPER_CHAIN_METHODS
        ):
            expr = expr.func.value
        if (isinstance(expr, ast.Name) and expr.id == "logger") or (
            isinstance(expr, ast.Attribute) and expr.attr == "logger"
        ):
            if isinstance(expr, ast.Attribute):
                # ``x = pkg.logger`` is already reported by visit_Attribute's
                # .logger ban; avoid a second diagnostic for the same line.
                return
            self.errors.append(
                f"{self.filename}:{lineno}: "
                f"assigning a logger to another name is banned in "
                f"secure-logging dirs — aliased loggers evade the "
                f"exception-variable checks; call 'logger' directly"
            )

    def visit_Assign(self, node: ast.Assign) -> None:
        if self.in_secure_dir:
            for target in node.targets:
                self._flag_logger_binding(target, node.lineno)
            self._check_logger_derivation(node.value, node.lineno)
            for target in node.targets:
                self._mark_target_shadow(target)
        self.generic_visit(node)

    def visit_AnnAssign(self, node: ast.AnnAssign) -> None:
        if self.in_secure_dir:
            self._flag_logger_binding(node.target, node.lineno)
            if node.value is not None:
                self._check_logger_derivation(node.value, node.lineno)
            self._mark_target_shadow(node.target)
        self.generic_visit(node)

    def visit_AugAssign(self, node: ast.AugAssign) -> None:
        if self.in_secure_dir:
            self._flag_logger_binding(node.target, node.lineno)
            self._mark_target_shadow(node.target)
        self.generic_visit(node)

    def visit_NamedExpr(self, node: ast.NamedExpr) -> None:
        if self.in_secure_dir:
            self._flag_logger_binding(node.target, node.lineno)
            self._check_logger_derivation(node.value, node.lineno)
            self._mark_target_shadow(node.target)
        self.generic_visit(node)

    def visit_For(self, node: ast.For) -> None:
        if self.in_secure_dir:
            self._flag_logger_binding(node.target, node.lineno)
            self._mark_target_shadow(node.target)
        self.generic_visit(node)

    def visit_AsyncFor(self, node: ast.AsyncFor) -> None:
        if self.in_secure_dir:
            self._flag_logger_binding(node.target, node.lineno)
            self._mark_target_shadow(node.target)
        self.generic_visit(node)

    def visit_With(self, node: ast.With) -> None:
        if self.in_secure_dir:
            for item in node.items:
                if item.optional_vars is not None:
                    self._flag_logger_binding(item.optional_vars, node.lineno)
                    self._mark_target_shadow(item.optional_vars)
        self.generic_visit(node)

    def visit_AsyncWith(self, node: ast.AsyncWith) -> None:
        if self.in_secure_dir:
            for item in node.items:
                if item.optional_vars is not None:
                    self._flag_logger_binding(item.optional_vars, node.lineno)
                    self._mark_target_shadow(item.optional_vars)
        self.generic_visit(node)

    @staticmethod
    def _func_param_names(args: ast.arguments) -> List[str]:
        all_args = (
            list(args.posonlyargs) + list(args.args) + list(args.kwonlyargs)
        )
        if args.vararg:
            all_args.append(args.vararg)
        if args.kwarg:
            all_args.append(args.kwarg)
        return [arg.arg for arg in all_args]

    def _check_func_params(self, node) -> None:
        args = node.args
        for arg_name in self._func_param_names(args):
            if arg_name == "logger":
                self.errors.append(
                    f"{self.filename}:{node.lineno}: "
                    f"parameter named 'logger' shadows the secure logger in "
                    f"secure-logging dirs — rename the parameter"
                )

    def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
        if self.in_secure_dir:
            self._check_func_params(node)
        self._push_name_scope(
            self._func_param_names(node.args), kind="function"
        )
        self.generic_visit(node)
        self._pop_name_scope()

    def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
        if self.in_secure_dir:
            self._check_func_params(node)
        self._push_name_scope(
            self._func_param_names(node.args), kind="function"
        )
        self.generic_visit(node)
        self._pop_name_scope()

    def visit_Lambda(self, node: ast.Lambda) -> None:
        if self.in_secure_dir:
            self._check_func_params(node)
        self._push_name_scope(
            self._func_param_names(node.args), kind="function"
        )
        self.generic_visit(node)
        self._pop_name_scope()

    def visit_ClassDef(self, node: ast.ClassDef) -> None:
        self._push_name_scope(kind="class")
        self.generic_visit(node)
        self._pop_name_scope()

    # ------------------------------------------------------------------
    # Bypass-route detection (SECURE_LOGGING_DIRS only)
    # ------------------------------------------------------------------

    def visit_Attribute(self, node: ast.Attribute) -> None:
        if self.in_secure_dir:
            if node.attr in ("opt", "catch"):
                self.errors.append(
                    f"{self.filename}:{node.lineno}: "
                    f".{node.attr} is banned in secure-logging dirs — the "
                    f"secure_logging wrapper delegates it to raw loguru, "
                    f"reattaching tracebacks ungated; use logger.exception() "
                    f"with a scrubbed safe_msg"
                )
            elif node.attr == "patch":
                self.errors.append(
                    f"{self.filename}:{node.lineno}: "
                    f".patch is banned in secure-logging dirs — patchers can "
                    f'mutate record["exception"] and reattach tracebacks '
                    f"ungated; use logger.bind() for metadata"
                )
            elif node.attr == "logger":
                self.errors.append(
                    f"{self.filename}:{node.lineno}: "
                    f".logger attribute access is banned in secure-logging "
                    f"dirs — it reaches raw loguru re-exports "
                    f"(log_utils.logger, local_deep_research.logger); "
                    f"{WRAPPER_IMPORT_HINT}"
                )
            elif node.attr in ("_logger", "_loguru_logger", "__getattr__"):
                self.errors.append(
                    f"{self.filename}:{node.lineno}: "
                    f".{node.attr} accesses a raw loguru handle — banned in "
                    f"secure-logging dirs"
                )
            elif node.attr in BANNED_TRACEBACK_ATTRS:
                self.errors.append(
                    f"{self.filename}:{node.lineno}: "
                    f".{node.attr} produces raw traceback text — banned in "
                    f"secure-logging dirs; use logger.exception() "
                    f"(diagnose-gated) with a scrubbed safe_msg"
                )
            elif node.attr == "exc_info" and self._expr_is_sys_module(
                node.value
            ):
                receiver = (
                    node.value.id if isinstance(node.value, ast.Name) else "sys"
                )
                self.errors.append(
                    f"{self.filename}:{node.lineno}: "
                    f"{receiver}.exc_info is banned in secure-logging dirs — "
                    f"tracebacks must never be interpolated into log "
                    f"messages; use logger.exception() (diagnose-gated) "
                    f"with a scrubbed safe_msg"
                )
        self.generic_visit(node)

    def _check_secure_dir_call(self, node: ast.Call) -> None:
        """Literal getattr/import_module dodges + chained message checks."""
        func = node.func
        # getattr(x, "opt"/"catch"/"logger"/private handles)
        if (
            isinstance(func, ast.Name)
            and func.id == "getattr"
            and len(node.args) >= 2
            and isinstance(node.args[1], ast.Constant)
            and isinstance(node.args[1].value, str)
        ):
            attr = node.args[1].value
            if attr in BANNED_LOGGER_ATTRS:
                if attr == "patch":
                    reason = (
                        'patchers can mutate record["exception"] and '
                        "reattach tracebacks ungated"
                    )
                else:
                    reason = "it reaches raw loguru handles"
                self.errors.append(
                    f"{self.filename}:{node.lineno}: "
                    f'getattr(..., "{attr}") is banned in secure-logging '
                    f"dirs — {reason}"
                )
            elif attr in BANNED_TRACEBACK_ATTRS:
                self.errors.append(
                    f"{self.filename}:{node.lineno}: "
                    f'getattr(..., "{attr}") produces raw traceback text — '
                    f"banned in secure-logging dirs"
                )
            elif attr == "exc_info" and self._expr_is_sys_module(node.args[0]):
                receiver = (
                    node.args[0].id
                    if isinstance(node.args[0], ast.Name)
                    else "sys"
                )
                self.errors.append(
                    f"{self.filename}:{node.lineno}: "
                    f'getattr({receiver}, "exc_info") is banned in secure-logging '
                    f"dirs — tracebacks must never be interpolated into "
                    f"log messages"
                )

        # importlib.import_module("loguru"/"traceback"), __import__(...)
        dynamic_import_module = (
            node.args[0].value
            if node.args and isinstance(node.args[0], ast.Constant)
            else None
        )
        if dynamic_import_module in BANNED_DYNAMIC_IMPORTS and (
            self._is_literal_dynamic_import(node, dynamic_import_module)
        ):
            self.errors.append(
                f"{self.filename}:{node.lineno}: "
                f'dynamic import of "{dynamic_import_module}" is banned in '
                f"secure-logging dirs — {WRAPPER_IMPORT_HINT}"
            )

        # logger.bind(...) chains are wrapper-preserving, but the message is
        # production-visible — run the exception-variable check.
        if (
            isinstance(func, ast.Attribute)
            and func.attr
            in {"info", "warning", "error", "critical", "exception", "log"}
            and self._is_wrapper_chain(func.value)
        ):
            self._check_exception_var_in_log(node)

    def _is_wrapper_chain(self, expr: ast.AST) -> bool:
        """True for bind() call chains rooted at the name 'logger'."""
        if not isinstance(expr, ast.Call):
            return False
        while (
            isinstance(expr, ast.Call)
            and isinstance(expr.func, ast.Attribute)
            and expr.func.attr in WRAPPER_CHAIN_METHODS
        ):
            expr = expr.func.value
        return isinstance(expr, ast.Name) and expr.id == "logger"

    def visit_Call(self, node: ast.Call) -> None:
        """Check function calls for logging sensitive data."""
        # Check if it's a logger call
        if self._is_logger_call(node):
            self._check_sensitive_logging(node)
            self._check_exc_info_on_warning(node)
            self._check_exception_var_in_log(node)

        if self.in_secure_dir:
            self._check_secure_dir_call(node)

        self.generic_visit(node)

    def _is_logger_call(self, node: ast.Call) -> bool:
        """Check if the call is to a logger method."""
        if isinstance(node.func, ast.Attribute):
            # Check for logger.info, logger.debug, etc.
            attr_name = node.func.attr
            if attr_name in {
                "info",
                "debug",
                "warning",
                "error",
                "critical",
                "exception",
                "log",
            }:
                if isinstance(node.func.value, ast.Name):
                    return node.func.value.id == "logger"
                if isinstance(node.func.value, ast.Attribute):
                    # Handle self.logger, cls.logger, etc.
                    return node.func.value.attr == "logger"
        return False

    def _check_sensitive_logging(self, node: ast.Call) -> None:
        """Check if sensitive data is being logged."""
        for arg in node.args:
            self._check_expression_for_sensitive_data(arg, node.lineno)

        for keyword in node.keywords:
            self._check_expression_for_sensitive_data(
                keyword.value, node.lineno
            )

    def _get_log_level(self, node: ast.Call) -> Optional[str]:
        """Return the logging level name (e.g. 'warning', 'debug') or None."""
        if isinstance(node.func, ast.Attribute):
            return node.func.attr
        return None

    def _check_exc_info_on_warning(self, node: ast.Call) -> None:
        """Flag exc_info=True on warning/error/critical level logs.

        exc_info dumps full tracebacks which are visible in production.
        Use logger.exception() for dev-only traceback logging, or
        logger.debug(..., exc_info=True) which is off in production.
        """
        level = self._get_log_level(node)
        if level not in {"warning", "error", "critical"}:
            return
        for kw in node.keywords:
            if (
                kw.arg == "exc_info"
                and isinstance(kw.value, ast.Constant)
                and kw.value.value is True
            ):
                self.errors.append(
                    f"{self.filename}:{node.lineno}: "
                    f"exc_info=True on logger.{level}() exposes tracebacks in production — "
                    f"use logger.exception() or logger.debug(..., exc_info=True) instead"
                )

    def _check_exception_var_in_log(self, node: ast.Call) -> None:
        """Flag logging calls that interpolate the exception variable.

        Patterns like logger.warning(f"...{e}") or logger.warning("...%s", e)
        leak error details. Use logger.exception() instead, which is
        configured for dev-only output.
        """
        level = self._get_log_level(node)
        # logger.debug() is fine — not visible in production
        if level == "debug":
            return
        # Plain loguru logger.exception() output is dev-only — but in the
        # secure_logging dirs the wrapper makes the message production-visible
        # at ERROR, so the exception variable must not be interpolated there.
        if level == "exception" and not self.in_secure_dir:
            return

        # Collect current except variable names
        except_vars = {v for v in self._except_var_stack if v is not None}
        if not except_vars:
            return

        # Check positional args (f-strings and %-style args)
        for arg in node.args:
            if self._expr_references_vars(arg, except_vars):
                if level == "exception":
                    self.errors.append(
                        f"{self.filename}:{node.lineno}: "
                        f"Exception variable in logger.exception() — this message is "
                        f"production-visible at ERROR (secure_logging wrapper) — "
                        f"log a scrubbed message (scrub_error(); engines: self._scrub_error()) instead"
                    )
                else:
                    self.errors.append(
                        f"{self.filename}:{node.lineno}: "
                        f"Exception variable in logger.{level}() leaks error details — "
                        f"use logger.exception() or remove the variable"
                    )
                return

        # Check keyword args: loguru-style logger.exception("msg: {err}", err=e) bypasses
        # the positional check above since the exception variable is in a keyword value.
        for kw in node.keywords:
            if self._expr_references_vars(kw.value, except_vars):
                if level == "exception":
                    self.errors.append(
                        f"{self.filename}:{node.lineno}: "
                        f"Exception variable in logger.exception() keyword arg — this message is "
                        f"production-visible at ERROR (secure_logging wrapper) — "
                        f"log a scrubbed message (scrub_error(); engines: self._scrub_error()) instead"
                    )
                else:
                    self.errors.append(
                        f"{self.filename}:{node.lineno}: "
                        f"Exception variable in logger.{level}() keyword arg leaks error details — "
                        f"use logger.exception() or remove the variable"
                    )
                return

    def _expr_references_vars(self, expr: ast.AST, var_names: set) -> bool:
        """Check if an expression references any of the given variable names."""
        if isinstance(expr, ast.Name):
            return expr.id in var_names
        if isinstance(expr, ast.JoinedStr):
            # f-string: check each formatted value
            return any(
                isinstance(v, ast.FormattedValue)
                and self._expr_references_vars(v.value, var_names)
                for v in expr.values
            )
        if isinstance(expr, ast.Call):
            # str(e), repr(e), "...".format(x=e), str(e).upper(), etc. —
            # the exception can hide in the receiver (func) and in keywords,
            # not just positional args.
            return (
                self._expr_references_vars(expr.func, var_names)
                or any(
                    self._expr_references_vars(a, var_names) for a in expr.args
                )
                or any(
                    self._expr_references_vars(kw.value, var_names)
                    for kw in expr.keywords
                )
            )
        if isinstance(expr, ast.Attribute):
            # e.args, e.strerror, e.response.text, e.__cause__ carry the
            # same error detail as str(e). Allowed: the exception *class
            # name* — type(e).__name__ / e.__class__.__name__ — which is
            # the sanctioned pattern for scrubbed log lines.
            if expr.attr == "__name__" and self._is_exc_class_ref(
                expr.value, var_names
            ):
                return False
            return self._expr_references_vars(expr.value, var_names)
        if isinstance(expr, ast.Subscript):
            # e.args[0], e["detail"]
            return self._expr_references_vars(
                expr.value, var_names
            ) or self._expr_references_vars(expr.slice, var_names)
        if isinstance(expr, ast.BinOp):
            # "message: %s" % e
            return self._expr_references_vars(
                expr.right, var_names
            ) or self._expr_references_vars(expr.left, var_names)
        if isinstance(expr, ast.Tuple):
            # ("message %s %s", e, other)
            return any(
                self._expr_references_vars(el, var_names) for el in expr.elts
            )
        return False

    def _is_exc_class_ref(self, expr: ast.AST, var_names: set) -> bool:
        """True for type(e) or e.__class__ over an exception variable."""
        if (
            isinstance(expr, ast.Call)
            and isinstance(expr.func, ast.Name)
            and expr.func.id == "type"
            and len(expr.args) == 1
            and not expr.keywords
        ):
            return self._expr_references_vars(expr.args[0], var_names)
        if isinstance(expr, ast.Attribute) and expr.attr == "__class__":
            return self._expr_references_vars(expr.value, var_names)
        return False

    def _is_allowed_logging(self, variable_name: str) -> bool:
        """Check if this specific logging is allowed."""
        for allowed in ALLOWED_LOGGING:
            if (
                allowed["file_pattern"] in self.filename
                and allowed["variable"] == variable_name
            ):
                return True
        return False

    def _check_expression_for_sensitive_data(
        self, expr: ast.AST, lineno: int
    ) -> None:
        """Recursively check an expression for sensitive data."""
        if isinstance(expr, ast.Name):
            # Skip if it's a known false positive
            if expr.id in FALSE_POSITIVE_VARS:
                return

            # Skip if it's specifically allowed for this file
            if self._is_allowed_logging(expr.id):
                return

            # Direct variable logging
            if any(
                sensitive in expr.id.lower() for sensitive in SENSITIVE_VARS
            ):
                self.errors.append(
                    f"{self.filename}:{lineno}: Potential logging of sensitive variable '{expr.id}'"
                )
            # Check for sensitive dictionaries being logged directly
            if any(
                sensitive in expr.id.lower()
                for sensitive in SENSITIVE_DICT_KEYS
            ):
                self.errors.append(
                    f"{self.filename}:{lineno}: Potential logging of sensitive dictionary '{expr.id}'"
                )

        elif isinstance(expr, ast.Attribute):
            # Check for Flask request object attributes
            if isinstance(expr.value, ast.Name) and expr.value.id == "request":
                if expr.attr in {
                    "form",
                    "json",
                    "data",
                    "headers",
                    "authorization",
                    "cookies",
                }:
                    self.errors.append(
                        f"{self.filename}:{lineno}: Potential logging of sensitive Flask request.{expr.attr} which may contain passwords or tokens"
                    )
                    return

            # Check for Flask g object attributes
            if isinstance(expr.value, ast.Name) and expr.value.id == "g":
                if (
                    "password" in expr.attr.lower()
                    or "token" in expr.attr.lower()
                ):
                    self.errors.append(
                        f"{self.filename}:{lineno}: Potential logging of sensitive Flask g.{expr.attr}"
                    )
                    return

            # Check for os.environ
            if (
                isinstance(expr.value, ast.Name)
                and expr.value.id == "os"
                and expr.attr == "environ"
            ):
                self.errors.append(
                    f"{self.filename}:{lineno}: Potential logging of os.environ which contains environment variables"
                )
                return

            # Check for dict.keys(), dict.values(), dict.items()
            if expr.attr in {"keys", "values", "items"}:
                if isinstance(expr.value, ast.Name):
                    if any(
                        sensitive in expr.value.id.lower()
                        for sensitive in SENSITIVE_DICT_KEYS
                    ):
                        self.errors.append(
                            f"{self.filename}:{lineno}: Potential logging of sensitive dictionary keys/values from '{expr.value.id}.{expr.attr}()'"
                        )

            # Check for sensitive attributes like obj.password, obj.api_key, etc.
            if any(
                sensitive in expr.attr.lower() for sensitive in SENSITIVE_VARS
            ):
                # Skip false positives
                if expr.attr in FALSE_POSITIVE_VARS:
                    return
                self.errors.append(
                    f"{self.filename}:{lineno}: Potential logging of sensitive attribute '{expr.attr}'"
                )

        elif isinstance(expr, ast.Call):
            # Check for list(), str(), repr() of sensitive dicts
            if isinstance(expr.func, ast.Name):
                if expr.func.id in {"list", "str", "repr"}:
                    for arg in expr.args:
                        # Check for list(dict.keys()), list(dict.values()), etc.
                        if isinstance(arg, ast.Call) and isinstance(
                            arg.func, ast.Attribute
                        ):
                            if arg.func.attr in {"keys", "values", "items"}:
                                if isinstance(arg.func.value, ast.Name):
                                    if any(
                                        sensitive in arg.func.value.id.lower()
                                        for sensitive in SENSITIVE_DICT_KEYS
                                    ):
                                        self.errors.append(
                                            f"{self.filename}:{lineno}: Potential logging of sensitive dictionary keys/values via '{expr.func.id}({arg.func.value.id}.{arg.func.attr}())'"
                                        )
                        # Also check direct conversion of sensitive dicts
                        elif isinstance(arg, ast.Name):
                            if any(
                                sensitive in arg.id.lower()
                                for sensitive in SENSITIVE_DICT_KEYS
                            ):
                                self.errors.append(
                                    f"{self.filename}:{lineno}: Potential logging of sensitive dictionary via '{expr.func.id}({arg.id})'"
                                )

        elif isinstance(expr, ast.JoinedStr):
            # Check f-strings
            for value in expr.values:
                if isinstance(value, ast.FormattedValue):
                    self._check_expression_for_sensitive_data(
                        value.value, lineno
                    )

        elif isinstance(expr, ast.Dict):
            # Check dictionary literals for sensitive keys
            for key in expr.keys:
                if isinstance(key, ast.Constant) and isinstance(key.value, str):
                    if any(
                        sensitive in key.value.lower()
                        for sensitive in SENSITIVE_VARS
                    ):
                        self.errors.append(
                            f"{self.filename}:{lineno}: Potential logging of dictionary with sensitive key '{key.value}'"
                        )

        elif isinstance(expr, ast.BinOp):
            # Check string formatting
            self._check_expression_for_sensitive_data(expr.left, lineno)
            self._check_expression_for_sensitive_data(expr.right, lineno)


def check_file(filepath: Path) -> List[str]:
    """Check a single Python file for sensitive logging."""
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            content = f.read()

        tree = ast.parse(content, filename=str(filepath))
        checker = SensitiveLoggingChecker(str(filepath))
        checker.visit(tree)
        return checker.errors

    except SyntaxError as e:
        return [f"{filepath}:{e.lineno}: Syntax error - {e.msg}"]
    except Exception as e:
        return [f"{filepath}: Error checking file - {e}"]


def main():
    """Main entry point for the pre-commit hook."""
    files = sys.argv[1:]
    all_errors = []

    for filename in files:
        filepath = Path(filename)
        if filepath.suffix == ".py":
            errors = check_file(filepath)
            all_errors.extend(errors)

    if all_errors:
        print("Sensitive data logging detected:")
        for error in all_errors:
            print(f"  {error}")
        print("\nPlease ensure sensitive data is not logged directly.")
        print("Consider:")
        print(
            "  - In llm/providers/, embeddings/providers/, web_search_engines/:"
        )
        print(
            "    use scrub_error() from security.log_sanitizer (engines: "
            "self._scrub_error()) and log the safe_msg"
        )
        print(
            "  - Elsewhere: use logger.exception() (traceback is dev-only) or remove the variable"
        )
        print(
            "  - Using sanitized versions of sensitive data structures before logging"
        )
        return 1

    return 0


if __name__ == "__main__":
    sys.exit(main())