Driving systems safely: subprocess, SSH & HTTP
subprocess without shell=True, paramiko/SSH, and resilient API clients.
Security automation is mostly Python talking to other systems — running local commands, driving remote hosts over SSH, and calling HTTP APIs. Each of those is a place to either be safe by construction or to open a hole. The through-line is the same as in Bash: never let untrusted data reach a shell, always bound the wait, and always check the result.
subprocess without a shell
Call external commands with an argument list and no shell=True, so nothing re-parses your arguments and injected metacharacters are inert. Set a timeout so a hung child cannot hang your job, capture output explicitly, and check the return code (check=True raises on failure). If you think you need shell=True for a pipe or glob, do the pipe in Python or pass shlex-quoted arguments — the shell is almost never actually required.
import subprocess# GOOD: list form, no shell, bounded, checked, capturedtry:r = subprocess.run(["kubectl", "get", "pods", "-n", namespace, "-o", "json"],capture_output=True, text=True, timeout=30, check=True,)data = json.loads(r.stdout)except subprocess.TimeoutExpired:raise SystemExit("kubectl timed out")except subprocess.CalledProcessError as e:raise SystemExit(f"kubectl failed ({e.returncode}): {e.stderr}")# BAD: never do this with any untrusted value in the string# subprocess.run(f"kubectl get pods -n {namespace}", shell=True)
SSH automation with Paramiko
For remote work, a real SSH library beats shelling out to the ssh binary: you control host-key verification, auth, and timeouts programmatically. The critical security control is host-key policy — never auto-add unknown keys in production (that disables the very protection SSH provides against man-in-the-middle). Load a known_hosts file and reject anything not in it.
import paramikoclient = paramiko.SSHClient()client.load_system_host_keys()client.load_host_keys("/etc/app/known_hosts")# RejectPolicy: fail on an unknown host key. NEVER AutoAddPolicy in prod.client.set_missing_host_key_policy(paramiko.RejectPolicy())client.connect(host, username="deploy", key_filename=key, timeout=15)stdin, stdout, stderr = client.exec_command("systemctl is-active app", timeout=15)rc = stdout.channel.recv_exit_status() # always check the remote rcif rc != 0:raise SystemExit(f"{host}: service not active ({rc})")client.close()
Resilient HTTP clients
API automation needs the same discipline as any network call: an explicit timeout on every request (the default is often no timeout at all), retries only on transient statuses, respect for Retry-After on 429s, and a reused session/connection pool for throughput. Pull secrets for auth from the environment or a secrets manager, never from source, and never log the Authorization header.
import httpx, osdef client() -> httpx.Client:token = os.environ["API_TOKEN"] # from env / secrets managerreturn 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),)def get_json(c, path, params=None):r = c.get(path, params=params)if r.status_code == 429: # honour rate limitingraise httpx.ConnectError(f"rate limited, retry after {r.headers.get('Retry-After')}")r.raise_for_status()return r.json()