Merge pull request 'fix: disable auto-redirects to prevent SSRF (#3)' (#15) from fix/issue-3 into main

Reviewed-on: https://git.example.com/jarianc/NASAImageDownloader/pulls/15

downloader.py

@@ -42,6 +42,7 @@ Dependencies
 
 import argparse
 import datetime
+import ipaddress
 import json
 import os
 import re
@@ -49,6 +50,7 @@ import sys
 import time
 from pathlib import Path
 from typing import Dict, List
+from urllib.parse import urlparse
 
 import requests
 
@@ -166,12 +168,42 @@ 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:
+        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):