learningcircuit--local-deep-research/tests/security/test_password_redaction_invariant.py

"""AST invariant: no traceback-rendering log call with a password in scope.

Companion to ``test_password_leakage.py`` (issue #4182, PR #4530). That
file proves representative handlers redact correctly at runtime; this one
mechanically enforces the sweep's contract across ALL handlers, so a new
``logger.exception`` (or ``logger.debug(..., exc_info=True)``) added to a
credential-bearing function fails CI instead of silently re-opening the
leak.

Why this matters: loguru ``diagnose=True`` renders frame locals into the
exception block. Any function that holds the user's SQLCipher master
password in a local/parameter and renders a traceback can therefore
persist the plaintext password to log sinks. Unlike API keys, the master
password is unrecoverable (TRUST.md §5).

The required pattern at such sites is::

    except Exception as e:
        safe_msg = redact_secrets(str(e), password)
        logger.warning(f"... {safe_msg}")

Scope and limitations (deliberate, keep in sync with the sweep):

- Only the modules in ``SWEPT_MODULES`` are checked: the files swept by
  #4530 plus the password-bearing session helpers added in the #4182
  follow-up. Extend ``SWEPT_MODULES`` when a new module starts handling
  the master password.
- A "password name" is any identifier containing ``password`` (case-
  insensitive), excluding ``session_password_store`` (a module-level store
  object, not a credential value).
- The check is per innermost function: a traceback-rendering log call is a
  violation if any password name is bound anywhere in the same function.
  This is slightly conservative (the name might be bound on a disjoint
  branch) — that is fine; the fix is cheap and the false-negative
  direction is the dangerous one.

Known blind spots (none triggered by the swept files today; keep logging at
the top function level and rooted at the bare ``logger`` so they stay safe):

- Nested/inner functions: the check attributes a log call to its *innermost*
  enclosing function. A traceback log inside a nested function that closes
  over an outer ``password`` is NOT flagged, even though diagnose=True
  renders outer frames too. Don't put traceback logs in inner functions of
  password-bearing functions.
- Aliased loggers: only call chains rooted at the literal name ``logger`` are
  inspected. ``from loguru import logger as log`` (or stdlib ``logging``)
  would evade the check. The swept files all use the bare ``logger``.
"""

import ast
from pathlib import Path

import pytest

import local_deep_research.database.encrypted_db as encrypted_db_module
import local_deep_research.scheduler.background as background_module
import local_deep_research.web.queue.processor_v2 as processor_v2_module
import local_deep_research.database.thread_local_session as thread_local_session_module
import local_deep_research.library.download_management.status_tracker as status_tracker_module
import local_deep_research.database.thread_metrics as thread_metrics_module
import local_deep_research.metrics.search_tracker as search_tracker_module
import local_deep_research.metrics.token_counter as token_counter_module
import local_deep_research.database.library_init as library_init_module
import local_deep_research.database.backup.backup_executor as backup_executor_module
import local_deep_research.web_search_engines.rate_limiting.tracker as rate_limit_tracker_module
import local_deep_research.research_library.search.services.research_history_indexer as research_history_indexer_module

SWEPT_MODULES = [
    encrypted_db_module,
    background_module,
    processor_v2_module,
    # Consumers of ``DatabaseInitializationError`` that hold the master
    # password in scope while logging the catch (#4182 follow-up): the
    # raise site redacts, but these callers re-render via the logger.
    thread_local_session_module,
    status_tracker_module,
    # Holds the master password in scope while opening a per-thread
    # metrics session and logging the failure (#4182 follow-up).
    thread_metrics_module,
    # Credential-centric helpers that open an encrypted session with the
    # master password and log failures (#4182 targeted sweep). The big
    # multi-purpose route/service modules are NOT listed: they are
    # protected at the sink level (log_utils forces diagnose=False on the
    # persisted DB / frontend sinks) so their unrelated-error tracebacks
    # stay useful.
    search_tracker_module,
    token_counter_module,
    library_init_module,
    backup_executor_module,
    rate_limit_tracker_module,
    research_history_indexer_module,
]

# Names that match the password heuristic but are not credential values.
_ALLOWED_PASSWORD_NAMES = {"session_password_store"}


def _password_names_in(func_node: ast.AST) -> set:
    """Collect credential-looking identifiers bound or used in *func_node*."""
    names = set()
    for node in ast.walk(func_node):
        if isinstance(node, ast.Name) and "password" in node.id.lower():
            names.add(node.id)
        if isinstance(node, ast.arg) and "password" in node.arg.lower():
            names.add(node.arg)
    return names - _ALLOWED_PASSWORD_NAMES


def _logger_method_name(call: ast.Call):
    """If *call* is a method call ultimately rooted at the bare ``logger``
    name, return the final method name (e.g. ``"exception"``, ``"debug"``);
    otherwise ``None``.

    Follows chained loguru calls so ``logger.bind(...).debug(...)`` and
    ``logger.opt(...).exception(...)`` resolve to ``"debug"`` /
    ``"exception"`` — that ``.bind()`` form is used live in the swept files
    (scheduler/background.py), so missing it would leave a real blind spot.

    Aliased loggers (``from loguru import logger as log``) are NOT followed —
    the root must be the literal name ``logger``. None of the swept files
    alias it; see the limitation note in the module docstring.
    """
    func = call.func
    if not isinstance(func, ast.Attribute):
        return None
    method = func.attr
    node = func.value
    # Walk the receiver chain to its root, stepping through intermediate
    # method calls (logger.bind(...)/opt(...)) and attribute accesses.
    while True:
        if isinstance(node, ast.Name):
            return method if node.id == "logger" else None
        if isinstance(node, ast.Call):
            node = node.func
        elif isinstance(node, ast.Attribute):
            node = node.value
        else:
            return None


