From fd4e69fec3e43ecacd7c2d3d6cbdb660dfe596e4 Mon Sep 17 00:00:00 2001 From: Hodaya Berger Date: Mon, 13 Jul 2026 12:16:24 +0300 Subject: [PATCH] feat(PDRIVE-526, PDRIVE-687): add run_oc_command argument validation and SafeCmdString pre-commit check Add _validate_args_safe() to OcApiUtils that validates all run_oc_command arguments against an allowlist of safe patterns (flags, resource names, JSONPath expressions, field selectors). Raises ValueError on invalid input. - Add mypy-based pre-commit hook to check SafeCmdString type usage (PDRIVE-687) - Add unit tests for _validate_args_safe() covering allowed patterns and edge cases - Add dynamic test that discovers all rule classes, executes their methods, and validates the actual args passed to run_oc_command against the allowlist - Enforce _validate_args_safe() in test mock (OperatorTestBase.run_oc_command) so future rule tests catch invalid args at test time Assisted-by: Claude Code (Claude Opus 4.6) --- .pre-commit-config.yaml | 7 + pyproject.toml | 1 + src/in_cluster_checks/utils/oc_api_utils.py | 87 ++++++ tests/linters/check_safecmdstring_mypy.py | 61 ++++ tests/pytest_tools/test_operator_base.py | 6 + tests/utils/test_oc_api_utils_validation.py | 261 ++++++++++++++++ .../test_run_oc_command_args_validation.py | 282 ++++++++++++++++++ 7 files changed, 705 insertions(+) create mode 100755 tests/linters/check_safecmdstring_mypy.py create mode 100644 tests/utils/test_oc_api_utils_validation.py create mode 100644 tests/utils/test_run_oc_command_args_validation.py diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 81ab172..a1118f5 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -58,6 +58,13 @@ repos: types: [python] files: ^src/in_cluster_checks/ + - id: safecmdstring-mypy-check + name: check SafeCmdString types with mypy + entry: python tests/linters/check_safecmdstring_mypy.py + language: system + types: [python] + files: ^src/in_cluster_checks/ + - id: pytest-coverage name: pytest with coverage entry: python -m pytest diff --git a/pyproject.toml b/pyproject.toml index 6bfb215..16b0e25 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -34,6 +34,7 @@ dev = [ "black>=24.1.1", "flake8>=7.0.0", "isort>=5.13.2", + "mypy>=1.0.0", ] [project.scripts] diff --git a/src/in_cluster_checks/utils/oc_api_utils.py b/src/in_cluster_checks/utils/oc_api_utils.py index a3df6db..8c2aca0 100644 --- a/src/in_cluster_checks/utils/oc_api_utils.py +++ b/src/in_cluster_checks/utils/oc_api_utils.py @@ -22,6 +22,7 @@ def collect_data(self): import json import logging +import re from datetime import datetime, timezone from typing import Any, Dict, Optional @@ -516,6 +517,86 @@ def run_rsh_cmd(self, namespace: str, pod: str, command: SafeCmdString, timeout: self.logger.error(error_msg) return 1, "", error_msg + def _validate_args_safe(self, args: list) -> bool: + """ + Validate oc command arguments against allowlist patterns. + + Prevents argument injection where cluster-derived values could inject flags. + + Allowed patterns: + - Exact: -o, -n, --no-headers, --all-namespaces, -A + - Prefixes: jsonpath=, --since=, --tail=, --field-selector= + - Resources: alphanumeric start, can contain ., /, -, :, _ + """ + exact_allowed = {"-o", "-n", "--no-headers", "--all-namespaces", "-A"} + since_pattern = re.compile(r"^--since=\d+[smhd]$") + tail_pattern = re.compile(r"^--tail=\d+$") + field_selector_pattern = re.compile(r"^--field-selector=[\w./=_-]+$") + resource_pattern = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9.\-/:_]*$") + + for arg in args: + if arg in exact_allowed: + continue + + if arg.startswith("jsonpath="): + if not self._validate_jsonpath_structure(arg): + return False + continue + + if ( + since_pattern.match(arg) + or tail_pattern.match(arg) + or field_selector_pattern.match(arg) + or resource_pattern.match(arg) + ): + continue + + return False + + return True + + def _validate_jsonpath_structure(self, arg: str) -> bool: + """ + Validate JSONPath follows kubectl JSONPath grammar. + + Grammar naturally prevents injection - valid syntax never produces space-hyphen. + """ + jsonpath_value = arg.split("jsonpath=", 1)[1] + + if not jsonpath_value: + return False + + # Field: word chars, hyphens, with optional escaped dots (\.) or namespace (/) + # Examples: "name", "my-field", "k8s\\.v1\\.cni\\.cncf\\.io", "k8s\\.io/network-status" + field = r"[\w-]+(?:\\.[\w-]+)*(?:/[\w-]+(?:\\.[\w-]+)*)?" + + # Dot-prefixed field + dot_field = rf"\.{field}" + + # Bracket accessors: [*], [0], [0:5], [?(@.field=='value')] + brackets = r"\[(?:\*|\d+(?::\d+)?|\?\([^)]+\))\]" + + # Path segment: .field optionally followed by brackets + # Allows: .items[*] or .metadata or .field[0][1] + path_segment = rf"{dot_field}(?:{brackets})*" + + # Full path: one or more segments + # Allows: .items[*].metadata.name + full_path = rf"(?:{path_segment})+" + + # Block content: keyword+path, path, keyword, or empty + # Allows: {range .items[*]}, {.field}, {end}, {} + content = rf"(?:range\s+{full_path}|if\s+{full_path}|{full_path}|end)?" + + # Single block: {content} + block = rf"\{{{content}\}}" + + # Full pattern: one or more blocks optionally separated by || + # Allows: {...}{...}, {...}||{...}, or mixed like {range ...}{.name}||{.ns}||{end} + pattern = rf"^{block}(?:(?:\|\|)?{block})*$" + + return re.match(pattern, jsonpath_value) is not None + def run_oc_command(self, command: str, args: list, timeout: int = 120, raise_on_error: bool = True) -> tuple: """ Run oc command using openshift_client library. @@ -532,6 +613,12 @@ def run_oc_command(self, command: str, args: list, timeout: int = 120, raise_on_ Raises: UnExpectedSystemOutput: If command fails and raise_on_error is True """ + if not self._validate_args_safe(args): + raise ValueError( + f"Unsafe arguments detected in oc command. " + f"Command: oc {command}, Args: {args}. " + f"Only safe arguments are allowed (see _validate_args_safe() docstring for allowed patterns)." + ) cmd_str = f"oc {command} {' '.join(args)}" self.operator._add_cmd_to_log(cmd_str) diff --git a/tests/linters/check_safecmdstring_mypy.py b/tests/linters/check_safecmdstring_mypy.py new file mode 100755 index 0000000..71733eb --- /dev/null +++ b/tests/linters/check_safecmdstring_mypy.py @@ -0,0 +1,61 @@ +#!/usr/bin/env python3 +""" +Mypy wrapper that only reports SafeCmdString type violations. + +This script runs mypy and filters output to only show errors related to SafeCmdString, +providing focused feedback on command injection prevention. +""" + +import subprocess +import sys +from pathlib import Path + + +def main(): + """Run mypy and filter for SafeCmdString violations only.""" + if len(sys.argv) < 2: + # No files provided - check all src files + files = ["src/in_cluster_checks/"] + else: + files = sys.argv[1:] + + # Run mypy with minimal configuration + cmd = [ + "mypy", + *files, + "--check-untyped-defs", # Check functions without type annotations + "--show-error-codes", # Show error codes like [arg-type] + "--no-error-summary", # Don't show summary + ] + + result = subprocess.run(cmd, capture_output=True, text=True) + + # Filter output to only SafeCmdString-related errors + safecmd_errors = [] + for line in result.stdout.splitlines(): + if "SafeCmdString" in line and "error:" in line: + safecmd_errors.append(line) + + if safecmd_errors: + print("\nFound SafeCmdString type violations:\n") + for error in safecmd_errors: + print(f" {error}") + print() + print("Methods requiring SafeCmdString:") + print(" - run_cmd(cmd, ...)") + print(" - get_output_from_run_cmd(cmd, ...)") + print(" - execute_cmd(cmd, ...)") + print(" - run_cmd_return_is_successful(cmd, ...)") + print(" - run_and_get_the_nth_field(cmd, ...)") + print(" - run_rsh_cmd(namespace, pod, command, ...)") + print() + print("Use: SafeCmdString('cmd {var}').format(var=value)") + print() + sys.exit(1) + else: + # All SafeCmdString checks passed + sys.exit(0) + + +if __name__ == "__main__": + main() diff --git a/tests/pytest_tools/test_operator_base.py b/tests/pytest_tools/test_operator_base.py index 86ab80d..9ac009c 100644 --- a/tests/pytest_tools/test_operator_base.py +++ b/tests/pytest_tools/test_operator_base.py @@ -278,6 +278,12 @@ def _run_oc_command_side_effects(self, command: str, args: list, timeout: int = UnExpectedSystemOutput: If raise_on_error=True and return_code != 0 """ _ = timeout # Acknowledge parameter + + assert self.operator_object.oc_api._validate_args_safe(list(args)), ( + f"Unsafe run_oc_command args detected: oc {command} {args}. " + f"All arguments must pass _validate_args_safe() validation." + ) + key = (command, tuple(args)) assert key in self.oc_cmd_to_output_dict, ( diff --git a/tests/utils/test_oc_api_utils_validation.py b/tests/utils/test_oc_api_utils_validation.py new file mode 100644 index 0000000..1a958f6 --- /dev/null +++ b/tests/utils/test_oc_api_utils_validation.py @@ -0,0 +1,261 @@ +"""Test OcApiUtils argument validation.""" + +import pytest + +from in_cluster_checks.utils.oc_api_utils import OcApiUtils + + +class MockOperator: + """Mock operator for testing.""" + + def _add_cmd_to_log(self, cmd): + pass + + def get_host_ip(self): + return "localhost" + + +class TestValidateArgsSafe: + """Test _validate_args_safe() argument validation.""" + + def setup_method(self): + """Set up test fixtures.""" + self.oc_api = OcApiUtils(MockOperator()) + + def test_exact_match_flags(self): + """Test exact match flags are allowed.""" + assert self.oc_api._validate_args_safe(["-o"]) is True + assert self.oc_api._validate_args_safe(["-n"]) is True + assert self.oc_api._validate_args_safe(["-A"]) is True + assert self.oc_api._validate_args_safe(["--no-headers"]) is True + assert self.oc_api._validate_args_safe(["--all-namespaces"]) is True + + def test_prefix_patterns(self): + """Test prefix patterns with validated values are allowed.""" + # --since - Time durations + assert self.oc_api._validate_args_safe(["--since=1h"]) is True + assert self.oc_api._validate_args_safe(["--since=30m"]) is True + assert self.oc_api._validate_args_safe(["--since=2d"]) is True + assert self.oc_api._validate_args_safe(["--since=45s"]) is True + + # --tail - Numeric line counts + assert self.oc_api._validate_args_safe(["--tail=15"]) is True + assert self.oc_api._validate_args_safe(["--tail=100"]) is True + + # --field-selector - Field selectors + assert self.oc_api._validate_args_safe(["--field-selector=type=kubernetes.io/tls"]) is True + assert self.oc_api._validate_args_safe(["--field-selector=status.phase=Running"]) is True + + def test_jsonpath_valid(self): + """Test valid JSONPath expressions are allowed.""" + # Basic JSONPath + assert self.oc_api._validate_args_safe(["jsonpath={.spec.version}"]) is True + assert self.oc_api._validate_args_safe(["jsonpath={.metadata.name}"]) is True + + # JSONPath with arrays + assert self.oc_api._validate_args_safe(["jsonpath={.items[*].metadata.name}"]) is True + assert self.oc_api._validate_args_safe(["jsonpath={.items[0].spec.nodeName}"]) is True + + # JSONPath with nested paths + assert self.oc_api._validate_args_safe(["jsonpath={.items[*].spec.csi.volumeAttributes.subvolumeName}"]) is True + + # JSONPath with field names containing hyphens (legitimate use case) + assert self.oc_api._validate_args_safe(["jsonpath={.my-field-name}"]) is True + assert self.oc_api._validate_args_safe(["jsonpath={.metadata.annotations.k8s-io/app}"]) is True + + # JSONPath with range expressions (spaces are legitimate) + assert self.oc_api._validate_args_safe(["jsonpath={range .items[*]}"]) is True + assert self.oc_api._validate_args_safe(["jsonpath={.items[*].status.conditions[?(@.type=='Ready')].status}"]) is True + + # JSONPath with filters using parentheses + assert self.oc_api._validate_args_safe(["jsonpath={.items[?(@.metadata.name=='pod-1')]}"]) is True + + # Empty JSONPath (technically valid in kubectl) + assert self.oc_api._validate_args_safe(["jsonpath={}"]) is True + + def test_jsonpath_injection_attacks(self): + """Test that JSONPath injection attacks are blocked.""" + # Flag injection via space-hyphen pattern + assert self.oc_api._validate_args_safe(["jsonpath={.name} --kubeconfig=/evil"]) is False + assert self.oc_api._validate_args_safe(["jsonpath={.name} -o wide"]) is False + assert self.oc_api._validate_args_safe(["jsonpath={.name} --all-namespaces"]) is False + + # Command injection attempts + assert self.oc_api._validate_args_safe(["jsonpath=; rm -rf /"]) is False + assert self.oc_api._validate_args_safe(["jsonpath=$(malicious)"]) is False + assert self.oc_api._validate_args_safe(["jsonpath=`whoami`"]) is False + + # Shell operators (pipe, redirect, etc.) + assert self.oc_api._validate_args_safe(["jsonpath=| cat /etc/passwd"]) is False + assert self.oc_api._validate_args_safe(["jsonpath={.name}|ls"]) is False + + # Invalid characters (shell operators) + assert self.oc_api._validate_args_safe(["jsonpath={.name};"]) is False + assert self.oc_api._validate_args_safe(["jsonpath={.name}&"]) is False + assert self.oc_api._validate_args_safe(["jsonpath={.name}<"]) is False + assert self.oc_api._validate_args_safe(["jsonpath={.name}>"]) is False + # Note: backslash (\) is allowed - it's used for escaping in JSONPath + + def test_jsonpath_structural_validation(self): + """Test JSONPath structural validation (balanced brackets).""" + # Unbalanced brackets + assert self.oc_api._validate_args_safe(["jsonpath={{.name}"]) is False # Extra { + assert self.oc_api._validate_args_safe(["jsonpath={.name}}"]) is False # Extra } + assert self.oc_api._validate_args_safe(["jsonpath={.items[[*]}"]) is False # Extra [ + assert self.oc_api._validate_args_safe(["jsonpath={.items[*]]}"]) is False # Extra ] + assert self.oc_api._validate_args_safe(["jsonpath={(.name))"]) is False # Extra ( + assert self.oc_api._validate_args_safe(["jsonpath={((.name)}"]) is False # Extra ( + + # Closing before opening (order violation) + assert self.oc_api._validate_args_safe(["jsonpath=}.name{"]) is False # } before { + assert self.oc_api._validate_args_safe(["jsonpath=].items["]) is False # ] before [ + assert self.oc_api._validate_args_safe(["jsonpath=).name("]) is False # ) before ( + + # Mixed unbalanced + assert self.oc_api._validate_args_safe(["jsonpath={[.name}"]) is False # Missing ] + assert self.oc_api._validate_args_safe(["jsonpath={(.name]}"]) is False # ( with ] + + def test_jsonpath_empty(self): + """Test empty JSONPath is rejected.""" + assert self.oc_api._validate_args_safe(["jsonpath="]) is False # Empty value after prefix + + def test_resource_names(self): + """Test resource names are allowed.""" + # Simple resource names + assert self.oc_api._validate_args_safe(["pod"]) is True + assert self.oc_api._validate_args_safe(["deployment"]) is True + assert self.oc_api._validate_args_safe(["node"]) is True + + # Resource with dot notation + assert self.oc_api._validate_args_safe(["deployment.apps"]) is True + assert self.oc_api._validate_args_safe(["clusteroperators.config.openshift.io"]) is True + assert self.oc_api._validate_args_safe(["nodenetworkconfigurationpolicies.nmstate.io"]) is True + + # Resource with namespace/name + assert self.oc_api._validate_args_safe(["default/my-pod"]) is True + assert self.oc_api._validate_args_safe(["openshift-config/cluster"]) is True + + # Resource with hyphen + assert self.oc_api._validate_args_safe(["my-deployment"]) is True + assert self.oc_api._validate_args_safe(["cluster-name"]) is True + + # Resource with underscore + assert self.oc_api._validate_args_safe(["my_secret"]) is True + assert self.oc_api._validate_args_safe(["config_map_name"]) is True + + # Resource with colon (for namespaced resources) + assert self.oc_api._validate_args_safe(["network.operator/cluster"]) is True + + def test_combined_args(self): + """Test multiple arguments together (as in actual usage).""" + # From: oc get csr -o json + assert self.oc_api._validate_args_safe(["csr", "-o", "json"]) is True + + # From: oc get secret kube-root-ca.crt -n openshift-config -o json + assert self.oc_api._validate_args_safe(["secret", "kube-root-ca.crt", "-n", "openshift-config", "-o", "json"]) is True + + # From: oc get daemonsets --all-namespaces -o json + assert self.oc_api._validate_args_safe(["daemonsets", "--all-namespaces", "-o", "json"]) is True + + # From: oc get pod -A -o jsonpath={...} + assert ( + self.oc_api._validate_args_safe(["pod", "-A", "-o", "jsonpath={.items[*].metadata.name}"]) is True + ) + + # From: oc logs -n openshift-storage pod-name --since=1h --tail=15 + assert ( + self.oc_api._validate_args_safe(["-n", "openshift-storage", "pod-name", "--since=1h", "--tail=15"]) is True + ) + + # From: oc adm top nodes --no-headers + assert self.oc_api._validate_args_safe(["top", "nodes", "--no-headers"]) is True + + def test_unsafe_args(self): + """Test that unsafe arguments are rejected.""" + # Command injection attempts + assert self.oc_api._validate_args_safe(["; rm -rf /"]) is False + assert self.oc_api._validate_args_safe(["$(malicious)"]) is False + assert self.oc_api._validate_args_safe(["`whoami`"]) is False + assert self.oc_api._validate_args_safe(["| cat /etc/passwd"]) is False + + # Invalid characters in resource names + assert self.oc_api._validate_args_safe(["pod&"]) is False + assert self.oc_api._validate_args_safe(["pod|"]) is False + assert self.oc_api._validate_args_safe(["pod;"]) is False + + # Invalid prefix (no alphanumeric start for resource names) + assert self.oc_api._validate_args_safe([".hidden"]) is False + assert self.oc_api._validate_args_safe(["/absolute/path"]) is False + assert self.oc_api._validate_args_safe(["-unknown-flag"]) is False + + # Unsafe values in prefix patterns (command injection after prefix) + assert self.oc_api._validate_args_safe(["jsonpath=; rm -rf /"]) is False + assert self.oc_api._validate_args_safe(["--since=1h; malicious"]) is False + assert self.oc_api._validate_args_safe(["--tail=15 | cat /etc/passwd"]) is False + assert self.oc_api._validate_args_safe(["--field-selector=$(malicious)"]) is False + + # Invalid formats for prefix patterns + assert self.oc_api._validate_args_safe(["--since=invalid"]) is False # Not a valid duration + assert self.oc_api._validate_args_safe(["--tail=abc"]) is False # Not numeric + + def test_all_existing_usages(self): + """Test validation against all actual usages from the codebase.""" + # Security rules + assert self.oc_api._validate_args_safe(["csr", "-o", "json"]) is True + assert ( + self.oc_api._validate_args_safe(["secret", "kube-root-ca.crt", "-n", "openshift-config", "-o", "json"]) + is True + ) + assert ( + self.oc_api._validate_args_safe(["secret", "--field-selector=type=kubernetes.io/tls", "-A", "-o", "json"]) + is True + ) + + # Network rules + assert self.oc_api._validate_args_safe(["dns.operator.openshift.io/cluster", "-o", "json"]) is True + assert self.oc_api._validate_args_safe(["pod", "-A", "-o", "jsonpath={}"]) is True + assert ( + self.oc_api._validate_args_safe(["nodenetworkconfigurationpolicies.nmstate.io", "-A", "-o", "json"]) + is True + ) + + # Storage rules + assert ( + self.oc_api._validate_args_safe(["pv", "-o", "jsonpath={.items[*].spec.csi.volumeAttributes.subvolumeName}"]) + is True + ) + assert ( + self.oc_api._validate_args_safe(["-n", "openshift-storage", "pod-name", "--since=1h", "--tail=15"]) + is True + ) + + # K8s rules + assert self.oc_api._validate_args_safe(["top", "nodes", "--no-headers"]) is True + assert self.oc_api._validate_args_safe(["daemonsets", "--all-namespaces", "-o", "json"]) is True + assert self.oc_api._validate_args_safe(["clusteroperators.config.openshift.io", "--no-headers"]) is True + assert ( + self.oc_api._validate_args_safe( + ["policies.policy.open-cluster-management.io", "--all-namespaces", "-o", "json"] + ) + is True + ) + assert ( + self.oc_api._validate_args_safe(["config.imageregistry.operator.openshift.io", "cluster", "-o", "json"]) + is True + ) + assert self.oc_api._validate_args_safe(["clusteroperators", "-o", "json"]) is True + assert self.oc_api._validate_args_safe(["console.operator.openshift.io", "cluster", "-o", "json"]) is True + assert self.oc_api._validate_args_safe(["network.operator.openshift.io", "cluster", "-o", "json"]) is True + assert self.oc_api._validate_args_safe(["csv", "-n", "open-cluster-management", "-o", "json"]) is True + + # Resources utilization + assert self.oc_api._validate_args_safe(["node", "node-name"]) is True + + def test_field_selector_pattern(self): + """Test that --field-selector= prefix pattern works.""" + # This should pass because --field-selector= is in our prefix_allowed list + assert self.oc_api._validate_args_safe(["--field-selector=type=kubernetes.io/tls"]) is True + assert self.oc_api._validate_args_safe(["--field-selector=status.phase=Running"]) is True + + # Combined with other flags + assert self.oc_api._validate_args_safe(["secret", "-A", "-o", "json"]) is True diff --git a/tests/utils/test_run_oc_command_args_validation.py b/tests/utils/test_run_oc_command_args_validation.py new file mode 100644 index 0000000..8d59f58 --- /dev/null +++ b/tests/utils/test_run_oc_command_args_validation.py @@ -0,0 +1,282 @@ +""" +Test that all run_oc_command args in source code pass validation. + +This test discovers all OrchestratorRule and OrchestratorDataCollector subclasses, +instantiates each with a mock operator, and calls their methods. +The mock intercepts the ACTUAL resolved args (including f-strings) and validates +them against _validate_args_safe(). + +This catches injection attempts that static linters miss, because the code is +actually executed and f-strings are resolved to their real values. +""" + +import importlib +import inspect +import pkgutil +import types +import typing +from unittest.mock import MagicMock + +import pytest + +import in_cluster_checks.rules as rules_package +from in_cluster_checks.core.operations import DataCollector +from in_cluster_checks.core.rule import OrchestratorRule +from in_cluster_checks.utils.oc_api_utils import OcApiUtils + + +class MockOperator: + """Minimal mock operator for OcApiUtils instantiation.""" + + def _add_cmd_to_log(self, cmd): + pass + + def get_host_ip(self): + return "localhost" + + +class ApiObjectMock(str): + """str subclass that acts as a mock API object. + + Inherits __str__, __format__, __eq__, __hash__, __bool__ from str. + Only defines attribute access, calling, iteration, and dict-like methods. + """ + + def __new__(cls, name="mock-resource"): + return str.__new__(cls, name) + + def __getattr__(self, attr): + return ApiObjectMock(self) + + def __call__(self, *args, **kwargs): + return ApiObjectMock(self) + + def __getitem__(self, key): + return ApiObjectMock(self) + + def __iter__(self): + return iter([ApiObjectMock(self)]) + + def get(self, key, default=None): + return ApiObjectMock(self) + + def items(self): + return [("mock-key", ApiObjectMock(self))] + + def values(self): + return [ApiObjectMock(self)] + + def keys(self): + return ["mock-key"] + + +def _make_mock_api_object(name="mock-resource"): + """Create a mock that behaves like an openshift_client APIObject. + + Uses ApiObjectMock so any attribute accessed in f-strings automatically + produces a valid string. No explicit attribute setup needed — the mock + handles arbitrary nesting dynamically. + """ + return ApiObjectMock(name) + + +def _import_all_rule_modules(): + """Import all modules under in_cluster_checks.rules to discover subclasses.""" + package_path = rules_package.__path__ + for _, module_name, _ in pkgutil.walk_packages(package_path, prefix=rules_package.__name__ + "."): + try: + importlib.import_module(module_name) + except Exception: + continue + + +def _get_all_subclasses(base): + """Recursively collect all subclasses of a base class.""" + result = [] + for cls in base.__subclasses__(): + result.append(cls) + result.extend(_get_all_subclasses(cls)) + return result + + +def _class_uses_run_oc_command(cls): + """Check if class or any non-framework parent uses run_oc_command. + + Walks the MRO so child classes that inherit run_oc_command calls from + intermediate parents (e.g. WhereaboutsBaseRule) are discovered. + """ + for klass in cls.__mro__: + if klass in (object, OrchestratorRule, DataCollector): + continue + if klass.__module__.startswith("in_cluster_checks.core"): + continue + try: + source = inspect.getsource(klass) + except (TypeError, OSError): + continue + if "run_oc_command" in source: + return True + return False + + +def _find_classes_with_run_oc_command(): + """Find all OrchestratorRule and DataCollector subclasses that use run_oc_command.""" + _import_all_rule_modules() + + classes = [] + for base_class in (OrchestratorRule, DataCollector): + for cls in _get_all_subclasses(base_class): + if not hasattr(cls, "unique_name") or not hasattr(cls, "title"): + continue + if _class_uses_run_oc_command(cls): + classes.append(cls) + + return classes + + +def _create_mock_instance(cls): + """Create an instance of a rule/collector class with mock operator. + + Tries proper __init__ to preserve rule-specific instance attributes, + falls back to __new__ if __init__ fails with mocks. + """ + mock_executor = MagicMock() + mock_executor.node_name = "test-node" + mock_executor.ip = "192.168.1.10" + mock_executor.host_name = "test-node" + + try: + if issubclass(cls, OrchestratorRule): + instance = cls(host_executor=mock_executor, node_executors=None) + else: + instance = cls(host_executor=mock_executor) + except Exception: + instance = cls.__new__(cls) + instance.logger = MagicMock() + instance._host_executor = mock_executor + instance.node_name = "test-node" + + instance.oc_api = OcApiUtils(MockOperator()) + + return instance + + +def _return_type_contains(return_type, target): + """Check if a return type annotation contains a target type, handling Unions and generics.""" + if return_type is target: + return True + origin = typing.get_origin(return_type) + if origin is target: + return True + if origin is types.UnionType or origin is typing.Union: + return any(_return_type_contains(arg, target) for arg in typing.get_args(return_type)) + return False + + +def _mock_oc_api_methods(instance, mock_obj): + """Dynamically mock all public OcApiUtils methods based on return type annotations. + + Inspects OcApiUtils for all public methods, skips run_oc_command (which is + replaced with a capturing version by the caller), and picks a mock return + value from the method's return type annotation: + list -> [mock_obj] (so loop bodies execute) + dict -> {"items": []} (so .get("items", []) works) + tuple -> (0, "{}", "") (rc, stdout, stderr) + str -> "mock-value" + other -> mock_obj (ApiObjectMock, valid for f-strings) + """ + for name, method in inspect.getmembers(OcApiUtils, predicate=inspect.isfunction): + if name.startswith("_") or name == "run_oc_command": + continue + + try: + hints = typing.get_type_hints(method) + except Exception: + hints = {} + + return_type = hints.get("return") + if return_type is None: + mock_return = mock_obj + elif _return_type_contains(return_type, list): + mock_return = [mock_obj] + elif _return_type_contains(return_type, dict): + mock_return = {"items": []} + elif _return_type_contains(return_type, tuple): + mock_return = (0, "{}", "") + elif _return_type_contains(return_type, str): + mock_return = "mock-value" + else: + mock_return = mock_obj + + setattr(instance.oc_api, name, MagicMock(return_value=mock_return)) + + +def _collect_run_oc_command_args(instance): + """ + Call all methods on the instance and capture args passed to run_oc_command. + + Args are converted to strings via str() so that ApiObjectMock instances + resolve to their name, matching how f-string interpolation works in real code. + + Returns: + List of (command, args_list) tuples + """ + captured_calls = [] + + def capturing_run_oc_command(command, args, **kwargs): + captured_calls.append((command, [str(a) for a in args])) + return (0, "{}", "") + + mock_obj = _make_mock_api_object() + _mock_oc_api_methods(instance, mock_obj) + instance.oc_api.run_oc_command = capturing_run_oc_command + instance.run_data_collector = MagicMock(return_value=mock_obj) + + for name, method in inspect.getmembers(instance, predicate=inspect.ismethod): + if name == "__init__": + continue + + sig = inspect.signature(method) + params = [ + p for p in sig.parameters.values() + if p.name != "self" and p.default is inspect.Parameter.empty + ] + if not params: + try: + method() + except Exception: + pass + else: + mock_args = [_make_mock_api_object() for _ in params] + try: + method(*mock_args) + except Exception: + pass + + return captured_calls + + +_classes_with_run_oc = _find_classes_with_run_oc_command() + + +@pytest.mark.parametrize( + "rule_class", + _classes_with_run_oc, + ids=[cls.__name__ for cls in _classes_with_run_oc], +) +def test_run_oc_command_args_are_safe(rule_class): + """Validate that all run_oc_command args in a rule class pass _validate_args_safe().""" + instance = _create_mock_instance(rule_class) + captured_calls = _collect_run_oc_command_args(instance) + + oc_api = OcApiUtils(MockOperator()) + errors = [] + + for command, args in captured_calls: + if not oc_api._validate_args_safe(args): + errors.append(f"oc {command} {args}") + + assert not errors, ( + f"Unsafe run_oc_command args found in {rule_class.__name__}:\n" + + "\n".join(f" - {e}" for e in errors) + )