diff --git a/tests/test_editor_photoshop_features_more4.py b/tests/test_editor_photoshop_features_more4.py index f3dcc578c..365f281ad 100644 --- a/tests/test_editor_photoshop_features_more4.py +++ b/tests/test_editor_photoshop_features_more4.py @@ -93,7 +93,7 @@ def test_custom_filter_reuses_convolve_kernel(): assert "convolve(readSource(snap), k, divisor, bias, 1)" in EFFECTS -def test_new_filter_aliases_registered(): +def test_blur_and_custom_filter_aliases_registered(): assert "'box-blur': 'blur-box'" in EDITOR assert "'surface-blur': 'blur-surface'" in EDITOR assert "custom: 'other-custom'" in EDITOR diff --git a/tests/test_procfs_scan_guard.py b/tests/test_procfs_scan_guard.py index 9d3f91ef1..5e9d3b9c0 100644 --- a/tests/test_procfs_scan_guard.py +++ b/tests/test_procfs_scan_guard.py @@ -12,6 +12,8 @@ exercising every procfs-touching code path on both a Linux and a non-Linux host. ``test_cookbook_stop_without_procfs.py`` covers the behaviour itself. """ import ast + +import pytest from pathlib import Path REPO_ROOT = Path(__file__).resolve().parent.parent @@ -75,37 +77,112 @@ def _scan_sites(tree: ast.AST) -> list[ast.Call]: return sites -def _guard_lines(tree: ast.AST) -> list[int]: - """Line numbers of calls that test whether procfs is present.""" - lines = [] - for node in ast.walk(tree): - if not isinstance(node, ast.Call): - continue - func = node.func - name = func.attr if isinstance(func, ast.Attribute) else getattr(func, "id", "") - if name not in _GUARD_FUNCS: - continue - subject_is_procfs = ( - name in {"has_procfs", "is_wsl"} - or (isinstance(func, ast.Attribute) and _is_procfs_root(func.value)) - or any(_is_procfs_root(arg) for arg in node.args) - ) - if subject_is_procfs: - lines.append(node.lineno) - return lines +def _condition_requires_procfs(node: ast.AST, truth: bool) -> bool: + """Whether this Boolean outcome proves a direct procfs check succeeded. + + An ignored call, another function's check, or a check on only one possible + OR branch provides no proof. Keep unknown expressions conservative. + """ + if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.Not): + return _condition_requires_procfs(node.operand, not truth) + if isinstance(node, ast.BoolOp): + facts = [_condition_requires_procfs(value, truth) for value in node.values] + all_values_required = isinstance(node.op, ast.And) == truth + return any(facts) if all_values_required else all(facts) + if not isinstance(node, ast.Call): + return False + func = node.func + name = func.attr if isinstance(func, ast.Attribute) else getattr(func, "id", "") + subject_is_procfs = ( + name == "has_procfs" + or (isinstance(func, ast.Attribute) and _is_procfs_root(func.value)) + or any(_is_procfs_root(arg) for arg in node.args) + ) + return truth and name in _GUARD_FUNCS and subject_is_procfs -def _enclosing_scope(tree: ast.AST, node: ast.AST) -> ast.AST: - """Smallest function/module scope containing ``node``.""" - best = tree - for candidate in ast.walk(tree): - if not isinstance(candidate, (ast.FunctionDef, ast.AsyncFunctionDef)): - continue - end = getattr(candidate, "end_lineno", None) or candidate.lineno - if candidate.lineno <= node.lineno <= end: - if best is tree or candidate.lineno > best.lineno: - best = candidate - return best +def _guarded_scan_ids(tree: ast.AST) -> set[int]: + """Follow conditional/early-exit paths instead of comparing line numbers. + + Each function starts with no procfs assumption. Facts from a loop or try + body cannot escape to code which may run without that body completing. + This intentionally requires direct checks; arbitrary aliases/call graphs + are not a proof that procfs exists at the enumeration site. + """ + scans = {id(site) for site in _scan_sites(tree)} + guarded = set() + + def expression(node, present): + if node is None: + return + if id(node) in scans and present: + guarded.add(id(node)) + if isinstance(node, ast.BoolOp): + for value in node.values: + expression(value, present) + # Later AND operands run only after a true prefix; later OR + # operands run only after a false prefix. + present = present or _condition_requires_procfs(value, isinstance(node.op, ast.And)) + elif isinstance(node, ast.IfExp): + expression(node.test, present) + expression(node.body, present or _condition_requires_procfs(node.test, True)) + expression(node.orelse, present or _condition_requires_procfs(node.test, False)) + elif isinstance(node, (ast.ListComp, ast.SetComp, ast.GeneratorExp, ast.DictComp)): + for generator in node.generators: + expression(generator.iter, present) + for condition in generator.ifs: + expression(condition, present) + present = present or _condition_requires_procfs(condition, True) + for attr in ("elt", "key", "value"): + expression(getattr(node, attr, None), present) + elif isinstance(node, ast.Lambda): + expression(node.body, False) + else: + for child in ast.iter_child_nodes(node): + expression(child, present) + + def block(statements, present): + for statement in statements: + if isinstance(statement, (ast.FunctionDef, ast.AsyncFunctionDef)): + block(statement.body, False) + for value in statement.decorator_list + statement.args.defaults: + expression(value, present) + elif isinstance(statement, ast.ClassDef): + block(statement.body, present) + elif isinstance(statement, ast.If): + expression(statement.test, present) + yes = block(statement.body, present or _condition_requires_procfs(statement.test, True)) + no = block(statement.orelse, present or _condition_requires_procfs(statement.test, False)) + continuing = [fact for fact in (yes, no) if fact is not None] + if not continuing: + return None + present = all(continuing) + elif isinstance(statement, (ast.Return, ast.Raise, ast.Break, ast.Continue)): + for child in ast.iter_child_nodes(statement): + expression(child, present) + return None + elif isinstance(statement, (ast.For, ast.AsyncFor, ast.While)): + condition = statement.test if isinstance(statement, ast.While) else statement.iter + expression(condition, present) + body_present = present or (isinstance(statement, ast.While) and _condition_requires_procfs(condition, True)) + block(statement.body, body_present) + block(statement.orelse, present) + elif isinstance(statement, (ast.Try, getattr(ast, "TryStar", ast.Try))): + block(statement.body, present) + block(statement.orelse, present) + block(statement.finalbody, present) + for handler in statement.handlers: + block(handler.body, present) + elif isinstance(statement, (ast.With, ast.AsyncWith)): + for item in statement.items: + expression(item.context_expr, present) + block(statement.body, present) + else: + expression(statement, present) + return present + + block(tree.body, False) + return guarded def _app_python_files() -> list[Path]: @@ -139,12 +216,10 @@ def _collect_sites() -> tuple[list[str], list[str]]: sites = _scan_sites(tree) if not sites: continue - guards = _guard_lines(tree) + guards = _guarded_scan_ids(tree) for site in sites: - scope = _enclosing_scope(tree, site) - start = getattr(scope, "lineno", 0) label = f"{path.relative_to(REPO_ROOT)}:{site.lineno}" - if any(start <= g < site.lineno for g in guards): + if id(site) in guards: guarded.append(label) else: unguarded.append(label) @@ -154,7 +229,7 @@ def _collect_sites() -> tuple[list[str], list[str]]: def test_every_procfs_scan_is_guarded_by_an_existence_check(): _guarded, unguarded = _collect_sites() assert not unguarded, ( - "procfs pid scans with no existence check in the enclosing function — " + "procfs pid scans without an effective conditional existence guard — " "these raise FileNotFoundError on macOS and Windows: " + ", ".join(unguarded) ) @@ -176,3 +251,49 @@ def test_the_guard_detector_still_sees_the_known_scans(): f"root was renamed. Found: {sorted(found)}" ) assert len(found) >= 3, f"expected at least 3 procfs scans, found {sorted(found)}" + + +@pytest.mark.parametrize("body", [ + 'Path("/proc").exists(); os.listdir("/proc")', + 'if Path("/proc").exists():\n pass\nos.listdir("/proc")', + 'if not Path("/proc").exists():\n os.listdir("/proc")', + 'if ready or Path("/proc").exists():\n os.listdir("/proc")', + 'if ready:\n Path("/proc").exists()\nos.listdir("/proc")', + 'os.listdir("/proc"); Path("/proc").exists()', + 'if Path("/tmp").exists():\n os.listdir("/proc")', + 'def other():\n Path("/proc").exists()\nos.listdir("/proc")', + 'if not has_procfs():\n pass\nos.listdir("/proc")', + 'for x in []:\n if not has_procfs():\n return\nos.listdir("/proc")', + 'if has_procfs():\n return\nelse:\n os.listdir("/proc")', + '[pid for pid in os.listdir("/proc") if has_procfs()]', +], ids=["ignored", "expired-branch", "inverted", "unsafe-or", "sibling-branch", + "too-late", "wrong-path", "other-function", "no-early-exit", "loop-may-not-run", + "wrong-else", "late-comprehension-filter"]) +def test_guard_detector_rejects_ineffective_checks(body): + tree = ast.parse("def scan():\n" + "\n".join(" " + line for line in body.splitlines())) + sites = _scan_sites(tree) + assert sites, "negative control must contain a recognized enumeration" + assert any(id(site) not in _guarded_scan_ids(tree) for site in sites) + + +@pytest.mark.parametrize("body", [ + 'if Path("/proc").exists():\n os.listdir("/proc")', + 'if not Path("/proc").exists():\n return\nos.listdir("/proc")', + 'if not has_procfs():\n raise RuntimeError()\nos.listdir(PROC_ROOT)', + 'if has_procfs():\n pass\nelse:\n return\nPROC_ROOT.iterdir()', + 'if ready and has_procfs():\n os.scandir("/proc")', + 'if not has_procfs() or disabled:\n return\nPROC_ROOT.glob("*")', + 'if sys.platform == "linux" and has_procfs():\n PROC_ROOT.iterdir()', + 'for item in items:\n if not has_procfs():\n continue\n os.listdir("/proc")', + 'if os.path.isdir("/proc"):\n os.listdir("/proc")', + 'if has_procfs() and list(PROC_ROOT.iterdir()):\n pass', + 'if not has_procfs() or list(PROC_ROOT.iterdir()):\n pass', + '[pid for item in items if has_procfs() for pid in os.listdir("/proc")]', +], ids=["positive-branch", "early-return", "early-raise", "else-exits", "and", + "negative-or", "platform-and-procfs", "loop-continue", "isdir", + "short-circuit-and", "short-circuit-or", "guarded-comprehension"]) +def test_guard_detector_preserves_effective_checks(body): + tree = ast.parse("def scan():\n" + "\n".join(" " + line for line in body.splitlines())) + sites = _scan_sites(tree) + assert sites, "positive control must contain a recognized enumeration" + assert all(id(site) in _guarded_scan_ids(tree) for site in sites)