Subprocess Module: Safe Process Execution and Command Running

June 15, 2025

TL;DR

The subprocess module provides subprocess.run(), subprocess.Popen(), and subprocess.PIPE for secure process execution with proper input/output handling, error management, and cross-platform command execution.

Interesting!

The subprocess module is much safer than os.system() because it prevents shell injection attacks by default - when you pass a list of arguments instead of a string, it doesn’t interpret shell metacharacters that could be exploited.

Basic Command Execution

import subprocess

# Simple command execution
result = subprocess.run(['ls', '-la'], capture_output=True, text=True)

print(f"Return code: {result.returncode}")
print(f"stdout: {result.stdout}")
print(f"stderr: {result.stderr}")

# Check if command succeeded
if result.returncode == 0:
    print("Command succeeded")
else:
    print("Command failed")

# Raise exception on non-zero exit
try:
    result = subprocess.run(['ls', '/nonexistent'], 
                          capture_output=True, 
                          text=True, 
                          check=True)
except subprocess.CalledProcessError as e:
    print(f"Command failed with return code {e.returncode}")
    print(f"Error output: {e.stderr}")

Different Ways to Handle Output

import subprocess

# Capture output as strings
result = subprocess.run(['echo', 'Hello World'], 
                       capture_output=True, 
                       text=True)
print(result.stdout)  # 'Hello World\n'

# Capture output as bytes
result = subprocess.run(['echo', 'Hello World'], 
                       capture_output=True)
print(result.stdout)  # b'Hello World\n'

# Don't capture output (print to console)
subprocess.run(['echo', 'Hello World'])

# Redirect output to file
with open('output.txt', 'w') as f:
    subprocess.run(['ls', '-la'], stdout=f, text=True)

# Combine stdout and stderr
result = subprocess.run(['ls', '/nonexistent'], 
                       capture_output=True, 
                       text=True, 
                       stderr=subprocess.STDOUT)
print(result.stdout)  # Contains both stdout and stderr

Input Handling

import subprocess

# Pass input to command
result = subprocess.run(['grep', 'python'], 
                       input='I love python\nJava is okay\n', 
                       capture_output=True, 
                       text=True)
print(result.stdout)  # 'I love python\n'

# Pass input from file
with open('input.txt', 'r') as f:
    result = subprocess.run(['sort'], 
                           stdin=f, 
                           capture_output=True, 
                           text=True)

# Chain commands with pipes
# Equivalent to: echo "hello world" | tr 'a-z' 'A-Z'
p1 = subprocess.run(['echo', 'hello world'], capture_output=True, text=True)
p2 = subprocess.run(['tr', 'a-z', 'A-Z'], 
                   input=p1.stdout, 
                   capture_output=True, 
                   text=True)
print(p2.stdout)  # 'HELLO WORLD\n'

Advanced Process Management with Popen

import subprocess
import time

# Long-running process
process = subprocess.Popen(['python', '-c', '''
import time
for i in range(5):
    print(f"Working... {i}")
    time.sleep(1)
'''], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)

# Check if process is still running
if process.poll() is None:
    print("Process is still running")

# Wait for completion with timeout
try:
    stdout, stderr = process.communicate(timeout=10)
    print(f"Process completed with return code: {process.returncode}")
    print(f"Output: {stdout}")
except subprocess.TimeoutExpired:
    print("Process timed out")
    process.kill()
    stdout, stderr = process.communicate()

Real-time Process Communication

import subprocess
import threading

def read_output(process):
    """Read process output in real-time"""
    for line in iter(process.stdout.readline, ''):
        print(f"Output: {line.strip()}")
    process.stdout.close()

# Start long-running process
process = subprocess.Popen(['python', '-c', '''
import time
import sys
for i in range(10):
    print(f"Step {i}", flush=True)
    time.sleep(0.5)
'''], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)

# Read output in background thread
output_thread = threading.Thread(target=read_output, args=(process,))
output_thread.start()

# Do other work while process runs
print("Doing other work...")
time.sleep(2)

# Wait for process to complete
process.wait()
output_thread.join()

Environment Variables and Working Directory

import subprocess
import os

# Set environment variables
env = os.environ.copy()
env['MY_VAR'] = 'custom_value'
env['PATH'] = '/custom/path:' + env['PATH']

result = subprocess.run(['printenv', 'MY_VAR'], 
                       env=env, 
                       capture_output=True, 
                       text=True)
