Driving systems safely: subprocess, SSH & HTTP

subprocess without shell=True, paramiko/SSH, and resilient API clients.

Most security automation is glue. Your Python script reaches out and drives something else. It runs a local command to read the state of a box, opens an SSH (Secure Shell, the standard encrypted way to log into a remote machine and run commands) session to a fleet of servers, or calls a web API (Application Programming Interface, a service's set of network endpoints your code can talk to) to pull findings out of a scanner. Every one of those is a door to the world outside your process, and every door is a place you either build safe or leave propped open.

The rule behind all three is the same. Never hand data you don't control to something that will re-interpret it. Put a time limit on every wait. Check the result before you trust it. Get those three habits right and a whole class of outages and break-ins never gets started.

Run commands without a shell in the middle

When you type into a terminal, a shell (a program like bash or sh that reads your command line and decides what to run) does a lot of work before anything executes. It hunts for special punctuation and acts on it. A semicolon means 'now run a second command.' A pipe character means 'feed this program's output into the next one.' The dollar-parenthesis form, $(...), means 'run what's inside and paste the result here.' That eagerness to help is the danger. If any part of the command line came from a user, a filename, a web response, or a log line, then whoever controlled that text can slip in their own punctuation, and their own commands ride along.

In Python you invite the shell in by passing one big string with shell=True. Python hands that whole string to /bin/sh, which happily re-reads it, punctuation and all. Here is what that habit costs.

The shell saw echo scanning prod; id, ran the echo, then obeyed the semicolon and ran id as a second command. Swap id for something that deletes files or opens a reverse shell (a hidden connection back to the attacker that hands them a command prompt on your box) and the story is much worse. Now the same value, through the list form.

No shell ran. Python called execve, a system call (a direct request to the kernel, the core of the operating system that starts and manages programs) that launches echo with its arguments handed over already split into separate items. Nothing re-reads them, so the semicolon stays text, the whole prod; id is a single argument, and echo prints it back word for word. This is the safe default for every external command you run.

import json, subprocess
 
def get_pods(namespace: str) -> dict:
    try:
        r = subprocess.run(
            ["kubectl", "get", "pods", "-n", namespace, "-o", "json"],
            capture_output=True,   # collect stdout and stderr
            text=True,             # decode bytes to str
            timeout=30,            # a hung kubectl can't hang the job
            check=True,            # non-zero exit raises CalledProcessError
        )
    except subprocess.TimeoutExpired:
        raise SystemExit("kubectl timed out after 30s")
    except subprocess.CalledProcessError as e:
        raise SystemExit(f"kubectl failed ({e.returncode}): {e.stderr.strip()}")
    return json.loads(r.stdout)

Four arguments do the safety work. capture_output pulls stdout and stderr back to you instead of leaking them to the console. text decodes the bytes so you get strings. timeout puts a hard ceiling on the wait, and a child that blows past it raises TimeoutExpired instead of hanging your pipeline. check=True turns a non-zero exit code into a CalledProcessError, so a failed command stops the script instead of sailing on with empty output. If you think you need a shell for a pipe or a wildcard, you almost never do: run each stage as its own process and wire them together in Python, or expand a wildcard with glob. And if you genuinely must build a shell command line, wrap every untrusted piece in shlex.quote first.

The quotes fold the whole value into one inert argument the shell won't split on. Use that only as a last resort, though; the list form is the thing you should reach for by default.

SSH that checks who actually answered

SSH guards against more than eavesdroppers. It also keeps you from talking to an impostor. Every SSH server has a host key, a unique cryptographic identity, like a face you learn to recognize. The first time you connect, you write that face down in a file named known_hosts. Every connection after that, SSH compares the face on the wire to the one on file. If they differ, something is off: the server was rebuilt, or someone is sitting between you and it, wearing a mask.

That someone-in-the-middle attack (called man-in-the-middle, or MITM, where an attacker quietly relays your traffic while reading and rewriting it) is the exact thing host keys exist to stop. So the single most important line in any SSH automation is your host-key policy: what your code does the moment it meets a key it has never seen before.

Paramiko (a popular pure-Python SSH library) gives you three choices. AutoAddPolicy records whatever key it is handed and connects, trusting a stranger on sight. WarningPolicy prints a warning and connects anyway. RejectPolicy refuses. In production you want RejectPolicy, backed by a known_hosts file you filled in on purpose. To fill it in safely, read the server's key yourself and check its fingerprint before you decide to trust it.

Compare that SHA256 value (a fingerprint from a standard hashing function, so a one-character change in the key changes the whole line) against what your server team published through a channel that isn't the connection you're trying to secure: a ticket, a config repo, the machine's console. Once it matches, pin it with ssh-keyscan -t ed25519 web01.prod.internal >> /etc/app/known_hosts. Now your code has a face to check against.

import paramiko
 
client = paramiko.SSHClient()
client.load_system_host_keys()                 # ~/.ssh/known_hosts (read-only)
client.load_host_keys("/etc/app/known_hosts") # your curated, pinned keys
 
# RejectPolicy: refuse any host key that isn't already trusted.
# Never AutoAddPolicy in production.
client.set_missing_host_key_policy(paramiko.RejectPolicy())
 
client.connect(
    "web01.prod.internal",
    username="deploy",
    key_filename="/etc/app/deploy_ed25519",
    timeout=15,         # TCP connect
    banner_timeout=15,  # server must send its SSH banner in time
    auth_timeout=15,    # authentication must finish in time
)
 
stdin, stdout, stderr = client.exec_command("systemctl is-active app", timeout=15)
rc = stdout.channel.recv_exit_status()   # wait for the command, read its exit code
out = stdout.read().decode().strip()
if rc != 0:
    raise SystemExit(f"web01: app not active (rc={rc}, said {out!r})")
client.close()

Point that script at a host whose key you have not pinned and it fails closed, before it runs a single remote command.

One check people skip. exec_command hands back three streams: stdin, stdout, and stderr. The remote command's exit code doesn't arrive until the command actually finishes, so you read it with stdout.channel.recv_exit_status and you test it. Skip that and a remote command that failed will look exactly like one that worked, and your script will report a healthy deploy over a broken server. The connect timeout covers the initial TCP (Transmission Control Protocol, the connection two machines set up before they exchange any data) handshake; banner_timeout and auth_timeout close the gap where a server accepts the socket but then stalls during the SSH greeting or the login.

HTTP clients that give up on time

An HTTP (HyperText Transfer Protocol, the request-and-response language the web runs on) call is a phone call to a machine that might not pick up, might leave you on hold forever, or might tell you it's slammed and to call back. Left on their defaults, plenty of HTTP libraries place that call with no hang-up timer at all. One unresponsive server and your job waits, and waits, and nothing pages a human about a process that is merely blocked. On a quiet Sunday that is a very long wait.

Four habits keep a client honest under load. Put a timeout on every request, and split it: a connect timeout for how long to wait to reach the server, and a read timeout for how long to wait for the answer once you're through. Retry only what's worth retrying: a 429 (Too Many Requests, the server asking you to slow down) or a 5xx (a server-side error in the 500 range) may clear on a second try, but a 400 or a 403 means your request is wrong and repeating it only repeats the mistake. When a 429 arrives, the server usually sends a Retry-After header with the number of seconds to wait; use its number instead of guessing. And reuse one client so your connections stay warm instead of paying the setup cost on every call.

import os, time, random, logging
import httpx
 
log = logging.getLogger("client")
RETRIABLE = {429, 500, 502, 503, 504}
 
def make_client() -> httpx.Client:
    token = os.environ["API_TOKEN"]            # from env / secrets manager, never source
    return httpx.Client(
        base_url="https://api.example.com",
        headers={"Authorization": f"Bearer {token}"},
        timeout=httpx.Timeout(10.0, connect=5.0),
        limits=httpx.Limits(max_connections=20, max_keepalive_connections=10),
    )
 
def get_json(c: httpx.Client, path: str, params=None, max_tries: int = 4):
    for attempt in range(1, max_tries + 1):
        r = c.get(path, params=params)
        if r.status_code not in RETRIABLE:
            r.raise_for_status()               # 4xx (except 429) fails fast, no retry
            return r.json()
        if attempt == max_tries:
            r.raise_for_status()
        retry_after = r.headers.get("Retry-After")
        delay = float(retry_after) if retry_after else min(2 ** attempt + random.random(), 30)
        log.warning("%s from %s, waiting %.1fs (try %d/%d)",
                    r.status_code, path, delay, attempt, max_tries)
        time.sleep(delay)

The token comes from the environment or a secrets manager, never from source, and the Authorization header never goes into a log line. The retry loop only sleeps on the statuses in RETRIABLE, honours Retry-After when it's present, and otherwise falls back to exponential backoff (each failed try waits roughly twice as long as the one before) with a little random jitter (a small random offset so a fleet of clients doesn't all retry on the same tick and stampede the server). Anything else, including a 404 or a 401, raises right away through raise_for_status.

None of this helps if it lives in your head instead of your code. So audit for the three footguns directly.

Two lines, two files, each pointing straight at the next fix. If that command prints nothing, a shell can't re-parse your arguments, no SSH client trusts a stranger's key, and no request can hang without limit. Wire the same grep into your continuous integration and it stays that way, because the day someone reintroduces one of these, the build tells you before the code ever reaches a server.

Try this

Work through “HTTP clients that give up on time” yourself on a sandbox you can throw away, following the commands above in order. Then break one step deliberately and re-run, so you have seen the failure before it finds you.

Takeaway

The trap worth remembering here: autoAddPolicy trusts the impostor. Check that on your own systems before you need to, because it is cheaper to find on a quiet afternoon than during an incident.

Related