backend / security / 09_ssrf_command_injection.md

SSRF and Command Injection

6 interview angles 6 min read source

SSRF and Command Injection

Two of the most common Python web app vulnerabilities. Both are OWASP Top 10 (SSRF since 2021). Both have shaped real high-profile breaches.

SSRF — Server-Side Request Forgery

Your server makes an HTTP request based on user-controlled input. The attacker controls the URL → makes your server hit internal resources they couldn’t reach themselves.

The classic SSRF

@app.get("/fetch-thumbnail")
async def fetch(url: str):
    async with httpx.AsyncClient() as c:
        r = await c.get(url)
    return Response(r.content, media_type=r.headers["content-type"])

User provides ?url=http://example.com/foo.png — fine, returns the image.

User provides ?url=http://169.254.169.254/latest/meta-data/iam/security-credentials/. That’s the AWS EC2 metadata service. Your server hits it from inside the VPC; returns IAM credentials.

User provides ?url=http://internal-admin.svc.cluster.local/admin/users. Reaches the internal admin API.

User provides ?url=file:///etc/passwd. Reads local files (if the HTTP client honors file://).

Famous example: Capital One 2019

WAF was misconfigured + SSRF in their image-processing service → request to EC2 metadata → IAM credentials → S3 bucket access → 100M customer records.

Defense layers

1. Use IMDSv2 (mandatory)

AWS EC2 Instance Metadata Service v2 requires a token obtained via PUT, not GET. SSRF via simple GET can’t get to v2. Enforce v2-only across your AWS account:

aws ec2 modify-instance-metadata-options \
  --instance-id i-... \
  --http-tokens required \
  --http-endpoint enabled

In Terraform / CloudFormation, set metadata_options { http_tokens = "required" } everywhere. This is the single biggest SSRF defense for AWS workloads.

2. Validate the URL — DNS rebinding-safe

import ipaddress
from urllib.parse import urlparse
import socket

PRIVATE_RANGES = [
    ipaddress.ip_network("10.0.0.0/8"),
    ipaddress.ip_network("172.16.0.0/12"),
    ipaddress.ip_network("192.168.0.0/16"),
    ipaddress.ip_network("169.254.0.0/16"),    # link-local + IMDS
    ipaddress.ip_network("127.0.0.0/8"),
    ipaddress.ip_network("::1/128"),
    ipaddress.ip_network("fc00::/7"),
]

def is_private(ip):
    addr = ipaddress.ip_address(ip)
    return any(addr in net for net in PRIVATE_RANGES)

def safe_url(url):
    parsed = urlparse(url)
    if parsed.scheme not in ("http", "https"):
        return False
    if not parsed.hostname:
        return False
    # Resolve the hostname — and watch every address
    try:
        addrs = socket.getaddrinfo(parsed.hostname, None)
    except socket.gaierror:
        return False
    for addr in addrs:
        if is_private(addr[4][0]):
            return False
    return True

DNS rebinding is the gotcha: attacker controls a DNS name that resolves to a public IP at validation time, then a private IP at fetch time. Solutions:

  • Resolve once, pass the resolved IP to the HTTP client (skip DNS at fetch).
  • Use a connection-aware allowlist that re-validates at connect time.

httpx supports custom transports where you can validate the resolved IP before letting the connection proceed:

class SafeTransport(httpx.AsyncHTTPTransport):
    async def handle_async_request(self, request):
        # request.url.host already resolved by httpx?  No.
        # We must resolve and check.
        host = request.url.host
        addrs = socket.getaddrinfo(host, None)
        if any(is_private(a[4][0]) for a in addrs):
            raise httpx.RequestError("Blocked private IP")
        return await super().handle_async_request(request)

A more robust approach: a dedicated outbound HTTP service in a separate network with no access to internal IPs. Your main service calls it; it relays the actual request. The proxy can’t reach private space at the network level.

3. Allowlist over denylist

If your use case is “fetch thumbnails from a specific image hosting service”, allowlist the hostnames:

ALLOWED_HOSTS = {"cdn.example.com", "images.example.com"}

def safe_host(host):
    return host in ALLOWED_HOSTS

Allowlists are airtight; denylists always miss something (127.0.1, IPv6 representations, etc.).

4. Strip dangerous schemes

if parsed.scheme not in ("http", "https"):
    return False

Block file://, gopher://, ftp://, dict://. Most HTTP libraries follow them if allowed.

Command injection

Concatenating user input into shell commands:

import os
filename = request.args.get("filename")
os.system(f"rm /tmp/{filename}")             # filename = "foo; cat /etc/passwd"

Or subprocess.run(... shell=True):

subprocess.run(f"convert {filename} out.png", shell=True)  # same problem