print(result.stdout)  # 'custom_value\n'

# Change working directory
result = subprocess.run(['pwd'], 
                       cwd='/tmp', 
                       capture_output=True, 
                       text=True)
print(result.stdout)  # '/tmp\n'

# Use shell=True for shell features (be careful!)
result = subprocess.run('echo $HOME', 
                       shell=True, 
                       capture_output=True, 
                       text=True)
print(result.stdout)

Practical Examples

Git Operations

import subprocess

def git_status():
    """Get git status"""
    try:
        result = subprocess.run(['git', 'status', '--porcelain'], 
                               capture_output=True, 
                               text=True, 
                               check=True)
        return result.stdout.strip().split('\n') if result.stdout.strip() else []
    except subprocess.CalledProcessError:
        return None

def git_commit(message):
    """Create git commit"""
    try:
        # Add all changes
        subprocess.run(['git', 'add', '.'], check=True)
        
        # Commit with message
        subprocess.run(['git', 'commit', '-m', message], check=True)
        print(f"Committed: {message}")
        return True
    except subprocess.CalledProcessError as e:
        print(f"Git commit failed: {e}")
        return False

# Usage
changes = git_status()
if changes:
    print(f"Found {len(changes)} changes")
    # git_commit("Auto-commit changes")

File Processing Pipeline

import subprocess

def process_log_files(directory):
    """Process log files using command-line tools"""
    
    # Find all log files
    find_result = subprocess.run(['find', directory, '-name', '*.log'], 
                                capture_output=True, 
                                text=True)
    
    log_files = find_result.stdout.strip().split('\n')
    
    for log_file in log_files:
        if not log_file:
            continue
            
        print(f"Processing: {log_file}")
        
        # Count lines
        wc_result = subprocess.run(['wc', '-l', log_file], 
                                  capture_output=True, 
                                  text=True)
        line_count = wc_result.stdout.split()[0]
        
        # Count errors
        grep_result = subprocess.run(['grep', '-c', 'ERROR', log_file], 
                                    capture_output=True, 
                                    text=True)
        error_count = grep_result.stdout.strip() if grep_result.returncode == 0 else '0'
        
        print(f"  Lines: {line_count}, Errors: {error_count}")

# Usage
# process_log_files('/var/log')

System Information

import subprocess
import json

def get_system_info():
    """Gather system information"""
    info = {}
    
    # Operating system
    try:
        result = subprocess.run(['uname', '-a'], 
                               capture_output=True, 
                               text=True)
        info['os'] = result.stdout.strip()
    except FileNotFoundError:
        # Windows
        result = subprocess.run(['systeminfo'], 
                               capture_output=True, 
                               text=True)
        info['os'] = 'Windows'
    
    # Python version
    result = subprocess.run(['python', '--version'], 
                           capture_output=True, 
                           text=True)
    info['python'] = result.stdout.strip()
    
    # Disk usage
    try:
        result = subprocess.run(['df', '-h'], 
                               capture_output=True, 
                               text=True)
        info['disk_usage'] = result.stdout
    except FileNotFoundError:
        # Windows
        result = subprocess.run(['dir', '/-c'], 
                               shell=True, 
                               capture_output=True, 
                               text=True)
        info['disk_usage'] = result.stdout
    
    return info

# Usage
system_info = get_system_info()
print(json.dumps(system_info, indent=2))

Process Monitoring

import subprocess
import time
import signal

class ProcessMonitor:
    def __init__(self, command):
        self.command = command
        self.process = None
        self.running = False
    
    def start(self):
        """Start the monitored process"""
        if self.process is None:
            self.process = subprocess.Popen(self.command, 
                                          stdout=subprocess.PIPE, 
                                          stderr=subprocess.PIPE, 
                                          text=True)
            self.running = True
            print(f"Started process {self.process.pid}")
        
    def is_running(self):
        """Check if process is still running"""
        if self.process:
            return self.process.poll() is None
        return False
    
    def stop(self, timeout=5):
        """Stop the process gracefully"""
        if self.process and self.is_running():
            print(f"Stopping process {self.process.pid}")
            self.process.terminate()
            
            try:
                self.process.wait(timeout=timeout)
            except subprocess.TimeoutExpired:
                print("Process didn't terminate, killing it")
                self.process.kill()
                self.process.wait()
            
            self.running = False
    
    def restart(self):
        """Restart the process"""
        self.stop()
        time.sleep(1)
        self.process = None
        self.start()

