Skip to content
Merged
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
78 changes: 78 additions & 0 deletions tests/test_CheckManager.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import pytest
import ast
import importlib
import logging
import time
Expand Down Expand Up @@ -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
Expand Down
Loading