feat: SSRF protection via NEXUS_ALLOWED_HOSTS, _env_int validation warnings

guarddog_nexus/config.py

@@ -1,7 +1,8 @@
 """Configuration via environment variables."""
 
 import os
-from dataclasses import dataclass
+from dataclasses import dataclass, field
+from urllib.parse import urlparse
 
 from guarddog_nexus.constants import (
     DEFAULT_MAX_CONCURRENT_SCANS,
@@ -21,13 +22,28 @@ def _env_int(name: str, default: int) -> int:
     try:
         return int(val)
     except ValueError:
+        import logging
+
+        logging.getLogger("guarddog_nexus").warning(
+            "Invalid value for %s=%r, using default %d", name, val, default
+        )
         return default
 
 
+def _resolve_allowed_hosts() -> list[str]:
+    raw = os.getenv("NEXUS_ALLOWED_HOSTS")
+    if raw:
+        return [h.strip() for h in raw.split(",") if h.strip()]
+    parsed = urlparse(os.getenv("NEXUS_URL", "http://localhost:8081"))
+    host = parsed.hostname or "localhost"
+    return [host]
+
+
 @dataclass
 class Config:
     # Nexus connection
     nexus_url: str = os.getenv("NEXUS_URL", "http://localhost:8081")
+    nexus_allowed_hosts: list[str] = field(default_factory=lambda: _resolve_allowed_hosts())
     nexus_download_timeout: int = _env_int("NEXUS_DOWNLOAD_TIMEOUT_SECONDS", HTTP_TIMEOUT_DOWNLOAD)
     nexus_api_timeout: int = _env_int("NEXUS_API_TIMEOUT_SECONDS", HTTP_TIMEOUT_API)
 

guarddog_nexus/core/nexus.py

@@ -3,7 +3,7 @@
 import asyncio
 import hashlib
 import os
-from urllib.parse import unquote
+from urllib.parse import unquote, urlparse
 
 import httpx
 
@@ -15,6 +15,15 @@ from ..constants import (
 from ..logging_setup import log
 
 
+def _validate_download_url(url: str) -> bool:
+    parsed = urlparse(url)
+    if parsed.scheme not in ("http", "https"):
+        return False
+    if parsed.hostname not in config.nexus_allowed_hosts:
+        return False
+    return True
+
+
 def extract_pypi_info(asset_path: str) -> tuple[str, str] | None:
     """Extract package name and version from a PyPI asset path.
 
@@ -101,6 +110,10 @@ def parse_package_path(path: str) -> tuple[str, str]:
 
 async def download_asset(download_url: str, dest_dir: str) -> str | None:
     """Download an asset from Nexus using async httpx."""
+    if not _validate_download_url(download_url):
+        log.warning("SSRF prevention: blocked download from %s", download_url)
+        return None
+
     dest_path = os.path.join(dest_dir, os.path.basename(download_url.split("?")[0]))
 
     async with httpx.AsyncClient(