Skip to the solution
MediumBackend13 min read

How do you run external commands with subprocess safely?

1.0k views

The problem

Question presented to candidate: "Your Python service needs to call an external tool, say an image converter, on a file a user uploaded. How do you run it with subprocess, and what can go wrong if you do it carelessly?"

What a strong answer should cover:

  • Use subprocess.run() with the command as a list of arguments; the default shell=False starts the program directly, so user text stays one argument and cannot add commands.
  • Why shell=True with string building is command injection (name; rm -rf ~), and shlex.quote / shlex.join for the rare case a shell is unavoidable.
  • Robustness flags: check=True (raise CalledProcessError on a non-zero exit), capture_output=True and text=True (str instead of bytes), timeout= (the child is killed and TimeoutExpired raised), plus FileNotFoundError when the program is missing.
  • Control what the child sees: input=, cwd=, env= (merge with os.environ rather than replacing it), shutil.which to find the program.
  • Popen for streaming output or building pipelines without a shell, with communicate() to avoid pipe deadlocks; asyncio.create_subprocess_exec in async code.
  • Least privilege: an allow-list of programs and options, and -- before user-supplied file names so they cannot be read as options.

Clarifying questions expected:

  • Which parts of the command come from the user, and does the tool accept options that could be abused?
  • Do we need the output as a stream, or only at the end?

Code / implementation expected: Yes โ€” a subprocess.run([...], check=True, capture_output=True, text=True, timeout=...) call with its error handling.

  • subprocess
  • security

Try it yourself

Sketch your approach before you read the solution. That is what the interview tests.

Stuck? Get a hint.

A short nudge toward the approach. It will not give the answer away.

Solution

The solution is hidden so you can try first

13 min read ยท 10 sections

Run the code

Edit the code and run it here. Each console line with an expected value becomes a test.

Pythonshlex: how a command string becomes argv, injection and quoting (runs in the page)
Pythonsubprocess.run: check, capture_output, text, timeout, input, env and cwd
# subprocess.run the safe way: a list, check=True, captured text, a timeout.
# (The page runtime cannot start processes, so this runs on CPython 3.12.)
import os
import subprocess
import sys

PY = sys.executable          # use the current Python as a portable "external program"

# 1. The everyday call.
r = subprocess.run(
    [PY, "-c", "import sys; print('hello', sys.argv[1:])", "a b", "c"],
    capture_output=True, text=True, check=True, timeout=10,
)
print(type(r).__name__, r.returncode, repr(r.stdout))

# 2. Non-zero exit: without check=True it is silently ignored.
quiet = subprocess.run([PY, "-c", "import sys; sys.exit(3)"])
print("returncode without check:", quiet.returncode)
try:
    subprocess.run([PY, "-c", "import sys; print('boom', file=sys.stderr); sys.exit(3)"],
                   capture_output=True, text=True, check=True)
except subprocess.CalledProcessError as e:
    print("CalledProcessError:", e.returncode, repr(e.stderr))

# 3. A hung child: timeout kills it and raises.
try:
    subprocess.run([PY, "-c", "import time; time.sleep(30)"], timeout=0.5)
except subprocess.TimeoutExpired as e:
    print("TimeoutExpired: child killed, timeout about", round(e.timeout, 1), "s")

# 4. A program that does not exist raises before anything runs.
try:
    subprocess.run(["no-such-program-xyz", "--help"])
except FileNotFoundError as e:
    print("FileNotFoundError:", e.filename)

# 5. Send input on stdin; control the environment and working directory.
r = subprocess.run(
    [PY, "-c", "import os, sys; print(sys.stdin.read().upper(), os.environ['MODE'], os.getcwd() == os.path.realpath('/'))"],
    input="data from python", capture_output=True, text=True, check=True,
    env={**os.environ, "MODE": "batch"}, cwd="/",
)
print(r.stdout.strip())

# 6. Bytes unless you ask for text.
r = subprocess.run([PY, "-c", "print('caf\\u00e9')"], capture_output=True, check=True,
                   env={**os.environ, "PYTHONIOENCODING": "utf-8"})
print(type(r.stdout).__name__, r.stdout, r.stdout.decode("utf-8").strip())
Pythonshell=True injection vs an argument list, a pipeline and streaming with Popen
# shell=True injection, a pipeline without a shell, and streaming with Popen.
# (Needs real processes, so it runs on CPython 3.12, not in the page.)
import shlex
import shutil
import subprocess
import sys

print("echo found on PATH:", shutil.which("echo") is not None)

# A "file name" an attacker typed into a form:
name = "report.txt; echo INJECTED"

# UNSAFE: the whole string goes to /bin/sh, which runs two commands.
unsafe = subprocess.run("echo processing " + name, shell=True,
                        capture_output=True, text=True)
print("shell=True :", unsafe.stdout.splitlines())

# SAFE: a list. The name is one argv element; no shell ever parses it.
safe = subprocess.run(["echo", "processing", name], capture_output=True, text=True)
print("list argv  :", safe.stdout.splitlines())

# If a shell is unavoidable, quote every untrusted piece.
quoted = subprocess.run("echo processing " + shlex.quote(name), shell=True,
                        capture_output=True, text=True)
print("quoted     :", quoted.stdout.splitlines())

# A pipeline (like  printf ... | sort  ) built from two processes, no shell.
p1 = subprocess.Popen(["printf", "pear\\napple\\nfig\\n"], stdout=subprocess.PIPE)
p2 = subprocess.Popen(["sort"], stdin=p1.stdout, stdout=subprocess.PIPE, text=True)
p1.stdout.close()            # so p1 gets SIGPIPE if sort exits early
out, _ = p2.communicate(timeout=10)
p1.wait(timeout=10)
print("pipeline   :", out.split(), "exit codes", p1.returncode, p2.returncode)

# Streaming: read a long-running child line by line as it prints.
code = "import time\nfor i in range(3):\n    print('step', i, flush=True)\n    time.sleep(0.05)"
with subprocess.Popen([sys.executable, "-c", code], stdout=subprocess.PIPE, text=True) as proc:
    for line in proc.stdout:
        print("got:", line.rstrip())
print("streamed child exit code:", proc.returncode)

Discussion

Other approaches, follow-ups and how it went in real interviews.

Sign in to join the discussion.

No comments yet. Share how you would answer it.

Question 52 of 139 in Python

Keep your momentum going.

Mark this question solved to track your progress, then take the next one while it is fresh.

All Python questions
Up nextMedium

What are decorators in Python?

Continue