Secure a Download Endpoint Against SSRF

Steps

1. Build the naive endpoint — and attack it

   Start by building the vulnerable version, so you can watch it fail.
   Create an endpoint that accepts a file_url and downloads it:

   POST /downloads
   body: { "file_url": "https://example.com/report.pdf" }
   
   def create(file_url):
       response = http.get(file_url)          # <-- naive, follows anything
       saved = storage.write(response.body)
       return { id: saved.id }
   

   Now be the attacker. Point file_url at targets that should be off-limits
   and confirm the naive endpoint happily reaches them:

   # 1) Cloud metadata — the crown jewel
   file_url = "http://169.254.169.254/latest/meta-data/iam/security-credentials/"
   
   # 2) Internal service on the box
   file_url = "http://127.0.0.1:6379/"        # Redis
   
   # 3) Local file read via non-HTTP scheme
   file_url = "file:///etc/passwd"
   

   Observe that the naive handler:

   • follows any scheme its HTTP client supports,
   • resolves and connects to any IP, including internal ones,
   • follows redirects blindly,
   • and may echo the upstream body or error straight back to the caller.

   Write down, for each attack above, what the endpoint returned. That's your
   "before" evidence. The next five steps close each hole. Do not deploy this
   version — it exists only to demonstrate the attack in a controlled setting.

2. Require HTTPS and validate the URL

   The first gate is the URL itself. Parse it with a real URL parser (never
   with string matching / regex on the raw input) and enforce a strict shape.

   ALLOWED_SCHEMES = { "https" }
   
   def validate_url(raw):
       url = parse(raw)                       # real parser; reject on parse error
       if url.scheme not in ALLOWED_SCHEMES:  # blocks http, file, gopher, ftp, data...
           reject("scheme not allowed")
       if url.host is empty:
           reject("missing host")
       if url.userinfo present:               # e.g. https://user@evil@internal/
           reject("credentials in URL not allowed")
       return url
   

   Why each rule:

   • HTTPS-only kills file:// (local file read), gopher:// and ftp://
     (protocol smuggling), and plaintext http://. An allowlist of one scheme
     is far safer than a blocklist you have to keep complete.
   • Reject embedded credentials (userinfo): https://trusted.com@169.254.169.254/
     parses with host 169.254.169.254 in a correct parser, but sloppy code
     that reads up to the @ gets fooled. Rejecting userinfo removes the
     ambiguity entirely.
   • Require a non-empty host so relative or malformed inputs can't slip
     through.

   This step alone stops attacks #1's http:// variant and #3's file://
   entirely. But https://169.254.169.254/ is still valid HTTPS pointing at
   metadata — that's the next step's job.

   Do not try to validate the IP by inspecting the hostname string here.
   https://localhost/, https://0x7f000001/, and a domain whose DNS record
   points at 127.0.0.1 all look different as strings. IP filtering must
   happen on the resolved address (Step 3), not on the text.

3. Resolve DNS and block private/loopback/link-local IPs

   This is the core defense. Resolve the hostname to its actual IP addresses
   and refuse the fetch if any resolved address falls in a blocked range.

   BLOCKED_V4 = [
       "127.0.0.0/8", "10.0.0.0/8", "172.16.0.0/12",
       "192.168.0.0/16", "169.254.0.0/16", "0.0.0.0/8",
   ]
   BLOCKED_V6 = [ "::1/128", "fc00::/7", "fe80::/10" ]
   
   def assert_public_host(host):
       ips = dns_resolve(host)                # may return several A/AAAA records
       if ips is empty:
           reject("host does not resolve")
       for ip in ips:
           for cidr in BLOCKED_V4 + BLOCKED_V6:
               if ip in cidr:
                   reject("destination is a private/internal address")
       return ips
   

   Key details that make this correct:

   • Check every resolved IP, not just the first. A hostname can return
     multiple A/AAAA records; an attacker can list one public and one internal.
     If any is internal, reject.
   • Use a numeric IP-in-CIDR check, not string comparison. Parse the IP to
     its integer/bytes form and test containment. This is immune to
     0x7f.0.0.1, 2130706433 (decimal 127.0.0.1), and zero-padding tricks —
     they all parse to the same blocked address.
   • Cover IPv6 and IPv4-mapped IPv6 (::ffff:127.0.0.1). Normalize mapped
     addresses to their IPv4 form before checking, or block the mapped range too.

   Now https://169.254.169.254/ is refused, and so is any domain an attacker
   registers that resolves to 127.0.0.1 or 10.x.x.x.

   Advanced — DNS rebinding & TOCTOU. There's a gap between resolving the
   host and connecting: an attacker's DNS can return a public IP for the
   check, then a private IP microseconds later for the actual connection
   (a time-of-check/time-of-use race). The strongest fix is to resolve once,
   pick a validated IP, and connect to that exact IP while sending the
   original Host header — so the connection can't be re-pointed. Note this
   as a hardening goal; the per-hop revalidation in Step 4 also narrows the window.

