From dbd8d33217e7f821ebc4d956c3412e342e421613 Mon Sep 17 00:00:00 2001 From: yashrajbasav Date: Wed, 21 Oct 2026 14:06:48 +0530 Subject: [PATCH 1/2] fix(python): inspect imported initialization callables Signed-off-by: yashrajbasav --- src/skillspector/python_import_bounds.py | 21 ++- .../test_python_bundled_shell_ownership.py | 122 ++++++++++++++++++ 2 files changed, 139 insertions(+), 4 deletions(-) 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..130fbe3ef 100644 --- a/tests/nodes/analyzers/test_python_bundled_shell_ownership.py +++ b/tests/nodes/analyzers/test_python_bundled_shell_ownership.py @@ -428,3 +428,125 @@ 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\n" + "import side_effect\n" + "def 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 + ) From c04b9c37adca05b93f95e2737fa7afc9e1c60841 Mon Sep 17 00:00:00 2001 From: yashrajbasav Date: Wed, 21 Oct 2026 14:17:59 +0530 Subject: [PATCH 2/2] style(tests): format imported initialization fixtures Signed-off-by: yashrajbasav --- .../analyzers/test_python_bundled_shell_ownership.py | 9 ++------- 1 file changed, 2 insertions(+), 7 deletions(-) diff --git a/tests/nodes/analyzers/test_python_bundled_shell_ownership.py b/tests/nodes/analyzers/test_python_bundled_shell_ownership.py index 130fbe3ef..8faf3ec6f 100644 --- a/tests/nodes/analyzers/test_python_bundled_shell_ownership.py +++ b/tests/nodes/analyzers/test_python_bundled_shell_ownership.py @@ -452,10 +452,7 @@ def _imported_initialization_files(import_statement, initialization, *, executor "output(command)\n" ), "helpers.py": ( - "runner = print\n" - "import side_effect\n" - "def output(value):\n" - " runner(value)\n" + "runner = print\nimport side_effect\ndef output(value):\n runner(value)\n" ), "side_effect.py": f"{import_statement}\n{initialization}\n", "engine.py": engine, @@ -529,9 +526,7 @@ def test_imported_setup_does_not_select_unrelated_function_bodies(): "from engine import install", "install()", executor=False ) files["engine.py"] += ( - "def unrelated(value):\n" - " import subprocess\n" - ' subprocess.run(["bash", "-c", value])\n' + '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