feat(PDRIVE-526, PDRIVE-687): add run_oc_command argument validation …

.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

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]

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)
 

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()

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, (