diff --git a/tests/test_CheckManager.py b/tests/test_CheckManager.py index fe4a7ee..82d6b11 100644 --- a/tests/test_CheckManager.py +++ b/tests/test_CheckManager.py @@ -1,4 +1,5 @@ import pytest +import ast import importlib import logging import time @@ -172,6 +173,83 @@ def test_total_checks(api_only, debug_function, expected_total): assert cm.total_checks == expected_total +def test_api_checks_only_use_approved_external_commands(): + with open(script.__file__, "r") as source_file: + module = ast.parse(source_file.read()) + + functions = { + node.name: node + for node in module.body + if isinstance(node, ast.FunctionDef) + } + approved_api_boundaries = {"icurl"} + forbidden_calls = {"Connection", "run_cmd", "os.system", "os.popen"} + forbidden_prefixes = ("subprocess.", "pexpect.") + + def get_call_name(node): + if isinstance(node, ast.Name): + return node.id + if isinstance(node, ast.Attribute): + parent = get_call_name(node.value) + return "{}.{}".format(parent, node.attr) if parent else node.attr + return "" + + def is_literal_icurl_command(node): + if not isinstance(node, (ast.List, ast.Tuple)) or not node.elts: + return False + executable = node.elts[0] + return isinstance(executable, ast.Str) and executable.s == "icurl" + + def is_literal_icurl_call(node, function): + if not get_call_name(node.func).startswith("subprocess.") or not node.args: + return False + command = node.args[0] + if is_literal_icurl_command(command): + return True + if not isinstance(command, ast.Name): + return False + return any( + isinstance(candidate, ast.Assign) + and any( + isinstance(target, ast.Name) and target.id == command.id + for target in candidate.targets + ) + and is_literal_icurl_command(candidate.value) + for candidate in ast.walk(function) + ) + + def find_forbidden_calls(function_name, visited=None): + if visited is None: + visited = set() + if function_name in visited or function_name in approved_api_boundaries: + return set() + visited.add(function_name) + + findings = set() + function = functions.get(function_name) + if function is None: + return findings + for node in ast.walk(function): + if not isinstance(node, ast.Call): + continue + call_name = get_call_name(node.func) + if ( + call_name in forbidden_calls + or call_name.startswith(forbidden_prefixes) + ) and not is_literal_icurl_call(node, function): + findings.add(call_name) + elif call_name in functions: + findings.update(find_forbidden_calls(call_name, visited)) + return findings + + violations = { + check.__name__: sorted(find_forbidden_calls(check.__name__)) + for check in CheckManager.api_checks + if find_forbidden_calls(check.__name__) + } + assert violations == {} + + def test_exception_in_initialize(): """Exception in initialize is not captured by CheckManager. The exception should go up to the script's main() and abort the script