Prevent DOM XSS in port-scan results

services/network.py

@@ -7,7 +7,7 @@
 import time
 import json
 from typing import Dict, List, Optional
-from urllib.parse import urljoin, urlparse
+from urllib.parse import urljoin, urlparse, urlsplit, urlunsplit
 
 import httpx
 from bs4 import BeautifulSoup
@@ -16,31 +16,74 @@
 
 
 
-def _is_public_host(host: str) -> bool:
-    """SSRF guard: refuse private/loopback/link-local/metadata hosts."""
+def _resolve_public_addresses(host: str, port: int) -> List[str]:
+    """Resolve a host and return its addresses only if all are globally routable."""
     if not host:
-        return False
-    # numeric IP?
+        return []
     try:
         addr = ipaddress.ip_address(host)
-        return addr.is_global
+        return [str(addr)] if addr.is_global else []
     except ValueError:
         pass
-    # resolve and check every address
     try:
-        infos = socket.getaddrinfo(host, None)
-    except socket.gaierror:
-        return False
+        infos = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)
+    except (socket.gaierror, UnicodeError):
+        return []
     if not infos:
-        return False
+        return []
+    addresses = []
     for info in infos:
         try:
             addr = ipaddress.ip_address(info[4][0])
         except ValueError:
-            return False
+            return []
         if not addr.is_global:
-            return False
-    return True
+            return []
+        addresses.append(str(addr))
+    return list(dict.fromkeys(addresses))
+
+
+def _prepare_public_request(url: str):
+    try:
+        parsed = urlsplit(url)
+        host = parsed.hostname
+        port = parsed.port
+    except ValueError:
+        return None
+
+    if (
+        parsed.scheme not in ("http", "https")
+        or not parsed.netloc
+        or not host
+        or parsed.username is not None
+        or parsed.password is not None
+    ):
+        return None
+
+    try:
+        host = host.encode("idna").decode("ascii")
+    except UnicodeError:
+        return None
+
+    if port is None:
+        port = 443 if parsed.scheme == "https" else 80
+    elif not 1 <= port <= 65535:
+        return None
+    addresses = _resolve_public_addresses(host, port)
+    if not addresses:
+        return None
+
+    address = ipaddress.ip_address(addresses[0])
+    request_host = f"[{address}]" if address.version == 6 else str(address)
+    host_header = f"[{host}]" if ":" in host else host
+    if parsed.port is not None:
+        request_host = f"{request_host}:{port}"
+        host_header = f"{host_header}:{port}"
+
+    request_url = urlunsplit(
+        (parsed.scheme, request_host, parsed.path or "/", parsed.query, "")
+    )
+    return request_url, host_header, host
 
 
 def _is_public_url(url: str) -> bool:
@@ -228,20 +271,30 @@ def _get_cert():
     async def web_scraper(self, url: str) -> Dict:
         try:
             # 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).
+            # pin each request to the address checked before connecting.
             cur = url
-            if not _is_public_url(cur):
-                return {"url": url, "error": "blocked: non-public host"}
             resp = None
             for _ in range(Config.MAX_REDIRECTS):
+                prepared = _prepare_public_request(cur)
+                if not prepared:
+                    error = (
+                        "blocked: redirect to non-public host"
+                        if cur != url
+                        else "blocked: non-public host"
+                    )
+                    return {"url": url, "error": error}
+                request_url, host_header, server_hostname = prepared
                 async with httpx.AsyncClient(
                     timeout=self.timeout,
                     follow_redirects=False,
                     headers={"User-Agent": Config.USER_AGENT},
+                    trust_env=False,
                 ) as client:
-                    resp = await client.get(cur)
+                    request = client.build_request(
+                        "GET", request_url, headers={"Host": host_header}
+                    )
+                    request.extensions["sni_hostname"] = server_hostname
+                    resp = await client.send(request)
                 if resp.status_code in (301, 302, 303, 307, 308):
                     loc = resp.headers.get("location")
                     if not loc:

static/app.js

@@ -157,11 +157,13 @@ async function runPortScan() {
     if (!target) return alert('Enter a target');
     showLoading('network-results');
     const data = await api(`/api/network/port-scan?target=${encodeURIComponent(target)}`);
-    let html = `<div class="section-title">Port Scan: ${target}</div>`;
+    let html = '<div class="section-title"></div>';
     html += `<div>Open: ${(data.open || []).map(p => `<span class="port-open">${p}</span>`).join(', ') || 'None'}</div>`;
     html += `<div>Closed: ${(data.closed || []).map(p => `<span class="port-closed">${p}</span>`).join(', ') || 'None'}</div>`;
     html += `<div>Filtered: ${(data.filtered || []).map(p => `<span class="port-filtered">${p}</span>`).join(', ') || 'None'}</div>`;
-    document.getElementById('network-results').innerHTML = html;
+    const results = document.getElementById('network-results');
+    results.innerHTML = html;
+    results.querySelector('.section-title').textContent = `Port Scan: ${target}`;
 }
 
 async function runWhois() {

tests/test_network.py

@@ -1,5 +1,6 @@
 import sys
 import os
+import socket
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 import pytest