Harden scraper against SSRF

services/network.py

@@ -7,7 +7,7 @@
 import time
 import json
 from typing import Dict, List, Optional
-from urllib.parse import urlparse
+from urllib.parse import urljoin, urlparse
 
 import httpx
 from bs4 import BeautifulSoup
@@ -43,6 +43,17 @@ def _is_public_host(host: str) -> bool:
     return True
 
 
+def _is_public_url(url: str) -> bool:
+    try:
+        parsed = urlparse(url)
+        return (
+            parsed.scheme.lower() in ("http", "https")
+            and _is_public_host(parsed.hostname or "")
+        )
+    except ValueError:
+        return False
+
+
 class NetworkRecon:
     def __init__(self):
         self.timeout = Config.REQUEST_TIMEOUT
@@ -220,10 +231,8 @@ async def web_scraper(self, url: str) -> Dict:
             # never follow a redirect into one. Manual redirect loop so every
             # hop is re-validated (httpx follow_redirects=True would skip
             # the check on the second hop).
-            from urllib.parse import urljoin, urlparse
             cur = url
-            host0 = urlparse(cur).hostname or ""
-            if not _is_public_host(host0):
+            if not _is_public_url(cur):
                 return {"url": url, "error": "blocked: non-public host"}
             resp = None
             for _ in range(Config.MAX_REDIRECTS):
@@ -238,55 +247,55 @@ async def web_scraper(self, url: str) -> Dict:
                     if not loc:
                         break
                     nxt = urljoin(cur, loc)
-                    if not _is_public_host(urlparse(nxt).hostname or ""):
+                    if not _is_public_url(nxt):
                         return {"url": url, "error": "blocked: redirect to non-public host"}
                     cur = nxt
                     continue
                 break
             else:
                 return {"url": url, "error": "too many redirects"}
-                if resp is None:
-                    return {"url": url, "error": "fetch failed"}
-                soup = BeautifulSoup(resp.text, "html.parser")
-
-                title = soup.title.string.strip() if soup.title and soup.title.string else ""
-                meta_desc = ""
-                meta_tag = soup.find("meta", attrs={"name": "description"})
-                if meta_tag:
-                    meta_desc = meta_tag.get("content", "")
-
-                headings = []
-                for tag in ["h1", "h2", "h3"]:
-                    for h in soup.find_all(tag):
-                        text = h.get_text(strip=True)
-                        if text:
-                            headings.append({"tag": tag.upper(), "text": text[:200]})
-
-                links = []
-                for a in soup.find_all("a", href=True)[:50]:
-                    text = a.get_text(strip=True)[:100]
-                    href = a["href"]
-                    if text and href:
-                        links.append({"text": text, "href": href})
-
-                emails = set()
-                import re
-                for match in re.findall(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", resp.text):
-                    emails.add(match)
-
-                tech = self._detect_technologies(resp.text, dict(resp.headers))
-
-                return {
-                    "url": url,
-                    "status": resp.status_code,
-                    "title": title,
-                    "description": meta_desc,
-                    "headings": headings,
-                    "links": links,
-                    "emails": list(emails)[:20],
-                    "technologies": tech,
-                    "content_length": len(resp.text),
-                }
+            if resp is None:
+                return {"url": url, "error": "fetch failed"}
+            soup = BeautifulSoup(resp.text, "html.parser")
+
+            title = soup.title.string.strip() if soup.title and soup.title.string else ""
+            meta_desc = ""
+            meta_tag = soup.find("meta", attrs={"name": "description"})
+            if meta_tag:
+                meta_desc = meta_tag.get("content", "")
+
+            headings = []
+            for tag in ["h1", "h2", "h3"]:
+                for h in soup.find_all(tag):
+                    text = h.get_text(strip=True)
+                    if text:
+                        headings.append({"tag": tag.upper(), "text": text[:200]})
+
+            links = []
+            for a in soup.find_all("a", href=True)[:50]:
+                text = a.get_text(strip=True)[:100]
+                href = a["href"]
+                if text and href:
+                    links.append({"text": text, "href": href})
+
+            emails = set()
+            import re
+            for match in re.findall(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", resp.text):
+                emails.add(match)
+
+            tech = self._detect_technologies(resp.text, dict(resp.headers))
+
+            return {
+                "url": url,
+                "status": resp.status_code,
+                "title": title,
+                "description": meta_desc,
+                "headings": headings,
+                "links": links,
+                "emails": list(emails)[:20],
+                "technologies": tech,
+                "content_length": len(resp.text),
+            }
         except Exception as e:
             return {"url": url, "error": str(e)}
 

tests/test_network.py

@@ -4,6 +4,9 @@
 
 import pytest
 import asyncio
+from types import SimpleNamespace
+
+import services.network as network_service
 from services.network import NetworkRecon
 
 
@@ -38,3 +41,82 @@ def test_tech_signatures_categories(network):
     assert "CMS" in categories
     assert "Framework" in categories
     assert "Server" in categories
+
+
+@pytest.mark.asyncio
+async def test_web_scraper_blocks_non_http_and_private_urls(network, monkeypatch):
+    monkeypatch.setattr(network_service, "_is_public_host", lambda host: False)
+
+    class UnexpectedClient:
+        def __init__(self, *args, **kwargs):
+            raise AssertionError("blocked URL must not be fetched")
+
+    monkeypatch.setattr(network_service.httpx, "AsyncClient", UnexpectedClient)
+
+    for url in ("file:///etc/passwd", "http://127.0.0.1/"):
+        result = await network.web_scraper(url)
+        assert result["error"] == "blocked: non-public host"
+
+
+@pytest.mark.asyncio
+async def test_web_scraper_blocks_private_redirect(network, monkeypatch):
+    monkeypatch.setattr(
+        network_service,
+        "_is_public_host",
+        lambda host: host == "public.example",
+    )
+    fetched_urls = []
+
+    class FakeClient:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *args):
+            pass
+
+        async def get(self, url):
+            fetched_urls.append(url)
+            return SimpleNamespace(
+                status_code=302,
+                headers={"location": "http://127.0.0.1/"},
+            )
+
+    monkeypatch.setattr(network_service.httpx, "AsyncClient", FakeClient)
+
+    result = await network.web_scraper("https://public.example/")
+
+    assert result["error"] == "blocked: redirect to non-public host"
+    assert fetched_urls == ["https://public.example/"]
+
+
+@pytest.mark.asyncio
+async def test_web_scraper_returns_parsed_response(network, monkeypatch):
+    monkeypatch.setattr(network_service, "_is_public_host", lambda host: True)
+
+    class FakeClient:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *args):
+            pass
+
+        async def get(self, url):
+            return SimpleNamespace(
+                status_code=200,
+                headers={"server": "nginx"},
+                text="<html><title>Example</title><h1>Welcome</h1></html>",
+            )
+
+    monkeypatch.setattr(network_service.httpx, "AsyncClient", FakeClient)
+
+    result = await network.web_scraper("https://public.example/")
+
+    assert result["status"] == 200
+    assert result["title"] == "Example"
+    assert result["headings"] == [{"tag": "H1", "text": "Welcome"}]