diff --git a/src/skillspector/python_import_bounds.py b/src/skillspector/python_import_bounds.py index b5e1de2c7..b8c092e89 100644 --- a/src/skillspector/python_import_bounds.py +++ b/src/skillspector/python_import_bounds.py @@ -322,12 +322,15 @@ def transparent_decorators(declaration: ast.stmt) -> bool: selected = trimmed scope = ast.Module(body=selected, type_ignores=[]) scope_nodes = [] + loaded_names: set[str] = set() for node in ast.walk(scope): check_runtime() nodes_seen += 1 if nodes_seen > _MAX_IMPORT_NODES: return False scope_nodes.append(node) + if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Load): + loaded_names.add(node.id) if depth: try: scope_source = ast.unparse(scope) @@ -343,20 +346,30 @@ def transparent_decorators(declaration: ast.stmt) -> bool: for node in scope_nodes: check_runtime() if isinstance(node, ast.Import): - requests = [(alias.name, None, 0) for alias in node.names] + requests = [ + (alias.name, None, 0, alias.asname or alias.name.split(".")[0]) + for alias in node.names + ] elif isinstance(node, ast.ImportFrom): - requests = [(node.module or "", alias.name, node.level) for alias in node.names] + requests = [ + (node.module or "", alias.name, node.level, alias.asname or alias.name) + for alias in node.names + ] else: continue - for module, imported_symbol, level in requests: + for module, imported_symbol, level, binding in requests: matches = resolve(module, module_path, level) if matches is None or len(matches) > 1: return False + # Initialization can call or expose an imported callable + # through defaults, decorators or callback arguments. Only + # unused imports can omit their selected callable bodies. + initialization_only = symbol == _INITIALIZATION and binding not in loaded_names if matches and ( imported_symbol == "*" or not inspect( matches[0], - _INITIALIZATION if symbol == _INITIALIZATION else imported_symbol, + _INITIALIZATION if initialization_only else imported_symbol, depth + 1, ) ): diff --git a/tests/nodes/analyzers/test_python_bundled_shell_ownership.py b/tests/nodes/analyzers/test_python_bundled_shell_ownership.py index 405e0c344..8faf3ec6f 100644 --- a/tests/nodes/analyzers/test_python_bundled_shell_ownership.py +++ b/tests/nodes/analyzers/test_python_bundled_shell_ownership.py @@ -428,3 +428,120 @@ def test_local_initialization_call_cannot_install_an_unchecked_executor(initiali assert not PythonImportBounds(list(files), files).allows( "caller.py", files["caller.py"], lambda: None ) + + +def _imported_initialization_files(import_statement, initialization, *, executor): + engine = ( + "def install(value=None):\n" + " import helpers\n" + " import subprocess\n" + " def execute(command):\n" + ' subprocess.run(["bash", "-c", command])\n' + " helpers.runner = execute\n" + " return value\n" + if executor + else 'def install(value=None):\n print("initializing")\n return value\n' + ) + return { + "caller.py": ( + "from helpers import output\n" + 'tool = "echo rm"\n' + 'command = f"Report `{tool}`"\n' + 'command = command.removeprefix("Report ")\n' + 'command += " -rf /"\n' + "output(command)\n" + ), + "helpers.py": ( + "runner = print\nimport side_effect\ndef output(value):\n runner(value)\n" + ), + "side_effect.py": f"{import_statement}\n{initialization}\n", + "engine.py": engine, + } + + +@pytest.mark.parametrize("executor", [False, True], ids=["print-only", "shell-executor"]) +@pytest.mark.parametrize( + "import_statement,initialization", + [ + ("from engine import install", "install()"), + ("from engine import install as setup", "setup()"), + ("import engine", "engine.install()"), + ("import engine as setup", "setup.install()"), + ("from engine import install", "list(map(install, [None]))"), + ("import engine", "list(map(engine.install, [None]))"), + ("from engine import install", "def unused(value=install()):\n pass"), + ("from engine import install", "def unused(callback=install):\n pass"), + ("from engine import install", "@install\ndef unused():\n pass"), + ("from engine import install", "def unused() -> install():\n pass"), + ], + ids=[ + "from-import", + "from-alias", + "module-import", + "module-alias", + "callback", + "module-callback", + "default-call", + "default-callback", + "decorator", + "return-annotation", + ], +) +def test_imported_initialization_inspects_used_callable_bodies( + import_statement, initialization, executor +): + files = _imported_initialization_files(import_statement, initialization, executor=executor) + assert PythonImportBounds(list(files), files).allows( + "caller.py", files["caller.py"], lambda: None + ) is (not executor) + result = static_runner.run_static_patterns_with_ledger( + {"components": list(files), "file_cache": files}, [tm] + ) + event = next(event for event in result["inspection_ledger"] if event["path"] == "caller.py") + assert event["outcome"] is (LedgerOutcome.PARTIAL if executor else LedgerOutcome.COMPLETED) + + +@pytest.mark.parametrize("aliased", [False, True]) +@pytest.mark.parametrize("executor", [False, True]) +def test_dotted_module_initialization_tracks_the_bound_import_name(aliased, executor): + statement = "import pkg.engine as setup" if aliased else "import pkg.engine" + expression = "setup.install()" if aliased else "pkg.engine.install()" + files = _imported_initialization_files(statement, expression, executor=executor) + files["pkg/engine.py"] = files.pop("engine.py") + assert PythonImportBounds(list(files), files).allows( + "caller.py", files["caller.py"], lambda: None + ) is (not executor) + + +@pytest.mark.parametrize("import_statement", ["from engine import install", "import engine"]) +def test_unused_imported_callable_body_is_not_initialization(import_statement): + files = _imported_initialization_files(import_statement, "", executor=True) + assert PythonImportBounds(list(files), files).allows( + "caller.py", files["caller.py"], lambda: None + ) + + +def test_imported_setup_does_not_select_unrelated_function_bodies(): + files = _imported_initialization_files( + "from engine import install", "install()", executor=False + ) + files["engine.py"] += ( + 'def unrelated(value):\n import subprocess\n subprocess.run(["bash", "-c", value])\n' + ) + assert PythonImportBounds(list(files), files).allows( + "caller.py", files["caller.py"], lambda: None + ) + + +@pytest.mark.parametrize( + "budget,limit", + [("_MAX_IMPORT_FILES", 3), ("_MAX_IMPORT_DEPTH", 2), ("_MAX_IMPORT_NODES", 1)], +) +def test_imported_initialization_retains_bounded_fallback(monkeypatch, budget, limit): + files = _imported_initialization_files( + "from engine import install", "install()", executor=False + ) + monkeypatch.setattr(python_import_bounds, budget, limit) + assert not PythonImportBounds(list(files), files).allows( + "caller.py", files["caller.py"], lambda: None + )