""" Tests for the check-sensitive-logging pre-commit hook. Ensures the hook detects logging of passwords, API keys, tokens, and other sensitive data that could leak user information to log files. """ import ast import sys from importlib import import_module from pathlib import Path HOOKS_DIR = Path(__file__).parent.parent.parent / ".pre-commit-hooks" sys.path.insert(0, str(HOOKS_DIR)) hook_module = import_module("check-sensitive-logging") SensitiveLoggingChecker = hook_module.SensitiveLoggingChecker def _check_code(code: str, filename: str = "src/module.py") -> list: """Parse code and run the sensitive logging checker.""" tree = ast.parse(code) checker = SensitiveLoggingChecker(filename) checker.visit(tree) return checker.errors class TestDetectsPasswordLogging: """Ensures passwords are never logged.""" def test_detects_password_in_fstring(self): code = 'logger.info(f"User login with password={password}")\n' errors = _check_code(code) assert len(errors) >= 1 assert any("password" in e.lower() for e in errors) def test_detects_user_password_variable(self): code = 'logger.info(f"DB access: {user_password}")\n' errors = _check_code(code) assert len(errors) >= 1 def test_detects_pwd_variable(self): code = 'logger.warning(f"Connection failed: pwd={pwd}")\n' errors = _check_code(code) assert len(errors) >= 1 class TestDetectsApiKeyLogging: """Ensures API keys and tokens are never logged.""" def test_detects_api_key(self): code = 'logger.info(f"Using API key: {api_key}")\n' errors = _check_code(code) assert len(errors) >= 1 def test_detects_access_token(self): code = 'logger.info(f"Auth with token: {access_token}")\n' errors = _check_code(code) assert len(errors) >= 1 def test_detects_secret_key(self): code = 'logger.debug(f"Secret: {secret_key}")\n' errors = _check_code(code) assert len(errors) >= 1 class TestDetectsSensitiveDictLogging: """Ensures sensitive dicts (kwargs, credentials) are not logged.""" def test_detects_kwargs_logging(self): code = 'logger.info(f"Calling with: {kwargs}")\n' errors = _check_code(code) assert len(errors) >= 1 def test_detects_credentials_logging(self): code = 'logger.warning(f"Auth data: {credentials}")\n' errors = _check_code(code) assert len(errors) >= 1 def test_detects_settings_snapshot_logging(self): code = 'logger.info(f"Settings: {settings_snapshot}")\n' errors = _check_code(code) assert len(errors) >= 1 def test_detects_connection_string_logging(self): code = 'logger.info(f"DB: {connection_string}")\n' errors = _check_code(code) assert len(errors) >= 1 class TestDetectsExcInfoOnWarning: """Ensures exc_info=True is not used on warning/error (leaks tracebacks).""" def test_detects_exc_info_on_warning(self): code = 'logger.warning("Failed", exc_info=True)\n' errors = _check_code(code) assert len(errors) >= 1 assert any("exc_info" in e for e in errors) def test_detects_exc_info_on_error(self): code = 'logger.error("Crash", exc_info=True)\n' errors = _check_code(code) assert len(errors) >= 1 def test_allows_exc_info_on_debug(self): code = 'logger.debug("Debug trace", exc_info=True)\n' errors = _check_code(code) assert len(errors) == 0 class TestDetectsExceptionVarInLog: """Ensures exception variables are not interpolated in non-debug logs.""" def test_detects_exception_in_warning(self): code = """ try: do_work() except Exception as e: logger.warning(f"Failed: {e}") """ errors = _check_code(code) assert len(errors) >= 1 assert any("exception variable" in e.lower() for e in errors) def test_allows_exception_in_debug(self): code = """ try: do_work() except Exception as e: logger.debug(f"Failed: {e}") """ errors = _check_code(code) # debug level is allowed exc_var_errors = [ e for e in errors if "exception variable" in e.lower() ] assert len(exc_var_errors) == 0 def test_allows_logger_exception(self): code = """ try: do_work() except Exception as e: logger.exception(f"Failed: {e}") """ errors = _check_code(code) exc_var_errors = [ e for e in errors if "exception variable" in e.lower() ] assert len(exc_var_errors) == 0 class TestAllowsSafePatterns: """Ensures safe logging patterns are not flagged.""" def test_allows_normal_string_logging(self): code = 'logger.info("Processing 10 results")\n' errors = _check_code(code) assert len(errors) == 0 def test_allows_non_sensitive_variables(self): code = 'logger.info(f"Found {count} journals with score {score}")\n' errors = _check_code(code) assert len(errors) == 0 def test_allows_prompt_tokens(self): """prompt_tokens is about LLM tokens, not auth tokens.""" code = 'logger.info(f"Used {prompt_tokens} tokens")\n' errors = _check_code(code) assert len(errors) == 0 def test_allows_max_tokens(self): code = 'logger.info(f"Max tokens: {max_tokens}")\n' errors = _check_code(code) assert len(errors) == 0 def test_allows_test_files(self): """Test files should be allowed to log sensitive data for debugging.""" code = 'logger.info(f"Testing with password={password}")\n' # Tests may have relaxed rules for specific vars # (depends on ALLOWED_LOGGING config) _check_code(code, filename="tests/test_auth.py") # --------------------------------------------------------------------------- # #4183 step 2: wrapper-import enforcement + bypass-route detection in the # secure-logging dirs (llm/providers/, embeddings/providers/, # web_search_engines/). # --------------------------------------------------------------------------- SECURE_LOGGING_DIRS = hook_module.SECURE_LOGGING_DIRS # Representative in-dir filenames (one per protected dir) and an unprotected # control path. IN_DIR = "src/local_deep_research/web_search_engines/engines/search_engine_x.py" IN_DIR_LLM = "src/local_deep_research/llm/providers/implementations/x.py" IN_DIR_EMB = "src/local_deep_research/embeddings/providers/implementations/x.py" OUT_DIR = "src/local_deep_research/web/routes/x.py" WRAPPER_IMPORT = "from ...security.secure_logging import logger\n" class TestWrapperImportEnforcement: """Only the secure_logging wrapper may provide 'logger' in secure dirs.""" def test_raw_loguru_flagged_in_each_secure_dir(self): for filename in (IN_DIR, IN_DIR_LLM, IN_DIR_EMB): errors = _check_code("from loguru import logger\n", filename) assert any("raw loguru" in e for e in errors), filename def test_plain_and_aliased_loguru_import_flagged(self): assert _check_code("import loguru\n", IN_DIR) assert _check_code("import loguru as lg\n", IN_DIR) assert _check_code("from loguru import logger as log\n", IN_DIR) def test_raw_loguru_not_flagged_outside_secure_dirs(self): assert _check_code("from loguru import logger\n", OUT_DIR) == [] assert _check_code("import loguru\n", OUT_DIR) == [] def test_wrapper_import_allowed(self): assert _check_code(WRAPPER_IMPORT, IN_DIR) == [] assert ( _check_code( "from local_deep_research.security.secure_logging import logger\n", IN_DIR, ) == [] ) def test_wrapper_import_aliased_away_from_logger_flagged(self): errors = _check_code( "from ...security.secure_logging import logger as log\n", IN_DIR ) assert any("aliasing" in e for e in errors) def test_non_wrapper_logger_imports_flagged(self): for code in ( "from local_deep_research import logger\n", "from ...utilities.log_utils import logger\n", "from helper import logger\n", ): assert _check_code(code, IN_DIR), code def test_suffix_lookalike_module_is_not_the_wrapper(self): errors = _check_code( "from ...notsecurity.secure_logging import logger\n", IN_DIR ) assert any("other than" in e for e in errors) def test_secure_logging_internals_flagged(self): for code in ( "from ...security.secure_logging import _loguru_logger\n", "from ...security.secure_logging import SecureLogger\n", ): assert _check_code(code, IN_DIR), code def test_secure_logging_module_import_flagged(self): """v4 headline bypass: module import reaches _loguru_logger.""" for code in ( "from ...security import secure_logging\n", "from local_deep_research.security import secure_logging\n", "import local_deep_research.security.secure_logging\n", ): errors = _check_code(code, IN_DIR) assert any("secure_logging module" in e for e in errors), code def test_log_utils_imports_flagged_in_all_forms(self): """v5 headline bypass: log_utils re-exports the raw loguru logger.""" for code in ( "from ...utilities.log_utils import logger\n", "from ...utilities.log_utils import config_logger\n", "from ...utilities import log_utils\n", "from local_deep_research.utilities.log_utils import config_logger\n", "from local_deep_research.utilities import log_utils\n", "import local_deep_research.utilities.log_utils as lu\n", ): assert _check_code(code, IN_DIR), code assert ( _check_code("from ...utilities import log_utils\n", OUT_DIR) == [] ) def test_absolute_self_package_import_flagged(self): for code in ( "import local_deep_research\n", "import local_deep_research.web\n", ): errors = _check_code(code, IN_DIR) assert any("root package" in e for e in errors), code assert _check_code("import local_deep_research\n", OUT_DIR) == [] def test_traceback_imports_flagged(self): for code in ( "import traceback\n", "from traceback import format_exc\n", ): errors = _check_code(code, IN_DIR) assert any("traceback" in e for e in errors), code assert _check_code("import traceback\n", OUT_DIR) == [] def test_function_local_import_flagged(self): code = "def f():\n from loguru import logger\n" assert _check_code(code, IN_DIR) def test_dynamic_loguru_import_flagged(self): code = "import importlib\nx = importlib.import_module('loguru')\n" errors = _check_code(code, IN_DIR) assert any("dynamic import" in e for e in errors) def test_dynamic_import_of_other_modules_allowed(self): # The real auto_discovery.py pattern: f-string first arg, not "loguru" code = ( "import importlib\n" "m = importlib.import_module(\n" " f'.implementations.{module_name}', package='pkg'\n" ")\n" ) assert _check_code(code, IN_DIR) == [] class TestLoggerRebindingBans: """The local name 'logger' is reserved for the wrapper import.""" def test_assignment_rebinding_flagged(self): for code in ( "logger = get_raw()\n", "logger, x = a(), None\n", "[logger, x] = pair\n", "logger: object = get_raw()\n", "(logger := get_raw())\n", "logger += 1\n", ): assert _check_code(code, IN_DIR), code def test_control_flow_shadowing_flagged(self): for code in ( "for logger in xs:\n pass\n", "with open('f') as logger:\n pass\n", "try:\n f()\nexcept Exception as logger:\n pass\n", ): assert _check_code(code, IN_DIR), code def test_parameters_named_logger_flagged(self): for code in ( "def f(logger):\n pass\n", "async def f(logger):\n pass\n", "g = lambda logger: 1\n", ): assert _check_code(code, IN_DIR), code def test_import_binding_logger_flagged(self): for code in ( "import foo as logger\n", "from foo import bar as logger\n", ): assert _check_code(code, IN_DIR), code def test_deriving_new_name_from_logger_flagged(self): for code in ( "log = logger\n", "log = logger.bind(x=1)\n", "log = pkg.logger.bind(x=1)\n", "log = pkg.logger.patch(f).bind(x=1)\n", "x = log_utils.logger\n", "x = ldr.logger\n", ): assert _check_code(code, IN_DIR), code def test_ordinary_assignments_allowed(self): for code in ( "results = engine.run(q)\n", "log = get_audit_trail()\n", "safe_msg = scrub(e)\n", ): assert _check_code(code, IN_DIR) == [], code class TestBypassRouteDetection: """.opt/.catch/.patch/.logger/private handles/sys.exc_info are banned.""" def test_opt_flagged_in_all_forms(self): for code in ( "logger.opt(exception=True).error('x')\n", "logger.opt(**kwargs).error('x')\n", "raw = logger.opt(lazy=True)\n", "logger.bind(x=1).opt(depth=1)\n", ): errors = _check_code(code, IN_DIR) assert any(".opt" in e for e in errors), code def test_opt_not_flagged_outside_secure_dirs(self): assert ( _check_code("logger.opt(exception=True).error('x')\n", OUT_DIR) == [] ) def test_catch_flagged_in_all_forms(self): for code in ( "@logger.catch\ndef f():\n pass\n", "with logger.catch():\n pass\n", "h = logger.catch\n", ): errors = _check_code(code, IN_DIR) assert any(".catch" in e for e in errors), code def test_catch_decorator_with_args_reports_exactly_once(self): errors = _check_code( "@logger.catch(reraise=True)\ndef f():\n pass\n", IN_DIR ) assert len(errors) == 1 def test_patch_flagged_in_all_forms(self): for code in ( "logger.patch(lambda record: None).error('x')\n", "logger.bind(x=1).patch(lambda record: None).error('x')\n", "raw = logger.patch(lambda record: None)\n", ): errors = _check_code(code, IN_DIR) assert any(".patch" in e for e in errors), code def test_getattr_dodges_flagged(self): for attr in ( "opt", "catch", "patch", "logger", "_logger", "_loguru_logger", "__getattr__", ): code = f"getattr(obj, '{attr}')\n" errors = _check_code(code, IN_DIR) assert any("getattr" in e for e in errors), attr def test_logger_attribute_access_flagged(self): for code in ( "something.logger.exception('x')\n", "x = pkg.logger\n", "f(log_utils.logger)\n", ): errors = _check_code(code, IN_DIR) assert any(".logger" in e for e in errors), code def test_logger_attribute_assignment_reports_once(self): errors = _check_code("x = pkg.logger\n", IN_DIR) assert len(errors) == 1 assert ".logger" in errors[0] def test_private_handle_access_flagged(self): for code in ( "logger._logger\n", "secure_logging._loguru_logger\n", "logger.__getattr__('exception')\n", ): assert _check_code(code, IN_DIR), code def test_sys_exc_info_flagged(self): for code in ( "import sys\nx = sys.exc_info()\n", "import sys as s\nx = s.exc_info()\n", "import sys.monitoring\nx = sys.exc_info()\n", "import sys\nx = getattr(sys, 'exc_info')\n", "import sys as s\nx = getattr(s, 'exc_info')\n", "from sys import exc_info\nx = exc_info()\n", "from sys import exc_info as get_exc_info\nx = get_exc_info()\n", "from sys import *\nx = exc_info()\n", # noqa: F403 "x = __import__('sys').exc_info()\n", "x = getattr(__import__('sys'), 'exc_info')()\n", "import importlib\nx = importlib.import_module('sys').exc_info()\n", "from importlib import import_module\nx = import_module('sys').exc_info()\n", ): errors = _check_code(code, IN_DIR) assert any("exc_info" in e for e in errors), code def test_sys_alias_does_not_leak_between_function_scopes(self): code = ( "def f():\n" " import sys as s\n" " return s.exc_info()\n" "\n" "def g(s):\n" " return s.exc_info()\n" ) errors = [e for e in _check_code(code, IN_DIR) if "exc_info" in e] assert len(errors) == 1 def test_sys_alias_shadowing_shapes_not_flagged(self): cases = [ ( "import sys as s\n" "def g():\n" " s = obj\n" " return s.exc_info()\n" ), ( "import sys as s\n" "def g():\n" " import something as s\n" " return s.exc_info()\n" ), ( "import sys as s\n" "def g():\n" " try:\n" " raise RuntimeError()\n" " except Exception as s:\n" " return s.exc_info()\n" ), ( "class C:\n" " import sys as s\n" " def m(self):\n" " return s.exc_info()\n" ), ] for code in cases: errors = [e for e in _check_code(code, IN_DIR) if "exc_info" in e] assert errors == [], code def test_non_logger_exception_call_allowed(self): # tenacity pattern from rate_limiting/llm/wrapper.py:34 code = "exc = retry_state.outcome.exception()\n" assert _check_code(code, IN_DIR) == [] def test_dunder_getattr_method_definition_allowed(self): # shape of rate_limiting/llm/wrapper.py's attribute pass-through: # bans target Attribute access / literal getattr, not FunctionDef code = ( "class W:\n" " def __getattr__(self, name):\n" " return object.__getattribute__(self, name)\n" ) assert _check_code(code, IN_DIR) == [] class TestWrapperChainMessageChecks: """bind() chains keep the wrapper but messages stay checked.""" def test_bind_chain_with_safe_message_allowed(self): assert _check_code("logger.bind(x=1).exception('safe')\n", IN_DIR) == [] def test_bind_chain_with_exception_var_flagged(self): for method, args in ( ("exception", "f'failed: {e}'"), ("warning", "f'failed: {e}'"), ("error", "f'failed: {e}'"), ("critical", "f'failed: {e}'"), ("info", "f'failed: {e}'"), ("log", "'ERROR', f'failed: {e}'"), ): code = ( "try:\n" " f()\n" "except Exception as e:\n" f" logger.bind(x=1).{method}({args})\n" ) errors = _check_code(code, IN_DIR) assert any("Exception variable" in e for e in errors), method def test_bind_chain_debug_with_exception_var_allowed(self): code = ( "try:\n" " f()\n" "except Exception as e:\n" " logger.bind(x=1).debug(f'failed: {e}')\n" ) assert _check_code(code, IN_DIR) == [] def test_bind_chain_exc_info_true_not_flagged(self): # loguru stores exc_info=True in extra, not record["exception"] — # matches the pre-existing search_engine_elasticsearch.py site. code = "logger.bind(x=1).warning('failed', exc_info=True)\n" assert _check_code(code, IN_DIR) == [] class TestExceptionAttributeInterpolation: """Access *through* the exception variable leaks like str(e). Adversarial-review finding C1: e.args, e.response.text, e.strerror, e.__cause__, e.args[0] and receiver/keyword shapes evaded _expr_references_vars, which only recursed Name/JoinedStr/Call-args/ BinOp/Tuple. """ def test_attribute_access_flagged(self): for expr in ( "f'failed: {e.args}'", "f'disk: {e.strerror}'", "f'API {e.response.status_code}: {e.response.text}'", "f'cause: {e.__cause__}'", ): code = ( "try:\n" " f()\n" "except Exception as e:\n" f" logger.error({expr})\n" ) errors = _check_code(code, IN_DIR) assert any("Exception variable" in e for e in errors), expr def test_subscript_access_flagged(self): for expr in ("f'{e.args[0]}'", "f'{e[\"detail\"]}'"): code = ( "try:\n" " f()\n" "except Exception as e:\n" f" logger.error({expr})\n" ) errors = _check_code(code, IN_DIR) assert any("Exception variable" in e for e in errors), expr def test_receiver_and_keyword_shapes_flagged(self): for expr in ( "f'{str(e).upper()}'", # exception hidden in call receiver "'{x}'.format(x=e)", # exception hidden in call keyword "'%s' % e.args", # %-format with attribute "str(e.args)", ): code = ( "try:\n" " f()\n" "except Exception as e:\n" f" logger.error({expr})\n" ) errors = _check_code(code, IN_DIR) assert any("Exception variable" in e for e in errors), expr def test_percent_style_attribute_arg_flagged(self): code = ( "try:\n" " f()\n" "except Exception as e:\n" " logger.error('failed %s', e.args)\n" ) errors = _check_code(code, IN_DIR) assert any("Exception variable" in e for e in errors) def test_exception_level_attribute_flagged_in_secure_dir(self): code = ( "try:\n" " f()\n" "except Exception as e:\n" " logger.exception(f'boom: {e.args}')\n" ) errors = _check_code(code, IN_DIR) assert any("production-visible" in e for e in errors) def test_exception_class_name_allowed(self): # The sanctioned step-1 pattern: the class name carries no error # detail. Both spellings, incl. combined with a scrubbed message — # the exact shape used by search_engine_wayback.py et al. for expr in ( "f'{type(e).__name__}'", "f'{e.__class__.__name__}'", "f'failed ({type(e).__name__}): {safe_msg}'", ): code = ( "try:\n" " f()\n" "except Exception as e:\n" " safe_msg = sanitize_error_message(str(e))\n" f" logger.error({expr})\n" ) assert _check_code(code, IN_DIR) == [], expr def test_bare_type_call_still_flagged(self): # type(e) without .__name__ stringifies via repr — keep flagging it code = ( "try:\n" " f()\n" "except Exception as e:\n" " logger.error(f'{type(e)}')\n" ) errors = _check_code(code, IN_DIR) assert any("Exception variable" in e for e in errors) def test_debug_level_attribute_allowed(self): code = ( "try:\n" " f()\n" "except Exception as e:\n" " logger.debug(f'{e.args}')\n" ) assert _check_code(code, IN_DIR) == [] class TestTracebackDynamicRoutes: """Adversarial-review finding M1: dynamic imports of traceback and traceback-formatting attributes must be banned like the static forms.""" def test_dynamic_import_of_traceback_flagged(self): for code in ( "tb = importlib.import_module('traceback')\n", "tb = import_module('traceback')\n", "tb = __import__('traceback')\n", "raw = __import__('loguru')\n", ): errors = _check_code(code, IN_DIR) assert any("dynamic import" in e for e in errors), code def test_dynamic_import_of_other_modules_allowed(self): # auto_discovery.py's f-string import_module and constant imports of # ordinary modules must stay clean for code in ( "m = importlib.import_module(f'.implementations.{name}', pkg)\n", "m = importlib.import_module('json')\n", "m = __import__('os')\n", ): assert _check_code(code, IN_DIR) == [], code def test_format_exc_attribute_flagged(self): for code in ( "logger.error('boom: ' + tb.format_exc())\n", "x = tb.format_exception(e)\n", "x = tb.format_exception_only(e)\n", "x = tb.format_tb(e.__traceback__)\n", "tb.print_exc()\n", ): errors = _check_code(code, IN_DIR) assert any("traceback text" in e for e in errors), code def test_format_exc_getattr_flagged(self): errors = _check_code("f = getattr(tb, 'format_exc')\n", IN_DIR) assert any("traceback text" in e for e in errors) def test_traceback_routes_not_flagged_outside_secure_dirs(self): for code in ( "tb = importlib.import_module('traceback')\n", "x = tb.format_exc()\n", ): assert _check_code(code, OUT_DIR) == [], code class TestRecommendedPattern: def test_wrapper_import_plus_scrubbed_safe_msg_clean(self): code = ( "from ...security.secure_logging import logger\n" "\n" "\n" "def search(self, q):\n" " try:\n" " return self._run(q)\n" " except Exception as e:\n" " safe_msg = self._scrub_error(e)\n" " logger.exception(f'Search failed: {safe_msg}')\n" " return []\n" ) assert _check_code(code, IN_DIR) == [] class TestSourceTreeInvariant: """The real secure dirs must satisfy every ban, even past --no-verify.""" @staticmethod def _src_root() -> Path: return Path(__file__).parent.parent.parent / "src" @classmethod def _secure_dir_files(cls): repo_root = cls._src_root().parent for d in SECURE_LOGGING_DIRS: for path in sorted((repo_root / d).rglob("*.py")): yield path def test_checker_clean_on_real_secure_dirs(self): """Run the full checker over every file in the three dirs.""" failures = [] for path in self._secure_dir_files(): tree = ast.parse(path.read_text(encoding="utf-8")) checker = SensitiveLoggingChecker(str(path)) checker.visit(tree) failures.extend(checker.errors) assert not failures, "\n".join(failures) def test_no_raw_or_reexported_logger_imports_in_tree(self): """Independent import scan: headline v4/v5 bypasses asserted explicitly, without relying on the checker's own rules.""" failures = [] for path in self._secure_dir_files(): tree = ast.parse(path.read_text(encoding="utf-8")) for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: root = alias.name.split(".")[0] leaf = alias.name.split(".")[-1] if ( root in ("loguru", "traceback") or leaf in ("secure_logging", "log_utils") or root == "local_deep_research" ): failures.append( f"{path}:{node.lineno}: import {alias.name}" ) elif isinstance(node, ast.ImportFrom): module = node.module or "" is_wrapper = ( node.level > 0 and module == "security.secure_logging" ) or ( node.level == 0 and module == "local_deep_research.security.secure_logging" ) if module.split(".")[0] in ("loguru", "traceback"): failures.append( f"{path}:{node.lineno}: from {module} import ..." ) continue if module.split(".")[-1] == "log_utils": failures.append( f"{path}:{node.lineno}: from {module} import ..." ) continue for alias in node.names: bound = alias.asname or alias.name if bound == "logger" and not ( is_wrapper and alias.name == "logger" ): failures.append( f"{path}:{node.lineno}: non-wrapper logger " f"import from {module!r}" ) if alias.name in ("secure_logging", "log_utils") and ( not is_wrapper ): failures.append( f"{path}:{node.lineno}: module import of " f"{alias.name} from {module!r}" ) if is_wrapper and alias.name != "logger": failures.append( f"{path}:{node.lineno}: secure_logging " f"internal {alias.name!r} imported" ) assert not failures, "\n".join(failures) def test_no_banned_attribute_or_getattr_access_in_tree(self): """Attribute/literal-getattr shapes only — a FunctionDef named __getattr__ (rate_limiting/llm/wrapper.py) must not trip this.""" banned_attrs = { "opt", "catch", "patch", "logger", "_logger", "_loguru_logger", "__getattr__", } def is_literal_dynamic_sys_import(expr): if not isinstance(expr, ast.Call) or not expr.args: return False func = expr.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(expr.args[0], ast.Constant) and expr.args[0].value == "sys" ) def is_sys_module_expr(expr, sys_names): return ( isinstance(expr, ast.Name) and expr.id in sys_names ) or is_literal_dynamic_sys_import(expr) failures = [] saw_dunder_getattr_def = False for path in self._secure_dir_files(): tree = ast.parse(path.read_text(encoding="utf-8")) sys_names = {"sys"} for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: if alias.name == "sys" or ( alias.name.split(".")[0] == "sys" and alias.asname is None ): sys_names.add(alias.asname or "sys") for node in ast.walk(tree): if isinstance(node, ast.FunctionDef) and ( node.name == "__getattr__" ): saw_dunder_getattr_def = True if isinstance(node, ast.Attribute): if node.attr in banned_attrs: failures.append( f"{path}:{node.lineno}: .{node.attr} access" ) elif node.attr == "exc_info" and is_sys_module_expr( node.value, sys_names ): receiver = ( node.value.id if isinstance(node.value, ast.Name) else "sys" ) failures.append( f"{path}:{node.lineno}: {receiver}.exc_info access" ) elif ( isinstance(node, ast.ImportFrom) and node.level == 0 and node.module == "sys" and any( alias.name in ("*", "exc_info") for alias in node.names ) ): failures.append( f"{path}:{node.lineno}: from sys import exc_info" ) elif ( isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "getattr" and len(node.args) >= 2 and isinstance(node.args[1], ast.Constant) and isinstance(node.args[1].value, str) and ( node.args[1].value in banned_attrs or ( node.args[1].value == "exc_info" and is_sys_module_expr(node.args[0], sys_names) ) ) ): failures.append( f"{path}:{node.lineno}: literal getattr dodge" ) assert not failures, "\n".join(failures) # In-tree negative fixture: the attribute pass-through method exists # and did not trip the shape-based scan above. assert saw_dunder_getattr_def, ( "expected def __getattr__ in rate_limiting/llm/wrapper.py " "as the negative fixture for the shape-based scan" ) def test_no_logger_rebinding_or_derivation_in_tree(self): failures = [] for path in self._secure_dir_files(): tree = ast.parse(path.read_text(encoding="utf-8")) checker = SensitiveLoggingChecker(str(path)) checker.visit(tree) failures.extend( e for e in checker.errors if "rebinding" in e or "shadows" in e or "assigning a logger" in e ) assert not failures, "\n".join(failures) def test_every_secure_dir_file_using_logger_imports_wrapper(self): """Files calling logger.* must import it from the wrapper.""" failures = [] for path in self._secure_dir_files(): tree = ast.parse(path.read_text(encoding="utf-8")) uses_logger = any( isinstance(node, ast.Attribute) and isinstance(node.value, ast.Name) and node.value.id == "logger" for node in ast.walk(tree) ) if not uses_logger: continue has_wrapper_import = any( isinstance(node, ast.ImportFrom) and any( a.name == "logger" and a.asname is None for a in node.names ) and ( ( node.level > 0 and node.module == "security.secure_logging" ) or ( node.level == 0 and node.module == "local_deep_research.security.secure_logging" ) ) for node in ast.walk(tree) ) if not has_wrapper_import: failures.append(str(path)) assert not failures, ( "files using logger without the secure_logging wrapper import:\n" + "\n".join(failures) )