Python subprocess: Run Shell Commands Safely

Why subprocess?

Python scripts often need to interact with the operating system: run a shell command, call an external tool, or automate a pipeline that spans multiple programs. The subprocess module is the standard way to do this, replacing older functions like os.system and os.popen.

The module gives you full control over how external processes are launched, how their input and output are handled, and how errors are managed.

subprocess.run: The Modern API

For most use cases, subprocess.run() is all you need. It runs a command, waits for it to finish, and returns a CompletedProcess object.

Basic Command Execution

import subprocess

# Run a simple command
result = subprocess.run(['ls', '-la'], capture_output=True, text=True)

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

Always pass the command as a list of strings. Each argument is a separate element. This avoids shell interpretation issues and is more secure.

Capturing Output

import subprocess

# Get the output of a command
result = subprocess.run(
    ['python', '--version'],
    capture_output=True,
    text=True  # Return strings instead of bytes
)
print(result.stdout.strip())  # "Python 3.12.x"

# Without text=True, output is bytes
result_bytes = subprocess.run(
    ['echo', 'hello'],
    capture_output=True
)
print(result_bytes.stdout)  # b'hello\n'

Checking for Errors

import subprocess

# Method 1: Check returncode manually
result = subprocess.run(['ls', '/nonexistent'], capture_output=True, text=True)
if result.returncode != 0:
    print(f"Command failed: {result.stderr}")

# Method 2: Use check=True to raise an exception
try:
    subprocess.run(
        ['ls', '/nonexistent'],
        capture_output=True,
        text=True,
        check=True  # Raises CalledProcessError if returncode != 0
    )
except subprocess.CalledProcessError as e:
    print(f"Command '{e.cmd}' failed with code {e.returncode}")
    print(f"stderr: {e.stderr}")

Timeouts

Prevent commands from running forever:

import subprocess

try:
    result = subprocess.run(
        ['sleep', '30'],
        timeout=5  # Kill after 5 seconds
    )
except subprocess.TimeoutExpired:
    print("Command timed out!")

Working with Input

Passing stdin

import subprocess

# Send input to a command
result = subprocess.run(
    ['grep', 'error'],
    input="line 1: ok\nline 2: error found\nline 3: ok\n",
    capture_output=True,
    text=True
)
print(result.stdout)  # "line 2: error found\n"

# Useful for commands that read from stdin
result = subprocess.run(
    ['wc', '-l'],
    input="one\ntwo\nthree\n",
    capture_output=True,
    text=True
)
print(result.stdout.strip())  # "3"

Feeding a File as Input

import subprocess

# Read from a file
with open('data.txt', 'r') as f:
    result = subprocess.run(
        ['sort'],
        stdin=f,
        capture_output=True,
        text=True
    )
print(result.stdout)

Environment Variables and Working Directory

import subprocess
import os

# Set custom environment variables
env = os.environ.copy()
env['MY_VAR'] = 'hello'
env['DEBUG'] = '1'

result = subprocess.run(
    ['printenv', 'MY_VAR'],
    env=env,
    capture_output=True,
    text=True
)
print(result.stdout.strip())  # "hello"

# Run in a specific directory
result = subprocess.run(
    ['ls'],
    cwd='/tmp',
    capture_output=True,
    text=True
)
print(result.stdout)

Piping Commands Together

In the shell, you pipe commands like cat file | grep pattern | sort. With subprocess, you can do this safely without invoking a shell.

Using Popen for Pipes

import subprocess

# Equivalent to: cat /etc/passwd | grep root | cut -d: -f1
p1 = subprocess.Popen(
    ['cat', '/etc/passwd'],
    stdout=subprocess.PIPE
)

p2 = subprocess.Popen(
    ['grep', 'root'],
    stdin=p1.stdout,
    stdout=subprocess.PIPE
)
p1.stdout.close()  # Allow p1 to receive SIGPIPE if p2 exits

