Move subprocess.run into a wrapper

.github/workflows/security.yml

@@ -0,0 +1,54 @@
+name: Security Checks
+
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+    branches: [ main ]
+  schedule:
+    - cron: '0 2 * * 0' # weekly security scan on Sundays at 2 AM
+
+jobs:
+  security:
+    name: Security Analysis
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.12'
+
+      - name: Cache pip dependencies
+        uses: actions/cache@v3
+        with:
+          path: ~/.cache/pip
+          key: ${{ runner.os }}-pip-${{ hashFiles('**/pyproject.toml') }}
+          restore-keys: |
+            ${{ runner.os }}-pip-
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install bandit[toml]
+
+      - name: Make Bandit helper executable
+        run: chmod +x scripts/linters/run_bandit.sh
+
+      - name: Run Bandit security scan
+        run: ./scripts/linters/run_bandit.sh
+
+      - name: Upload Bandit results
+        uses: actions/upload-artifact@v4
+        if: always()
+        with:
+          name: bandit-security-report
+          path: bandit-report.json
+          retention-days: 30
+
+      - name: Check for high severity issues
+        if: failure()
+        run: echo "Bandit reported HIGH severity issues. See previous step output and artifact."

README.md

@@ -6,7 +6,7 @@
 [![CodeQL Analysis](https://github.com/mpowelson/arm-cli/actions/workflows/codeql-analysis.yml/badge.svg)](https://github.com/mpowelson/arm-cli/actions/workflows/codeql-analysis.yml)
 [![License](https://img.shields.io/badge/license-Apache%202.0-blue.svg)](https://github.com/mpowelson/arm-cli/blob/master/LICENSE)
 
-Experimental CLI for deploying robotic applications
+An experimental CLI for deploying robotic applications with opinionated defaults tailored for the ARM ecosystem.
 
 ## Installation
 

arm_cli/__init__.py

@@ -1,5 +1,10 @@
 """arm_cli package metadata."""
 
+from beartype.claw import beartype_this_package
+
+# Enable beartype on the package without polluting package __init__
+beartype_this_package()
+
 # Expose package version via setuptools-scm only
 try:
     from ._version import version as __version__  # type: ignore[attr-defined]

arm_cli/cli.py

@@ -1,8 +1,3 @@
-from beartype.claw import beartype_this_package
-
-# Enable beartype on the package without polluting package __init__
-beartype_this_package()
-
 import click
 
 from arm_cli import __version__

arm_cli/container/container.py

@@ -5,6 +5,7 @@
 import inquirer
 
 from arm_cli.settings import get_setting
+from arm_cli.utils.safe_subprocess import safe_run, sudo_run
 
 
 @click.group()
@@ -74,9 +75,7 @@ def attach_container(ctx):
     """
 
     try:
-        subprocess.run(
-            ["docker", "exec", "-it", selected_container_name, "bash", "-c", cmd], check=True
-        )
+        safe_run(["docker", "exec", "-it", selected_container_name, "bash", "-c", cmd], check=True)
     except subprocess.CalledProcessError as e:
         print(f"Error attaching to container: {e}")
     except KeyboardInterrupt:

arm_cli/projects/activate.py

@@ -60,14 +60,14 @@ def _activate(ctx, project: Optional[str] = None):
             # Extract project name (remove the active indicator if present)
             selected_choice = answers["project"]
             project = selected_choice.replace(" *", "")
+            if project is None:
+                raise RuntimeError("Project name cannot be None")
 
         except KeyboardInterrupt:
             print("\nCancelled.")
             return
 
     # Try to activate the project
-    # At this point, project is guaranteed to be a string
-    assert project is not None  # type guard
     project_config = activate_project(config, project)
 
     if project_config:

arm_cli/projects/remove.py

@@ -50,14 +50,14 @@ def _remove(ctx, project: Optional[str] = None):
             # Extract project name (remove the active indicator if present)
             selected_choice = answers["project"]
             project = selected_choice.replace(" *", "")
+            if project is None:
+                raise RuntimeError("Project name cannot be None")
 
         except KeyboardInterrupt:
             print("\nCancelled.")
             return
 
     # Try to remove the project
-    # At this point, project is guaranteed to be a string
-    assert project is not None  # type guard
 
     # Check if this is the active project
     available_projects = get_available_projects(config)

arm_cli/self/self.py

@@ -14,6 +14,7 @@
     save_config,
 )
 from arm_cli.settings import get_setting, load_settings, save_settings, set_setting
+from arm_cli.utils.safe_subprocess import safe_run, sudo_run
 
 
 @click.group()
@@ -49,11 +50,11 @@ def update(ctx, source, force):
 
         # Clear Python import cache
         print("Clearing Python caches...")
-        subprocess.run(["rm", "-rf", os.path.expanduser("~/.cache/pip")])
-        subprocess.run(["python", "-c", "import importlib; importlib.invalidate_caches()"])
+        safe_run(["rm", "-rf", os.path.expanduser("~/.cache/pip")])
+        safe_run(["python", "-c", "import importlib; importlib.invalidate_caches()"])
 
         # Install from the provided source path
-        subprocess.run([sys.executable, "-m", "pip", "install", "-e", source], check=True)
+        safe_run([sys.executable, "-m", "pip", "install", "-e", source], check=True)
         print(f"arm-cli installed from source at {source} successfully!")
     else:
         print("Updating arm-cli from PyPI...")
@@ -63,7 +64,7 @@ def update(ctx, source, force):
                 print("Update cancelled.")
                 return
 
-        subprocess.run([sys.executable, "-m", "pip", "install", "--upgrade", "arm-cli"], check=True)
+        safe_run([sys.executable, "-m", "pip", "install", "--upgrade", "arm-cli"], check=True)
         print("arm-cli updated successfully!")
 
 

arm_cli/system/setup_utils.py

@@ -6,6 +6,7 @@
 import click
 
 from arm_cli.system.shell_scripts import detect_shell, get_current_shell_addins
+from arm_cli.utils.safe_subprocess import safe_run, sudo_run
 
 
 def get_original_user():
@@ -17,7 +18,7 @@ def get_original_user():
 
     # Fallback: try to get from who am i
     try:
-        result = subprocess.run(["who", "am", "i"], capture_output=True, text=True, check=True)
+        result = safe_run(["who", "am", "i"], capture_output=True, text=True, check=True)
         if result.stdout.strip():
             return result.stdout.strip().split()[0]
     except (subprocess.CalledProcessError, IndexError):
@@ -33,13 +34,13 @@ def get_original_user_uid_gid():
 
     try:
         # Get UID
-        uid_result = subprocess.run(
+        uid_result = safe_run(
             ["id", "-u", original_user], capture_output=True, text=True, check=True
         )
         uid = int(uid_result.stdout.strip())
 
         # Get GID
-        gid_result = subprocess.run(
+        gid_result = safe_run(
             ["id", "-g", original_user], capture_output=True, text=True, check=True
         )
         gid = int(gid_result.stdout.strip())
@@ -53,7 +54,7 @@ def get_original_user_uid_gid():
 def check_xhost_setup():
     """Check if xhost is already configured for Docker"""
     try:
-        result = subprocess.run(["xhost"], capture_output=True, text=True, check=True)
+        result = safe_run(["xhost"], capture_output=True, text=True, check=True)
         return "LOCAL:docker" in result.stdout
     except subprocess.CalledProcessError:
         return False
@@ -79,7 +80,7 @@ def setup_xhost(force=False):
                 print("X11 setup cancelled.")
                 return
 
-        subprocess.run(["xhost", "+local:docker"], check=True)
+        safe_run(["xhost", "+local:docker"], check=True)
         print("xhost configured successfully.")
     except subprocess.CalledProcessError as e:
         print(f"Error configuring xhost: {e}")
@@ -88,7 +89,7 @@ def setup_xhost(force=False):
 def check_sudo_privileges():
     """Check if the user has sudo privileges"""
     try:
-        subprocess.run(["sudo", "-n", "true"], check=True, capture_output=True)
+        sudo_run(["-n", "true"], check=True, capture_output=True)
         return True
     except subprocess.CalledProcessError:
         return False
@@ -164,18 +165,18 @@ def setup_data_directories(force=False, data_directory="/DATA"):
         print("Creating directories and setting permissions...")
 
         # Create all directories in one sudo command
-        mkdir_cmd = ["sudo", "mkdir", "-p"] + data_dirs
-        subprocess.run(mkdir_cmd, check=True)
+        mkdir_cmd = ["mkdir", "-p"] + data_dirs
+        sudo_run(mkdir_cmd, check=True)
         print("Created directories.")
 
         # Set ownership for all directories in one sudo command
-        chown_cmd = ["sudo", "chown", "-R", f"{uid}:{gid}"] + data_dirs
-        subprocess.run(chown_cmd, check=True)
+        chown_cmd = ["chown", "-R", f"{uid}:{gid}"] + data_dirs
+        sudo_run(chown_cmd, check=True)
         print("Set ownership.")
 
         # Set permissions for all directories in one sudo command
-        chmod_cmd = ["sudo", "chmod", "-R", "775"] + data_dirs
-        subprocess.run(chmod_cmd, check=True)
+        chmod_cmd = ["chmod", "-R", "775"] + data_dirs
+        sudo_run(chmod_cmd, check=True)
         print("Set permissions.")
 
         print("Data directories setup completed successfully.")
@@ -193,7 +194,7 @@ def setup_data_directories(force=False, data_directory="/DATA"):
 def check_docker_group_setup():
     """Check if the user is already in the docker group"""
     try:
-        result = subprocess.run(["id", "-nG"], capture_output=True, text=True, check=True)
+        result = safe_run(["id", "-nG"], capture_output=True, text=True, check=True)
         groups = result.stdout.strip().split()
         return "docker" in groups
     except subprocess.CalledProcessError:
@@ -225,7 +226,7 @@ def setup_docker_group(force=False):
 
         # Add user to docker group (use original user when running with sudo)
         username = get_original_user()
-        subprocess.run(["sudo", "usermod", "-aG", "docker", username], check=True)
+        sudo_run(["usermod", "-aG", "docker", username], check=True)
 
         print(f"Added {username} to docker group successfully.")
         print("Please log out and back in for the docker group changes to take effect,")

arm_cli/utils/safe_subprocess.py

@@ -0,0 +1,39 @@
+import os
+import subprocess
+
+
+def _validate_cmd(cmd):
+    """
+    Ensure command is a list of strings, no shell=True, no None elements.
+    """
+    if not isinstance(cmd, list):
+        raise ValueError("Command must be a list of arguments")
+    if any(not isinstance(c, str) for c in cmd):
+        raise ValueError("All command arguments must be strings")
+    # Optionally sanitize paths or arguments further here if needed
+
+
+def safe_run(cmd, **kwargs):
+    """
+    Safe subprocess.run wrapper.
+    - cmd must be a list
+    - shell=True is prohibited
+    - check, capture_output, etc. passed through
+    """
+    _validate_cmd(cmd)
+    if kwargs.get("shell", False):
+        raise ValueError("shell=True not allowed for security reasons")
+    return subprocess.run(cmd, **kwargs)  # nosec B603
+
+
+def sudo_run(cmd, **kwargs):
+    """
+    Safe subprocess.run wrapper that runs command with sudo.
+    - Prepends 'sudo' to cmd
+    - Performs same safety checks
+    """
+    _validate_cmd(cmd)
+    if kwargs.get("shell", False):
+        raise ValueError("shell=True not allowed for security reasons")
+    full_cmd = ["sudo"] + cmd
+    return subprocess.run(full_cmd, **kwargs)  # nosec B603