feat(security): implement LiteralString for subprocess security

- Add `safe_docker_exec` wrapper for Docker command construction
- Implement `safe_solc_command` for compiler argument validation
- Add `safe_t8n_args` for transition tool parameter security
- Use `LiteralString` type hints to prevent command injection
- Validate file paths, EVM versions, and numeric parameters

Author: danceratopz

src/cli/extract_config.py

@@ -11,7 +11,7 @@
 import subprocess
 import sys
 from pathlib import Path
-from typing import Dict, Optional, Tuple, cast
+from typing import Dict, LiteralString, Optional, Tuple, cast
 
 import click
 from hive.simulation import Simulation
@@ -26,9 +26,21 @@
 from pytest_plugins.consume.simulators.helpers.ruleset import ruleset
 
 
+def safe_docker_exec(container_id: str, command: LiteralString, *args: LiteralString) -> list[str]:
+    """Safely construct docker exec commands with literal strings."""
+    return ["docker", "exec", container_id, command] + list(args)
+
+
+def safe_docker_command(command: LiteralString, *args: LiteralString) -> list[str]:
+    """Safely construct docker commands with literal strings."""
+    return ["docker", command] + list(args)
+
+
 def get_docker_containers() -> set[str]:
     """Get the current list of Docker container IDs."""
-    result = subprocess.run(["docker", "ps", "-q"], capture_output=True, text=True, check=True)
+    result = subprocess.run(
+        safe_docker_command("ps", "-q"), capture_output=True, text=True, check=True
+    )
     return set(result.stdout.strip().split("\n")) if result.stdout.strip() else set()
 
 