def _renders_traceback(call: ast.Call) -> bool:
    """True if *call* is a logger call that renders an exception traceback.

    Matches ``logger.exception(...)`` and any ``logger.<level>(...,
    exc_info=<anything>)`` (including ``logger.bind(...).<level>`` chains).
    A literal ``exc_info=False`` is treated as a violation too — it serves
    no purpose and invites a flip to ``True``.
    """
    method = _logger_method_name(call)
    if method is None:
        return False
    if method == "exception":
        return True
    return any(kw.arg == "exc_info" for kw in call.keywords)


def find_unredacted_traceback_logs(source: str, filename: str) -> list:
    """Return ``(lineno, function, password_names)`` violations in *source*."""
    tree = ast.parse(source, filename=filename)
    violations = []

    class Visitor(ast.NodeVisitor):
        def __init__(self):
            self.func_stack = []

        def _visit_func(self, node):
            self.func_stack.append(node)
            self.generic_visit(node)
            self.func_stack.pop()

        visit_FunctionDef = _visit_func
        visit_AsyncFunctionDef = _visit_func

        def visit_Call(self, node):
            if _renders_traceback(node) and self.func_stack:
                func = self.func_stack[-1]
                names = _password_names_in(func)
                if names:
                    violations.append((node.lineno, func.name, sorted(names)))
            self.generic_visit(node)

    Visitor().visit(tree)
    return violations


@pytest.mark.parametrize(
    "module",
    SWEPT_MODULES,
    ids=lambda m: m.__name__.rsplit(".", 1)[-1],
)
def test_no_traceback_log_with_password_in_scope(module):
    """Every traceback-rendering log call in the swept files must live in
    a function with no password-named variable in scope.
    """
    path = Path(module.__file__)
    source = path.read_text(encoding="utf-8")

    violations = find_unredacted_traceback_logs(source, str(path))

    assert not violations, (
        f"{path.name} has traceback-rendering log calls in functions that "
        f"hold the SQLCipher master password — loguru diagnose=True would "
        f"render it via frame locals. Replace each with "
        f"`safe_msg = redact_secrets(str(e), password)` + "
        f"`logger.warning(...)` (see test_password_leakage.py and PR "
        f"#4530):\n"
        + "\n".join(
            f"  line {lineno}: {func}() — password names in scope: {names}"
            for lineno, func, names in violations
        )
    )


class TestCheckerSelfTest:
    """Prove the checker actually detects violations — without this, a
    refactor that silently breaks the AST walk would make the invariant
    test pass vacuously.
    """

    def test_flags_logger_exception_with_password_param(self):
        src = (
            "def handler(username, password):\n"
            "    try:\n"
            "        open_db(username, password)\n"
            "    except Exception:\n"
            "        logger.exception('boom')\n"
        )
        violations = find_unredacted_traceback_logs(src, "<test>")
        assert violations == [(5, "handler", ["password"])]

    def test_flags_bind_chain_exc_info_with_password(self):
        # The logger.bind(...).debug(..., exc_info=True) form is used live
        # in scheduler/background.py — the checker must follow the chain.
        src = (
            "def handler(username, password):\n"
            "    try:\n"
            "        open_db(username, password)\n"
            "    except Exception:\n"
            "        logger.bind(audit=True).debug('boom', exc_info=True)\n"
        )
        violations = find_unredacted_traceback_logs(src, "<test>")
        assert violations == [(5, "handler", ["password"])]

    def test_flags_bind_chain_exception_with_password(self):
        src = (
            "def handler(username, password):\n"
            "    try:\n"
            "        open_db(username, password)\n"
            "    except Exception:\n"
            "        logger.bind(audit=True).exception('boom')\n"
        )
        violations = find_unredacted_traceback_logs(src, "<test>")
        assert violations == [(5, "handler", ["password"])]

    def test_ignores_aliased_logger(self):
        # Only chains rooted at the literal name ``logger`` are inspected;
        # an aliased logger is a documented blind spot.
        src = (
            "def handler(username, password):\n"
            "    try:\n"
            "        open_db(username, password)\n"
            "    except Exception:\n"
            "        log.exception('boom')\n"
        )
        assert find_unredacted_traceback_logs(src, "<test>") == []

    def test_flags_exc_info_with_password_local(self):
        src = (
            "def handler(username):\n"
            "    user_password = store.retrieve(username)\n"
            "    try:\n"
            "        open_db(username, user_password)\n"
            "    except Exception:\n"
            "        logger.debug('boom', exc_info=True)\n"
        )
        violations = find_unredacted_traceback_logs(src, "<test>")
        assert violations == [(6, "handler", ["user_password"])]

    def test_ignores_logger_exception_without_password(self):
        src = (
            "def handler(config):\n"
            "    try:\n"
            "        reload(config)\n"
            "    except Exception:\n"
            "        logger.exception('boom')\n"
        )
        assert find_unredacted_traceback_logs(src, "<test>") == []

    def test_ignores_redacted_warning_with_password(self):
        src = (
            "def handler(username, password):\n"
            "    try:\n"
            "        open_db(username, password)\n"
            "    except Exception as e:\n"
            "        safe_msg = redact_secrets(str(e), password)\n"
            "        logger.warning(f'boom: {safe_msg}')\n"
        )
        assert find_unredacted_traceback_logs(src, "<test>") == []

    def test_session_password_store_alone_is_not_a_credential(self):
        src = (
            "def handler(username):\n"
            "    session_password_store.touch(username)\n"
            "    try:\n"
            "        work()\n"
            "    except Exception:\n"
            "        logger.exception('boom')\n"
        )
        assert find_unredacted_traceback_logs(src, "<test>") == []