From cff631e2cf3d14eee61bd04f05ab05bc8a6642c0 Mon Sep 17 00:00:00 2001
From: twojlogin <twojlogin@users.noreply.github.com>
Date: Mon, 19 Oct 2026 11:48:44 +0200
Subject: [PATCH] fix(security): SSRF guard in web_scraper (no
 private/loopback/metadata hosts, manual redirect validation)

---
 requirements.txt    |  1 +
 services/network.py | 66 ++++++++++++++++++++++++++++++++++++++++-----
 2 files changed, 61 insertions(+), 6 deletions(-)

diff --git a/requirements.txt b/requirements.txt
index 12e91ec..a253aa6 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -6,3 +6,4 @@ sqlalchemy>=2.0.0
 psutil>=5.9.0
 cryptography>=41.0.0
 lxml>=4.9.0
+pytest-asyncio>=0.23.0
diff --git a/services/network.py b/services/network.py
index 587e9fe..4e3b57c 100644
--- a/services/network.py
+++ b/services/network.py
@@ -1,6 +1,7 @@
 """MatrixCommand - Network Recon Service"""
 
 import asyncio
+import ipaddress
 import socket
 import ssl
 import time
@@ -14,6 +15,34 @@
 from config import Config
 
 
+
+def _is_public_host(host: str) -> bool:
+    """SSRF guard: refuse private/loopback/link-local/metadata hosts."""
+    if not host:
+        return False
+    # numeric IP?
+    try:
+        addr = ipaddress.ip_address(host)
+        return addr.is_global
+    except ValueError:
+        pass
+    # resolve and check every address
+    try:
+        infos = socket.getaddrinfo(host, None)
+    except socket.gaierror:
+        return False
+    if not infos:
+        return False
+    for info in infos:
+        try:
+            addr = ipaddress.ip_address(info[4][0])
+        except ValueError:
+            return False
+        if not addr.is_global:
+            return False
+    return True
+
+
 class NetworkRecon:
     def __init__(self):
         self.timeout = Config.REQUEST_TIMEOUT
@@ -187,12 +216,37 @@ def _get_cert():
 
     async def web_scraper(self, url: str) -> Dict:
         try:
-            async with httpx.AsyncClient(
-                timeout=self.timeout,
-                follow_redirects=True,
-                headers={"User-Agent": Config.USER_AGENT},
-            ) as client:
-                resp = await client.get(url)
+            # SSRF guard: never fetch private/loopback/metadata hosts, and
+            # 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):
+                return {"url": url, "error": "blocked: non-public host"}
+            resp = None
+            for _ in range(Config.MAX_REDIRECTS):
+                async with httpx.AsyncClient(
+                    timeout=self.timeout,
+                    follow_redirects=False,
+                    headers={"User-Agent": Config.USER_AGENT},
+                ) as client:
+                    resp = await client.get(cur)
+                if resp.status_code in (301, 302, 303, 307, 308):
+                    loc = resp.headers.get("location")
+                    if not loc:
+                        break
+                    nxt = urljoin(cur, loc)
+                    if not _is_public_host(urlparse(nxt).hostname or ""):
+                        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 ""