From a6714a0ec3d7aa48b40c423de77e5fc8ba9559c5 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Fri, 14 Aug 2026 10:05:17 -0400 Subject: [PATCH 1/3] Enforce API-only check classification --- aci-preupgrade-validation-script.py | 2 +- tests/test_CheckManager.py | 51 +++++++++++++++++++++++++++++ 2 files changed, 52 insertions(+), 1 deletion(-) diff --git a/aci-preupgrade-validation-script.py b/aci-preupgrade-validation-script.py index f5664765..cf4dc7ff 100644 --- a/aci-preupgrade-validation-script.py +++ b/aci-preupgrade-validation-script.py @@ -7101,7 +7101,6 @@ class CheckManager: l3out_route_map_direction_check, l3out_route_map_missing_target_check, l3out_overlapping_loopback_check, - intersight_upgrade_status_check, isis_redis_metric_mpod_msite_check, bgp_golf_route_target_type_check, docker0_subnet_overlap_check, @@ -7171,6 +7170,7 @@ class CheckManager: cli_checks = [ # General apic_database_size_check, + intersight_upgrade_status_check, # Bugs apic_ca_cert_validation, diff --git a/tests/test_CheckManager.py b/tests/test_CheckManager.py index fe4a7eea..94266c76 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,56 @@ def test_total_checks(api_only, debug_function, expected_total): assert cm.total_checks == expected_total +def test_api_checks_do_not_run_ssh_or_cli_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 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): + 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 From f96e4bfbf82a1731bd20575405f444eaab7ffd07 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Fri, 14 Aug 2026 10:21:23 -0400 Subject: [PATCH 2/3] Allow direct icurl in API checks --- aci-preupgrade-validation-script.py | 11 ++++++----- tests/test_CheckManager.py | 16 ++++++++++++++-- 2 files changed, 20 insertions(+), 7 deletions(-) diff --git a/aci-preupgrade-validation-script.py b/aci-preupgrade-validation-script.py index cf4dc7ff..1c375a2c 100644 --- a/aci-preupgrade-validation-script.py +++ b/aci-preupgrade-validation-script.py @@ -3844,10 +3844,11 @@ def intersight_upgrade_status_check(**kwargs): recommended_action = 'Wait a few minutes for the upgrade to complete' doc_url = 'https://datacenter.github.io/ACI-Pre-Upgrade-Validation-Script/validations/#intersight-device-connector-upgrade-status' - cmd = ['icurl', '-gks', 'https://127.0.0.1/connector/UpgradeStatus'] - - log.info('cmd = ' + ' '.join(cmd)) - response = subprocess.check_output(cmd) + # The API-only container permits direct icurl access to this endpoint. + log.info('cmd = icurl -gks https://127.0.0.1/connector/UpgradeStatus') + response = subprocess.check_output([ + 'icurl', '-gks', 'https://127.0.0.1/connector/UpgradeStatus' + ]) try: resp_json = json.loads(response) @@ -7101,6 +7102,7 @@ class CheckManager: l3out_route_map_direction_check, l3out_route_map_missing_target_check, l3out_overlapping_loopback_check, + intersight_upgrade_status_check, isis_redis_metric_mpod_msite_check, bgp_golf_route_target_type_check, docker0_subnet_overlap_check, @@ -7170,7 +7172,6 @@ class CheckManager: cli_checks = [ # General apic_database_size_check, - intersight_upgrade_status_check, # Bugs apic_ca_cert_validation, diff --git a/tests/test_CheckManager.py b/tests/test_CheckManager.py index 94266c76..d9440da8 100644 --- a/tests/test_CheckManager.py +++ b/tests/test_CheckManager.py @@ -173,7 +173,7 @@ def test_total_checks(api_only, debug_function, expected_total): assert cm.total_checks == expected_total -def test_api_checks_do_not_run_ssh_or_cli_commands(): +def test_api_checks_only_use_approved_external_commands(): with open(script.__file__, "r") as source_file: module = ast.parse(source_file.read()) @@ -194,6 +194,15 @@ def get_call_name(node): return "{}.{}".format(parent, node.attr) if parent else node.attr return "" + def is_literal_icurl_call(node): + if not get_call_name(node.func).startswith("subprocess.") or not node.args: + return False + command = node.args[0] + if not isinstance(command, (ast.List, ast.Tuple)) or not command.elts: + return False + executable = command.elts[0] + return isinstance(executable, ast.Str) and executable.s == "icurl" + def find_forbidden_calls(function_name, visited=None): if visited is None: visited = set() @@ -209,7 +218,10 @@ def find_forbidden_calls(function_name, visited=None): 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): + if ( + call_name in forbidden_calls + or call_name.startswith(forbidden_prefixes) + ) and not is_literal_icurl_call(node): findings.add(call_name) elif call_name in functions: findings.update(find_forbidden_calls(call_name, visited)) From 8af80b7c5488ee05f1168195038be60cfa214dcf Mon Sep 17 00:00:00 2001 From: Gabriel Date: Fri, 14 Aug 2026 10:30:04 -0400 Subject: [PATCH 3/3] Preserve Intersight icurl command handling --- aci-preupgrade-validation-script.py | 9 ++++----- tests/test_CheckManager.py | 25 ++++++++++++++++++++----- 2 files changed, 24 insertions(+), 10 deletions(-) diff --git a/aci-preupgrade-validation-script.py b/aci-preupgrade-validation-script.py index 1c375a2c..f5664765 100644 --- a/aci-preupgrade-validation-script.py +++ b/aci-preupgrade-validation-script.py @@ -3844,11 +3844,10 @@ def intersight_upgrade_status_check(**kwargs): recommended_action = 'Wait a few minutes for the upgrade to complete' doc_url = 'https://datacenter.github.io/ACI-Pre-Upgrade-Validation-Script/validations/#intersight-device-connector-upgrade-status' - # The API-only container permits direct icurl access to this endpoint. - log.info('cmd = icurl -gks https://127.0.0.1/connector/UpgradeStatus') - response = subprocess.check_output([ - 'icurl', '-gks', 'https://127.0.0.1/connector/UpgradeStatus' - ]) + cmd = ['icurl', '-gks', 'https://127.0.0.1/connector/UpgradeStatus'] + + log.info('cmd = ' + ' '.join(cmd)) + response = subprocess.check_output(cmd) try: resp_json = json.loads(response) diff --git a/tests/test_CheckManager.py b/tests/test_CheckManager.py index d9440da8..82d6b116 100644 --- a/tests/test_CheckManager.py +++ b/tests/test_CheckManager.py @@ -194,14 +194,29 @@ def get_call_name(node): return "{}.{}".format(parent, node.attr) if parent else node.attr return "" - def is_literal_icurl_call(node): + 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 not isinstance(command, (ast.List, ast.Tuple)) or not command.elts: + if is_literal_icurl_command(command): + return True + if not isinstance(command, ast.Name): return False - executable = command.elts[0] - return isinstance(executable, ast.Str) and executable.s == "icurl" + 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: @@ -221,7 +236,7 @@ def find_forbidden_calls(function_name, visited=None): if ( call_name in forbidden_calls or call_name.startswith(forbidden_prefixes) - ) and not is_literal_icurl_call(node): + ) 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))