4. Handle redirects with per-hop host revalidation

   A single validated URL isn't enough: https://evil.com/start can return a
   302 Location: http://169.254.169.254/. If your HTTP client auto-follows
   redirects, all your Step 2–3 work is bypassed on the second hop.

   Turn off automatic redirect following and drive it yourself, revalidating
   every hop:

   MAX_REDIRECTS = 2
   
   def fetch_guarded(url):
       for hop in 0..MAX_REDIRECTS:
           validate_url(url)                  # Step 2: https-only, no userinfo
           ips = assert_public_host(url.host) # Step 3: block private ranges
           resp = http.get(url, follow_redirects=false, connect_to=ips)
           if resp.is_redirect:
               url = resolve_relative(url, resp.header["Location"])
               continue                       # re-run validation on the NEW url
           return resp
       reject("too many redirects")
   

   The critical rule: every hop goes through the full Step 2 + Step 3 gauntlet
   again. A redirect target is just another user-influenced URL. Cap the hop
   count (2 is plenty for legitimate use) so a redirect loop can't spin forever.

   Common mistakes this prevents:

   • Trusting the client's built-in redirect handling (validates hop 0 only).
   • Validating the original host but connecting to the redirected one.
   • Allowing unlimited hops (Location bouncing as a slow-loris / DoS).

   With redirects revalidated, attack "public domain → 302 → metadata IP" now
   fails on the second hop exactly as a direct request would.

5. Enforce size, timeout, and content-type limits

   Even a fully validated public URL can be hostile: a huge file, a slow stream,
   or an unexpected type. Add resource limits so the fetch can't be turned into
   a denial-of-service or a type-confusion vector.

   MAX_BYTES   = 100 * 1024 * 1024            # 100 MB
   TIMEOUT     = 30_seconds                   # total, including connect + read
   ALLOWED_TYPES = { "application/pdf", "image/png", "image/jpeg" }
   
   def read_limited(resp):
       # 1) Trust nothing: check the header AND the actual stream
       if resp.header["Content-Length"] and int(it) > MAX_BYTES:
           reject("file too large")
   
       ctype = resp.header["Content-Type"].split(";")[0].strip().lower()
       if ctype not in ALLOWED_TYPES:
           reject("content-type not allowed")
   
       # 2) Stream and hard-cut — Content-Length can lie or be absent
       total = 0
       for chunk in resp.stream(TIMEOUT):
           total += len(chunk)
           if total > MAX_BYTES:
               abort_connection()
               reject("file too large")
           buffer.write(chunk)
       return buffer
   

   Why each limit:

   • Check Content-Length and cap the stream. The header is a hint an
     attacker controls — it can lie, be absent, or under-report. Counting bytes
     as you stream and aborting past the cap is the real defense. This also
     blunts decompression bombs (a small gzipped body that expands to
     gigabytes): cap the decompressed bytes.
   • Total timeout. A server that trickles one byte per second holds your
     worker hostage (slow-loris). Bound connect + read time.
   • Content-type allowlist. If you asked for an avatar image, refuse
     text/html or application/octet-stream. This narrows what a compromised
     upstream can feed into downstream parsers.

   Combined with the background-job rule, these limits keep one malicious URL
   from exhausting memory, threads, or time.

6. Prove it with network-mocked tests (submission)

   A security control you can't test will rot. Write tests that mock the
   network and assert each defense fails closed. Mocking lets you simulate
   dangerous responses (a redirect to metadata, an oversized body) without
   standing up real malicious infrastructure.

   Cover, at minimum, these cases — each should raise a rejection / return the
   generic error, never fetch the forbidden target:

   test "rejects non-https scheme":
       expect_reject(fetch_guarded("http://example.com/x"))
       expect_reject(fetch_guarded("file:///etc/passwd"))
   
   test "rejects host resolving to a private / link-local IP":
       stub_dns("evil.test" => ["169.254.169.254"])   # cloud metadata
       expect_reject(fetch_guarded("https://evil.test/"))
       stub_dns("lan.test"  => ["10.0.0.5"])
       expect_reject(fetch_guarded("https://lan.test/"))
   
   test "rejects a redirect that points at an internal IP":
       stub_http("https://ok.test/" => redirect_to("http://169.254.169.254/"))
       expect_reject(fetch_guarded("https://ok.test/"))
   
   test "rejects an oversized file (header and streamed)":
       stub_http("https://big.test/" => body_of(200 MB))
       expect_reject(fetch_guarded("https://big.test/"))
   
   test "rejects a disallowed content-type":
       stub_http("https://ok.test/" => html_response())
       expect_reject(fetch_guarded("https://ok.test/"))
   
   test "allows a well-formed public https file":
       stub_dns("cdn.test" => ["93.184.216.34"])       # public IP
       stub_http("https://cdn.test/a.pdf" => pdf_response(1 MB))
       expect_success(fetch_guarded("https://cdn.test/a.pdf"))
   
   test "never leaks the upstream error to the caller":
       stub_http("https://ok.test/" => connection_refused())
       resp = endpoint_download("https://ok.test/")
       expect(resp.body) == generic_error()            # no timing/error oracle
   

   Also assert the cross-cutting rules: the fetch runs in a background job
   (the web request returns immediately), and the redirect hop cap is enforced.

   ---

   Submission criteria

   Submit when all of the following hold:

   1. A /downloads-style endpoint that accepts file_url, enqueues a
      background job, and returns a generic response (no upstream detail).
   2. The guarded fetch enforces, in order: HTTPS-only + URL validation →
      DNS resolution with private/loopback/link-local blocking → per-hop
      redirect revalidation (cap ~2) → size + timeout + content-type limits.
   3. A passing test suite with network mocked, proving each block:
      non-HTTPS, private/link-local IP (including 169.254.169.254), malicious
      redirect, oversized file, disallowed content-type — plus one happy-path
      public fetch and one "error is not leaked" test.

   Include a short note on how you'd address DNS rebinding (connect-by-IP
   with a pinned Host) as a future hardening step.