#!/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())