@@ -56,12 +68,12 @@ def extract_client_files(
         try:
             # Use docker exec to read the file from the container
             # First check if file exists
-            check_cmd = ["docker", "exec", container_id, "test", "-f", container_path]
+            check_cmd = safe_docker_exec(container_id, "test", "-f", container_path)
             check_result = subprocess.run(check_cmd, capture_output=True)
 
             if check_result.returncode == 0:
                 # File exists, now read it
-                read_cmd = ["docker", "exec", container_id, "cat", container_path]
+                read_cmd = safe_docker_exec(container_id, "cat", container_path)
                 result = subprocess.run(read_cmd, capture_output=True, text=True)
 
                 if result.returncode == 0 and result.stdout:
@@ -267,7 +279,7 @@ def extract_config(
                 # Optionally list files in container
                 if list_files:
                     click.echo("\nListing files in container root:")
-                    list_cmd = ["docker", "exec", container_id, "ls", "-la", "/"]
+                    list_cmd = safe_docker_exec(container_id, "ls", "-la", "/")
                     result = subprocess.run(list_cmd, capture_output=True, text=True)
                     if result.returncode == 0:
                         click.echo(result.stdout)

src/ethereum_clis/transition_tool.py

@@ -10,7 +10,7 @@
 from abc import abstractmethod
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Dict, List, Mapping, Optional, Type
+from typing import Any, Dict, List, LiteralString, Mapping, Optional, Type
 from urllib.parse import urlencode
 
 from requests import Response
@@ -20,6 +20,7 @@
 from ethereum_test_base_types import BlobSchedule
 from ethereum_test_exceptions import ExceptionMapper
 from ethereum_test_forks import Fork
+from ethereum_test_forks.helpers import get_development_forks, get_forks
 from ethereum_test_types import Alloc, Environment, Transaction
 
 from .ethereum_cli import EthereumCLI
@@ -38,6 +39,12 @@
 SLOW_REQUEST_TIMEOUT = 180
 
 
+def get_valid_transition_tool_names() -> set[str]:
+    """Get all valid transition tool names from deployed and development forks."""
+    all_available_forks = get_forks() + get_development_forks()
+    return {fork.transition_tool_name() for fork in all_available_forks}
+
+
 class TransitionTool(EthereumCLI):
     """
     Transition tool abstract base class which should be inherited by all transition tool
@@ -424,6 +431,49 @@ def _evaluate_stream(
 
         return output
 
+    def safe_t8n_args(
+        self, fork_name: str, chain_id: int, reward: int, temp_dir=None
+    ) -> List[str]:
+        """Safely construct t8n arguments with validated inputs."""
+        # Validate fork name against actual transition tool names from all available forks
+        valid_forks = get_valid_transition_tool_names()
+        if fork_name not in valid_forks:
+            raise ValueError(f"Invalid fork name: {fork_name}")
+
+        # Validate chain ID (should be positive integer)
+        if not isinstance(chain_id, int) or chain_id <= 0:
+            raise ValueError(f"Invalid chain ID: {chain_id}")
+
+        # Validate reward (should be non-negative integer)
+        if not isinstance(reward, int) or reward < 0:
+            raise ValueError(f"Invalid reward: {reward}")
+
+        # Use literal strings for command flags
+        input_alloc: LiteralString = "--input.alloc=stdin"
+        input_txs: LiteralString = "--input.txs=stdin"
+        input_env: LiteralString = "--input.env=stdin"
+        output_result: LiteralString = "--output.result=stdout"
+        output_alloc: LiteralString = "--output.alloc=stdout"
+        output_body: LiteralString = "--output.body=stdout"
+        trace_flag: LiteralString = "--trace"
+
+        args = [
+            input_alloc,
+            input_txs,
+            input_env,
+            output_result,
+            output_alloc,
+            output_body,
+            f"--state.fork={fork_name}",
+            f"--state.chainid={chain_id}",
+            f"--state.reward={reward}",
+        ]
+
+        if self.trace and temp_dir:
+            args.extend([trace_flag, f"--output.basedir={temp_dir.name}"])
+
+        return args
+
     def construct_args_stream(
         self, t8n_data: TransitionToolData, temp_dir: tempfile.TemporaryDirectory
     ) -> List[str]:
@@ -432,22 +482,10 @@ def construct_args_stream(
         if self.subcommand:
             command.append(self.subcommand)
 
-        args = command + [
-            "--input.alloc=stdin",
-            "--input.txs=stdin",
-            "--input.env=stdin",
-            "--output.result=stdout",
-            "--output.alloc=stdout",
-            "--output.body=stdout",
-            f"--state.fork={t8n_data.fork_name}",
-            f"--state.chainid={t8n_data.chain_id}",
-            f"--state.reward={t8n_data.reward}",
-        ]
-
-        if self.trace:
-            args.append("--trace")
-            args.append(f"--output.basedir={temp_dir.name}")
-        return args
+        safe_args = self.safe_t8n_args(
+            t8n_data.fork_name, t8n_data.chain_id, t8n_data.reward, temp_dir
+        )
+        return command + safe_args
 
     def dump_debug_stream(
         self,

src/ethereum_test_specs/static_state/common/compile_yul.py

@@ -1,6 +1,54 @@
 """compile yul with arguments."""
 
 import subprocess
+from pathlib import Path
+from typing import LiteralString
+
+
+def safe_solc_command(
+    source_file: Path | str, evm_version: str | None = None, optimize: str | None = None
+) -> list[str]:
+    """Safely construct solc command with validated inputs."""
+    # Validate source file path
+    source_path = Path(source_file)
+    if not source_path.exists():
+        raise FileNotFoundError(f"Source file not found: {source_file}")
+    if source_path.suffix not in (".yul", ".sol"):
+        raise ValueError(f"Invalid file extension for solc: {source_path.suffix}")
+
+    cmd: list[str] = ["solc"]
+
+    # Add EVM version if provided (validate against known versions)
+    if evm_version:
+        valid_versions = {
+            "homestead",
+            "tangerineWhistle",
+            "spuriousDragon",
+            "byzantium",
+            "constantinople",
+            "petersburg",
+            "istanbul",
+            "berlin",
+            "london",
+            "paris",
+            "shanghai",
+            "cancun",
+        }
+        if evm_version not in valid_versions:
+            raise ValueError(f"Invalid EVM version: {evm_version}")
+        cmd.extend(["--evm-version", evm_version])
+
+    # Add compilation flags (using literal strings)
+    strict_assembly: LiteralString = "--strict-assembly"
+    cmd.append(strict_assembly)
+
+    if optimize is None:
+        optimize_flag: LiteralString = "--optimize"
+        yul_opts: LiteralString = "--yul-optimizations=:"
+        cmd.extend([optimize_flag, yul_opts])
+
+    cmd.append(str(source_path))
+    return cmd
 
 
 def compile_yul(source_file: str, evm_version: str | None = None, optimize: str | None = None):
@@ -19,17 +67,7 @@ def compile_yul(source_file: str, evm_version: str | None = None, optimize: str
     Raises_:
         Exception: If the solc output contains an error message.
     """
-    cmd = ["solc"]
-    if evm_version:
-        cmd.extend(["--evm-version", evm_version])
-
-    # Choose flags based on whether flag is provided
-    if optimize is None:
-        # When flag is not provided, include extra optimization flags
-        cmd.extend(["--strict-assembly", "--optimize", "--yul-optimizations=:", source_file])
-    else:
-        # Otherwise, omit the optimization flags
-        cmd.extend(["--strict-assembly", source_file])
+    cmd = safe_solc_command(source_file, evm_version, optimize)
 
     # Execute the solc command and capture both stdout and stderr
     result = subprocess.run(