wehub-resource-sync
7a0da7932b
OSV-Scanner (Scheduled) / scan-scheduled (push) Failing after 0s
Create Release / test-gate (push) Has been cancelled
Create Release / release-gate (push) Has been cancelled
Create Release / ci-gate (push) Has been cancelled
Create Release / version-check (push) Has been cancelled
Create Release / e2e-test-gate (push) Has been cancelled
Create Release / responsive-test-gate (push) Has been cancelled
Create Release / compat-test-gate (push) Has been cancelled
Create Release / compose-integration-gate (push) Has been cancelled
Create Release / vulture-gate (push) Has been cancelled
Create Release / build (push) Has been cancelled
Create Release / provenance (push) Has been cancelled
Create Release / prerelease-docker (push) Has been cancelled
Create Release / publish-docker (push) Has been cancelled
Create Release / create-release (push) Has been cancelled
Create Release / cleanup-changelog (push) Has been cancelled
Create Release / trigger-pypi (push) Has been cancelled
Create Release / monitor-pypi (push) Has been cancelled
Create Release / Clean up orphan prerelease tags and signatures (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-form] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-metrics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-workflow] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-core] (push) Has been cancelled
CodeQL Advanced / Analyze (javascript-typescript) (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [history-news] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [library] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [link-analytics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-core] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-lifecycle] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [error-benchmark] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) (push) Has been cancelled
Docker Tests (Consolidated) / Accessibility Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Unit Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Example Tests (push) Has been cancelled
Docker Tests (Consolidated) / Production Image Smoke Test (push) Has been cancelled
Docker Tests (Consolidated) / Infrastructure Tests (push) Has been cancelled
OSSF Scorecard / OSSF Security Scorecard Analysis (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [mobile] (push) Has been cancelled
Backwards Compatibility / Verify Encryption Constants (push) Has been cancelled
Backwards Compatibility / PyPI Version Compatibility (push) Has been cancelled
Backwards Compatibility / Database Migration Tests (push) Has been cancelled
CodeQL Advanced / Analyze (python) (push) Has been cancelled
Docker Tests (Consolidated) / detect-changes (push) Has been cancelled
Docker Tests (Consolidated) / Build Test Image (push) Has been cancelled
Docker Tests (Consolidated) / All Pytest Tests + Coverage (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [accessibility] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [api-crud] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-login] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-register] (push) Has been cancelled
995 lines
37 KiB
Python
995 lines
37 KiB
Python
"""
|
|
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)
|
|
)
|