Subprocesses and the Shell
Running other programs safely: argument lists versus shell strings, capturing and streaming output, exit codes, timeouts, and the injection risk in shell=True.
Running other programs safely: argument lists versus shell strings, capturing and streaming output, exit codes, timeouts, and the injection risk in shell=True.
Your photo tool shells out to a converter. It has worked for
months. Then someone uploads a file called
holiday; rm -rf ~/photos.jpg, and the tool does exactly what
the filename says, because the filename was pasted into a
command line and the semicolon means what it always means.
Running another program is routine and the interface to it is sharp. By the end of this lesson you will run commands safely, capture and stream their output, handle exit codes and timeouts, and understand precisely why one keyword argument is the difference between a tool and a vulnerability.
import subprocess
result = subprocess.run(
["git", "rev-parse", "HEAD"],
capture_output=True,
text=True,
check=True,
)
print(result.stdout.strip()) # the commit hashFour arguments carry the lesson.
The list is the command and its arguments, already separated.
capture_output=True collects what the program writes rather
than letting it go to your terminal. text=True decodes bytes
to str — without it you get bytes and every comparison
fails. check=True raises CalledProcessError on a non-zero
exit rather than continuing quietly.
Leaving out check=True is the most common bug here. Without
it, a failed command produces a result object your code
happily carries on with, and the empty stdout looks like an
empty answer rather than a failure.
Bad — building a command string and asking a shell to interpret it.
subprocess.run(f"convert {filename} -resize 800x out.jpg", shell=True)Good — passing arguments directly, with no shell involved.
subprocess.run(
["convert", filename, "-resize", "800x", "out.jpg"], check=True
)Two commands run. The second one is not yours.
The string is handed to /bin/sh, which splits it on
whitespace and interprets ;, |, &&, $(), > and
quotes.
A filename containing any of those characters stops being a filename and becomes syntax.
One command runs, and cannot find the file.
The list is the argument vector. The operating system hands each element to the program as exactly one argument, whatever characters it contains.
Nothing parses it, so there is no syntax to inject.
The rule is short: never pass user input through
shell=True. In practice, never use shell=True at all —
everything it offers has a safe equivalent.
The things people reach for shell=True to get, and their
replacements:
Each replacement is clearer than the shell version and none can
be subverted by a filename. Filtering in Python rather than
piping to grep is also easier to test.
If you genuinely need a real pipeline between two external programs, connect them explicitly:
CalledProcessError
It ran and refused. The exit code and stderr say why, and
the fix is usually in the arguments you passed.
TimeoutExpired
It ran and never finished. Retrying the same call gets you the same hang, so this one needs a decision, not a retry.
FileNotFoundError
It never ran — the program is not installed or not on PATH.
An environment problem, not a data problem.
The exception-design lesson applied to someone else's program: three distinct failures, three distinct exceptions, each needing a different response.
error.stderr is where the useful message lives. A command that
fails almost always explains itself on stderr, and code that
logs only the return code discards the explanation. By
convention, stdout is the program's output and stderr is its
commentary.
Always pass a timeout for anything that could hang. Without one, a subprocess waiting on a network or a lock blocks your program indefinitely, and the symptom is a job that never finishes and never errors.
capture_output=True buffers everything in memory, which is
wrong for a command producing a gigabyte, and wrong when you
want to show progress as it happens.
Popen starts the process and returns immediately, so you read
its output as it arrives.
There is a deadlock worth knowing about. If you write to a
process's stdin and read its stdout with separate calls, both
sides can fill their pipe buffer and wait for the other forever.
communicate() handles both directions at once and is the
correct answer whenever you do both:
A bare name is looked up on PATH, which differs between your
terminal, a cron job and a container:
Checking at startup turns "the nightly job silently did nothing" into an error at the moment the program starts, which is the whole reason to validate at the boundary.
For anything security-sensitive, use an absolute path so the
program you run cannot be replaced by something earlier on a
PATH you do not control.
You can now run other programs without handing a shell your inputs, tell three kinds of failure apart, stream long output, and avoid the pipe deadlock. The rule to carry is the list: it is not a style preference, it is the boundary between an argument and a command.
Next is Configuration and Secrets, which is where several threads meet — the environment variables you just used to run a subprocess, the validation habit from the JSON lesson, and the question of where credentials live so they are neither in your repository nor typed in by hand.
Before you move on, take any shell=True in your code and
convert it to a list. Then, in a scratch directory you do not
mind losing, create a file whose name contains a semicolon and a
harmless command, and run both versions against it. Watching the
shell version execute the filename is the kind of demonstration
that changes habits permanently.
THE CALL
subprocess.run(
["git", "rev-parse", "HEAD"], a LIST, never a string
capture_output=True, collect stdout/stderr
text=True, str, not bytes
check=True, raise on non-zero exit
timeout=30, always, for anything remote
)
NEVER
shell=True with anything a user can influence
a semicolon in a filename becomes a second command
in practice: never shell=True at all
INSTEAD OF THE SHELL
pipeline read stdout, filter in Python
wildcard Path.glob
redirect stdout=open(path, "w")
VAR=x env={**os.environ, "VAR": "x"}
cd cwd="photos"
FAILURES - three different ones
CalledProcessError non-zero exit; error.stderr has the message
TimeoutExpired it hung
FileNotFoundError the program is not installed
log error.stderr, not just the return code
LONG OUTPUT
Popen + iterate over process.stdout stream it
stderr=subprocess.STDOUT keeps ordering
communicate() whenever you also WRITE
- separate calls deadlock
FINDING IT
shutil.which("convert") check at startup, fail early
absolute path when security matters
FIRST
is there a library? httpx, zipfile, pathlib, tarfile...# a pipeline: ls | grep jpg
listing = subprocess.run(["ls"], capture_output=True, text=True)
matches = [l for l in listing.stdout.splitlines() if "jpg" in l]
# a wildcard: convert *.jpg
from pathlib import Path
for photo in Path(".").glob("*.jpg"):
subprocess.run(["convert", str(photo), ...], check=True)
# redirection: command > out.txt
with open("out.txt", "w", encoding="utf-8") as file:
subprocess.run(["command"], stdout=file, check=True)
# environment: VAR=x command
env = {**os.environ, "PHOTO_QUALITY": "90"}
subprocess.run(["command"], env=env, check=True)
# working directory: cd photos && command
subprocess.run(["command"], cwd="photos", check=True)first = subprocess.Popen(["gzip", "-dc", "big.gz"],
stdout=subprocess.PIPE)
second = subprocess.run(["wc", "-l"], stdin=first.stdout,
capture_output=True, text=True)
first.stdout.close()
first.wait()try:
result = subprocess.run(
["convert", str(source), str(target)],
capture_output=True, text=True, check=True, timeout=30,
)
except subprocess.CalledProcessError as error:
log.error("convert failed (%d): %s", error.returncode,
error.stderr.strip())
raise ConversionError(f"could not convert {source}") from error
except subprocess.TimeoutExpired:
raise ConversionError(f"convert timed out on {source}")
except FileNotFoundError as error:
raise ConversionError("convert is not installed") from errorprocess = subprocess.Popen(
["ffmpeg", "-i", str(source), str(target)],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT, # merge, so ordering is preserved
text=True,
bufsize=1, # line buffered
)
with process:
for line in process.stdout:
log.info("ffmpeg: %s", line.rstrip())
if process.returncode != 0:
raise ConversionError(f"ffmpeg exited {process.returncode}")out, err = process.communicate(input=text, timeout=30)import shutil
converter = shutil.which("convert")
if converter is None:
raise ConversionError("convert is not on PATH")