Defense: never use shell=True with user input

# UNSAFE
subprocess.run(f"convert {filename} out.png", shell=True)

# SAFE — list form, no shell
subprocess.run(["convert", filename, "out.png"], check=True)

The list form passes args directly to execve; no shell interpretation. filename = "foo; cat /etc/passwd" becomes a single weird filename argument, not a command chain.

shlex.quote isn’t enough

subprocess.run(f"convert {shlex.quote(filename)} out.png", shell=True)

This works for simple cases but is fragile — a single missed escape, a shell version with different quoting rules, and you’re back to RCE. Use the list form instead.

Don’t shell out for what stdlib does

# Bad
subprocess.run(["rm", "-rf", path])

# Good
import shutil
shutil.rmtree(path)

Less ceremony, no shell, no command-line argument parsing bugs.

Other patterns

# os.system: SHELL
os.system(f"cp {src} {dst}")                  # unsafe

# subprocess + shell=False: NO SHELL
subprocess.run(["cp", src, dst])              # safe

# os.popen: SHELL
os.popen(f"ls {path}")                        # unsafe

# pathlib.Path operations: NO SHELL
Path(src).rename(dst)                         # safe

XML attacks (XXE — XML External Entity)

While we’re on the deserialization theme — XML parsers may evaluate external entities by default:

<?xml version="1.0"?>
<!DOCTYPE foo [<!ENTITY xxe SYSTEM "file:///etc/passwd">]>
<root>&xxe;</root>

Parsing this with lxml.etree.parse (without explicit XMLParser(resolve_entities=False)) reads /etc/passwd.

Defense:

# defusedxml — safe drop-in replacement
from defusedxml import ElementTree as ET
tree = ET.parse("file.xml")

defusedxml disables external entity loading by default. If you must use stock lxml:

from lxml import etree
parser = etree.XMLParser(resolve_entities=False, no_network=True)
tree = etree.parse("file.xml", parser)

Path traversal

@app.get("/files/{name}")
def get_file(name: str):
    with open(f"./uploads/{name}") as f:    # name = "../../etc/passwd"
        return f.read()

Defenses:

from pathlib import Path

UPLOADS = Path("./uploads").resolve()

def safe_path(name):
    p = (UPLOADS / name).resolve()
    if not p.is_relative_to(UPLOADS):
        raise HTTPException(400, "Bad path")
    return p

resolve() normalizes .. etc. is_relative_to (Python 3.9+) confirms the resolved path stays inside the allowed root. Pre-3.9: os.path.commonpath([UPLOADS, p]) == str(UPLOADS).

For S3 / object storage: store paths in a DB, never let users pick keys directly. Users provide an ID; you look up the key.

CSP and SSRF defense in depth

A Content Security Policy can mitigate certain SSRF-derived data exfiltration patterns (e.g., your image-fetch service returning credentials that get embedded into a page). CSP isn’t a primary defense against SSRF, but it can blunt the impact in browser contexts.

Interview angle

  • “What’s SSRF?” — Server-Side Request Forgery. Your server fetches a URL based on user input; attacker supplies a URL pointing at internal resources (EC2 metadata, internal services, localhost). Famous example: Capital One 2019 via SSRF → IAM credentials → 100M records.
  • “How do you defend against SSRF on AWS?” — first: enforce IMDSv2 (token-required metadata service) account-wide. Then: validate URLs against an allowlist, resolve DNS before fetching to a checked IP, run outbound requests through a dedicated proxy service with no internal network access.
  • “What’s DNS rebinding and how does it bypass SSRF defenses?” — attacker controls a DNS name. At validation, it resolves to a public IP. At fetch, the same name resolves to a private IP (because TTL=0 or attacker-controlled DNS). Defense: resolve once and pass the IP to the HTTP client; or re-validate at TCP connect time.
  • “What’s the safe way to call an external command from Python?” — subprocess.run([...], shell=False) with a list of args (no shell interpretation). Never use shell=True with user input. Prefer stdlib helpers (shutil.rmtree, Path.rename) over shelling out.
  • “What’s XXE and how do you prevent it?” — XML External Entity. Stock XML parsers may resolve <!ENTITY xxe SYSTEM "file://..."> and embed file contents. Use defusedxml or pass resolve_entities=False, no_network=True to lxml. JSON / Protobuf don’t have this class of issue.
  • “How do you prevent path traversal in a file-serving endpoint?” — resolve the requested path against your allowed root; reject if it’s not contained. Path(name).resolve() + is_relative_to(root). Or, better: don’t take filenames from users at all — use IDs and look up the file path internally.
Previous
Pickle, YAML, and Deserialization Attacks (Python-Specific)