From fdc66efe1b978a573a6bae54458534678ddf9f59 Mon Sep 17 00:00:00 2001
From: Jarian Cottingham <jarianc@proton.me>
Date: Sun, 5 Jul 2026 07:44:27 +0000
Subject: [PATCH] fix: disable auto-redirects and validate URLs to prevent SSRF
 (#3)

- Add _is_safe_url() checking for private/loopback/link-local/reserved IPs
- Block internal hostnames (.local, .internal, localhost, metadata.google.internal)
- Disable allow_redirects in requests.get()
- Validate redirect target before following single hop
- Raise RuntimeError on unsafe URL or redirect
---
 downloader.py | 42 +++++++++++++++++++++++++++++++++++++-----
 1 file changed, 37 insertions(+), 5 deletions(-)

diff --git a/downloader.py b/downloader.py
index 6cdf2e3..7ea6d4a 100644
--- a/downloader.py
+++ b/downloader.py
@@ -42,12 +42,14 @@ Dependencies
 
 import argparse
 import datetime
+import ipaddress
 import json
 import os
 import sys
 import time
 from pathlib import Path
 from typing import Dict, List
+from urllib.parse import urlparse
 
 import requests
 
@@ -165,16 +167,46 @@ def fetch_items(start: datetime.datetime, end: datetime.datetime) -> List[Dict]:
     return filtered
 
 
+def _is_safe_url(url: str) -> bool:
+    """Check that a URL points to a safe (non-internal) destination."""
+    parsed = urlparse(url)
+    host = parsed.hostname or ""
+    # Block private and reserved IP ranges
+    try:
+        addr = ipaddress.ip_address(host)
+        if addr.is_private or addr.is_loopback or addr.is_link_local or addr.is_reserved:
+            return False
+    except ValueError:
+        pass
+    # Block internal hostnames
+    if host.endswith((".local", ".internal", ".home.arpa")) or host in ("localhost", "metadata.google.internal"):
+        return False
+    return True
+
+
 def download_file(url: str, dest: Path) -> bool:
     """Stream‑download the file at *url* to *dest* if it does not already exist."""
     if dest.exists():
         return False
+    if not _is_safe_url(url):
+        raise RuntimeError(f"Unsafe download URL blocked (SSRF protection): {url}")
     try:
-        with requests.get(url, stream=True, timeout=30) as r:
-            r.raise_for_status()
-            with dest.open("wb") as f:
-                for chunk in r.iter_content(chunk_size=8192):
-                    f.write(chunk)
+        with requests.get(url, stream=True, timeout=30, allow_redirects=False) as r:
+            if r.status_code in (301, 302, 303, 307, 308):
+                location = r.headers.get("Location", "")
+                if not _is_safe_url(location):
+                    raise RuntimeError(f"Redirect to unsafe URL blocked (SSRF protection): {location}")
+                # Follow single validated redirect
+                with requests.get(location, stream=True, timeout=30, allow_redirects=False) as r2:
+                    r2.raise_for_status()
+                    with dest.open("wb") as f:
+                        for chunk in r2.iter_content(chunk_size=8192):
+                            f.write(chunk)
+            else:
+                r.raise_for_status()
+                with dest.open("wb") as f:
+                    for chunk in r.iter_content(chunk_size=8192):
+                        f.write(chunk)
         return True
     except Exception as e:
         raise RuntimeError(f"Download error for {url}: {e}") from e