# Usage
# monitor = ProcessMonitor(['python', 'my_server.py'])
# monitor.start()
# time.sleep(10)
# monitor.restart()
# monitor.stop()

Security Best Practices

import subprocess
import shlex

# Safe command construction
def safe_execute(command_parts):
    """Safely execute a command with user input"""
    
    # Validate input
    if not isinstance(command_parts, list):
        raise ValueError("Command must be a list of strings")
    
    # Avoid shell=True when possible
    try:
        result = subprocess.run(command_parts, 
                               capture_output=True, 
                               text=True, 
                               check=True)
        return result.stdout
    except subprocess.CalledProcessError as e:
        print(f"Command failed: {e}")
        return None

# If you must use shell=True, sanitize input
def shell_execute(command_string):
    """Execute shell command with proper sanitization"""
    
    # Use shlex.quote for shell arguments
    # This prevents shell injection
    safe_command = shlex.quote(command_string)
    
    result = subprocess.run(safe_command, 
                           shell=True, 
                           capture_output=True, 
                           text=True)
    return result.stdout

# Example: Safe file operations
def safe_file_search(directory, pattern):
    """Safely search for files"""
    # Validate inputs
    if not directory.replace('_', '').replace('-', '').replace('/', '').replace('.', '').isalnum():
        raise ValueError("Invalid directory name")
    
    # Use list form (safer)
    result = subprocess.run(['find', directory, '-name', pattern], 
                           capture_output=True, 
                           text=True)
    
    if result.returncode == 0:
        return result.stdout.strip().split('\n')
    return []

# Bad example (vulnerable to injection)
# user_input = "file.txt; rm -rf /"
# subprocess.run(f"cat {user_input}", shell=True)  # DON'T DO THIS!

# Good example (safe)
# user_input = "file.txt; rm -rf /"
# subprocess.run(['cat', user_input])  # Safe - treats as literal filename

Error Handling and Timeouts

import subprocess

def robust_command_execution(command, timeout=30):
    """Execute command with comprehensive error handling"""
    
    try:
        result = subprocess.run(command, 
                               capture_output=True, 
                               text=True, 
                               timeout=timeout, 
                               check=True)
        return {
            'success': True,
            'stdout': result.stdout,
            'stderr': result.stderr,
            'returncode': result.returncode
        }
    
    except subprocess.TimeoutExpired as e:
        return {
            'success': False,
            'error': 'timeout',
            'message': f"Command timed out after {timeout} seconds",
            'partial_stdout': e.stdout,
            'partial_stderr': e.stderr
        }
    
    except subprocess.CalledProcessError as e:
        return {
            'success': False,
            'error': 'non_zero_exit',
            'message': f"Command failed with return code {e.returncode}",
            'stdout': e.stdout,
            'stderr': e.stderr,
            'returncode': e.returncode
        }
    
    except FileNotFoundError:
        return {
            'success': False,
            'error': 'command_not_found',
            'message': f"Command not found: {command[0]}"
        }
    
    except Exception as e:
        return {
            'success': False,
            'error': 'unknown',
            'message': str(e)
        }

# Usage
result = robust_command_execution(['ls', '-la'], timeout=10)
if result['success']:
    print(result['stdout'])
else:
    print(f"Error ({result['error']}): {result['message']}")

Performance Tips

import subprocess
from concurrent.futures import ThreadPoolExecutor
import time

# Parallel command execution
def execute_commands_parallel(commands, max_workers=4):
    """Execute multiple commands in parallel"""
    
    def run_command(cmd):
        start_time = time.time()
        result = subprocess.run(cmd, capture_output=True, text=True)
        end_time = time.time()
        
        return {
            'command': cmd,
            'returncode': result.returncode,
            'stdout': result.stdout,
            'stderr': result.stderr,
            'duration': end_time - start_time
        }
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        results = list(executor.map(run_command, commands))
    
    return results

# Usage
commands = [
    ['ls', '-la'],
    ['ps', 'aux'],
    ['df', '-h'],
    ['free', '-h']
]

results = execute_commands_parallel(commands)
for result in results:
    print(f"Command {result['command']} took {result['duration']:.2f}s")

The subprocess module is the modern, secure way to execute external commands and manage processes in Python applications.

Subprocess operations complement os module system operations and work well with pathlib for file management in system automation tasks.

Reference: Python Subprocess Module Documentation