Running commands safely · Prasanta Kumar Behera

Call system commands from Python safely: argument lists instead of shell strings, checking exit codes, capturing output, timeouts, and streaming long-running output, plus a reusable run() helper.

Python as the conductor

Your Python tool will often call existing commands: kubectl, systemctl, git, terraform. The subprocess module does this, safely if you follow three rules: pass a list, check the exit code, set a timeout.

Asking the shell to run a sentence is like shouting instructions across a noisy playground: words can get mixed up, and a prankster can shout extra instructions in the middle ("…and then delete everything!"). Passing a list is like handing over a written card with each instruction in its own box. Nothing can sneak in between.

The basic call

import subprocess

result = subprocess.run(
    ["systemctl", "is-active", "nginx"],    # a list: program + arguments
    capture_output=True, text=True,         # capture stdout/stderr as str
    timeout=10,                             # never hang forever
)
print(result.returncode, result.stdout.strip())   # 0 active   (or 3 inactive)

Raise on failure with check=True

try:
    out = subprocess.run(
        ["kubectl", "get", "nodes", "-o", "name"],
        capture_output=True, text=True, check=True, timeout=30,
    ).stdout
except FileNotFoundError:
    raise SystemExit("kubectl is not installed or not on PATH")
except subprocess.CalledProcessError as e:
    raise SystemExit(f"kubectl failed ({e.returncode}): {e.stderr.strip()}")
except subprocess.TimeoutExpired:
    raise SystemExit("kubectl timed out after 30s")

nodes = out.splitlines()

Three different failures, three clear messages: not installed, returned an error, hung.

Never build shell strings from input

host = input("host: ")                                  # imagine: "x; rm -rf ~"

subprocess.run(f"ping -c1 {host}", shell=True)          # ❌ injection
subprocess.run(["ping", "-c1", host])                   # ✅ host is one argument, always

If you truly need shell features (pipes, globs), prefer doing that part in Python. If you must use a shell, quote every untrusted value with shlex.quote().

Streaming long-running output

run() waits until the command finishes. For long jobs where you want lines as they appear (a terraform apply, kubectl logs -f), use Popen:

import subprocess

with subprocess.Popen(
    ["kubectl", "rollout", "status", "deploy/web", "--timeout=120s"],
    stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
) as proc:
    for line in proc.stdout:
        print("rollout:", line.rstrip())
if proc.returncode != 0:
    raise SystemExit(f"rollout failed with {proc.returncode}")

A reusable helper

Most tools end up with one small wrapper so every command is logged, timed out and checked the same way:

import logging, shlex, subprocess

log = logging.getLogger(__name__)

def run(cmd: list[str], timeout: int = 60, check: bool = True) -> str:
    """Run a command, log it, return stdout. Raises on failure or timeout."""
    log.info("run: %s", shlex.join(cmd))
    res = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    if check and res.returncode != 0:
        log.error("failed (%s): %s", res.returncode, res.stderr.strip())
        raise subprocess.CalledProcessError(res.returncode, cmd, res.stdout, res.stderr)
    return res.stdout

Prefer a library when one exists

Parsing kubectl or aws CLI output works, but official client libraries (lessons 06 and 07) give you real objects, proper errors and no parsing. Use subprocess for tools without a good library (systemctl, git, vendor CLIs).

Try it: a service checker

Write check_services.py that takes service names as arguments (sys.argv[1:]) and runs systemctl is-active for each, with a 5-second timeout.
Print a table of service → state, and exit with code 1 if any are not active.
Handle a missing systemctl (e.g. run it in a container) with a clear message.
Try passing a name like "nginx; id" and confirm nothing but systemctl runs.
Add the run() helper with logging.basicConfig(level=logging.INFO) and look at the log lines.

Going deeper: subprocess at scale

Running commands on many hosts? Use concurrent.futures.ThreadPoolExecutor with a bounded pool and per-task timeouts (lesson 09), or hand the job to Ansible.
Pass environment deliberately (env={**os.environ, "KUBECONFIG": path}), and never log secrets that appear in commands or environments.
Remember that the child's exit code, stdout and stderr are your API with that tool. Pin tool versions if you parse their output.

Recap

subprocess.run([...], capture_output=True, text=True, check=True, timeout=N).
Lists, not shell strings. shell=True + input = injection.
Handle FileNotFoundError, CalledProcessError and TimeoutExpired separately.
Popen to stream long output; a small run() helper keeps every call consistent.

This site is a public version of my personal engineering knowledge hub. It intentionally excludes confidential company information and internal operational details.