Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 17 additions & 4 deletions src/skillspector/python_import_bounds.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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,
)
):
Expand Down
117 changes: 117 additions & 0 deletions tests/nodes/analyzers/test_python_bundled_shell_ownership.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
)