p3 = subprocess.Popen(
    ['cut', '-d:', '-f1'],
    stdin=p2.stdout,
    stdout=subprocess.PIPE,
    text=True
)
p2.stdout.close()

output, _ = p3.communicate()
print(output.strip())

Simpler Pipe with run()

For simple cases, you can pipe within Python itself:

import subprocess

# Step 1: Get output
step1 = subprocess.run(
    ['cat', '/etc/passwd'],
    capture_output=True,
    text=True
)

# Step 2: Filter it
step2 = subprocess.run(
    ['grep', 'root'],
    input=step1.stdout,
    capture_output=True,
    text=True
)

print(step2.stdout)

This is simpler and perfectly fine for most pipelines. Use Popen chaining only when you need true streaming between processes.

Popen for Advanced Control

subprocess.Popen gives you fine-grained control over the process lifecycle:

import subprocess

# Start a long-running process
proc = subprocess.Popen(
    ['ping', '-c', '4', 'localhost'],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

# Read output line by line as it arrives
for line in proc.stdout:
    print(f">> {line.strip()}")

# Wait for completion
proc.wait()
print(f"Exit code: {proc.returncode}")

Communicating with a Process

import subprocess

proc = subprocess.Popen(
    ['python', '-c', 'name = input("Name: "); print(f"Hello, {name}!")'],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

stdout, stderr = proc.communicate(input="Alice\n")
print(stdout)  # "Name: Hello, Alice!\n"

Non-Blocking Reads

import subprocess
import select
import sys

proc = subprocess.Popen(
    ['ping', '-c', '3', 'localhost'],
    stdout=subprocess.PIPE,
    text=True
)

# Poll for output without blocking
while proc.poll() is None:
    # Check if there's data to read
    if proc.stdout and select.select([proc.stdout], [], [], 0.1)[0]:
        line = proc.stdout.readline()
        if line:
            print(f"[live] {line.strip()}")

print(f"Process finished with code {proc.returncode}")

Security: Avoiding Shell Injection

The most important rule: never use shell=True with user input.

import subprocess

# DANGEROUS: shell injection vulnerability
user_input = "file.txt; rm -rf /"  # Malicious input
subprocess.run(f"cat {user_input}", shell=True)  # Executes rm -rf /!

# SAFE: pass arguments as a list
user_input = "file.txt; rm -rf /"
subprocess.run(['cat', user_input])  # Treats the whole string as a filename
# Error: No such file: "file.txt; rm -rf /"

When shell=True Is Acceptable

Sometimes you genuinely need shell features like globbing, environment variable expansion, or complex redirections:

import subprocess

# Acceptable: hardcoded command with shell features
result = subprocess.run(
    'echo $HOME',
    shell=True,
    capture_output=True,
    text=True
)
print(result.stdout.strip())

# If you must use shell=True with variables, use shlex.quote
import shlex

user_filename = "my file.txt"
safe_name = shlex.quote(user_filename)
subprocess.run(f"cat {safe_name}", shell=True, capture_output=True)

But even then, you can usually avoid shell=True:

import subprocess
import os
import glob

# Instead of shell=True for globbing
files = glob.glob('*.py')
result = subprocess.run(['wc', '-l'] + files, capture_output=True, text=True)

# Instead of shell=True for environment variables
home = os.environ['HOME']
result = subprocess.run(['ls', home], capture_output=True, text=True)

Practical Examples

Running Git Commands

import subprocess

def git_status():
    result = subprocess.run(
        ['git', 'status', '--porcelain'],
        capture_output=True,
        text=True,
        check=True
    )
    return result.stdout.strip()

def git_current_branch():
    result = subprocess.run(
        ['git', 'rev-parse', '--abbrev-ref', 'HEAD'],
        capture_output=True,
        text=True,
        check=True
    )
    return result.stdout.strip()

def git_commit_count():
    result = subprocess.run(
        ['git', 'rev-list', '--count', 'HEAD'],
        capture_output=True,
        text=True,
        check=True
    )
    return int(result.stdout.strip())

print(f"Branch: {git_current_branch()}")
print(f"Commits: {git_commit_count()}")
print(f"Status:\n{git_status()}")

Running Docker Commands

import subprocess
import json

def docker_ps():
    """List running containers as structured data."""
    result = subprocess.run(
        ['docker', 'ps', '--format', '{{json .}}'],
        capture_output=True,
        text=True,
        check=True
    )
    containers = []
    for line in result.stdout.strip().split('\n'):
        if line:
            containers.append(json.loads(line))
    return containers

def docker_build(tag, dockerfile='.'):
    """Build a Docker image with streaming output."""
    proc = subprocess.Popen(
        ['docker', 'build', '-t', tag, dockerfile],
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True
    )
    for line in proc.stdout:
        print(line, end='')

    proc.wait()
    if proc.returncode != 0:
        raise RuntimeError(f"Docker build failed with code {proc.returncode}")

A Subprocess Wrapper Utility

import subprocess
import logging
from dataclasses import dataclass

logger = logging.getLogger(__name__)

@dataclass
class CommandResult:
    command: list[str]
    returncode: int
    stdout: str
    stderr: str
    success: bool

def run_command(
    cmd: list[str],
    cwd: str | None = None,
    env: dict | None = None,
    timeout: int = 60,
    check: bool = False,
    input_data: str | None = None,
) -> CommandResult:
    """Run a command with sensible defaults and structured output."""
    logger.debug(f"Running: {' '.join(cmd)}")

    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            cwd=cwd,
            env=env,
            timeout=timeout,
            input=input_data,
        )
    except subprocess.TimeoutExpired:
        logger.error(f"Command timed out after {timeout}s: {cmd}")
        return CommandResult(
            command=cmd,
            returncode=-1,
            stdout="",
            stderr=f"Timed out after {timeout} seconds",
            success=False,
        )
    except FileNotFoundError:
        logger.error(f"Command not found: {cmd[0]}")
        return CommandResult(
            command=cmd,
            returncode=-1,
            stdout="",
            stderr=f"Command not found: {cmd[0]}",
            success=False,
        )

    cmd_result = CommandResult(
        command=cmd,
        returncode=result.returncode,
        stdout=result.stdout,
        stderr=result.stderr,
        success=result.returncode == 0,
    )

    if check and not cmd_result.success:
        raise subprocess.CalledProcessError(
            result.returncode, cmd, result.stdout, result.stderr
        )

    return cmd_result

# Usage
result = run_command(['python', '--version'])
if result.success:
    print(result.stdout)

Common Pitfalls

Deadlocks with Popen

If you read stdout and stderr separately, you can deadlock when the OS pipe buffer fills:

import subprocess

# WRONG: can deadlock
proc = subprocess.Popen(['cmd'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
out = proc.stdout.read()   # Blocks if stderr buffer fills
err = proc.stderr.read()

# RIGHT: use communicate()
proc = subprocess.Popen(['cmd'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
out, err = proc.communicate()  # Reads both streams safely

Forgetting to Handle Encoding

import subprocess

# Binary output (default)
result = subprocess.run(['echo', 'hello'], capture_output=True)
print(type(result.stdout))  # <class 'bytes'>

# Text output
result = subprocess.run(['echo', 'hello'], capture_output=True, text=True)
print(type(result.stdout))  # <class 'str'>

# Specific encoding for non-UTF-8 output
result = subprocess.run(
    ['some_command'],
    capture_output=True,
    encoding='latin-1'
)

Wrapping Up

subprocess.run() handles the vast majority of use cases for running external commands. Pass commands as lists (not strings), use capture_output=True and text=True for readable output, and check=True to catch failures. Reach for Popen only when you need streaming output, pipes between processes, or non-blocking reads. Most importantly, never use shell=True with untrusted input — argument lists are inherently safe from injection attacks.