From fddc0370777e05ff814694b5fdd88c4f8cf75f54 Mon Sep 17 00:00:00 2001 From: "github-agentic-autofix[bot]" <314318003+GitHub Agentic Autofix@users.noreply.github.com> Date: Wed, 28 Oct 2026 05:41:55 +0000 Subject: [PATCH 1/3] Initial plan From db5e2096de80a5f49f6505f8890ad1e6b0d58a59 Mon Sep 17 00:00:00 2001 From: "github-agentic-autofix[bot]" <314318003+GitHub Agentic Autofix@users.noreply.github.com> Date: Wed, 28 Oct 2026 05:43:13 +0000 Subject: [PATCH 2/3] Prevent XSS in port scan target display Co-authored-by: technoporada <171401859+technoporada@users.noreply.github.com> --- static/app.js | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/static/app.js b/static/app.js index e025bb9..6384b7b 100644 --- a/static/app.js +++ b/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 = `
Port Scan: ${target}
`; + let html = '
'; html += `
Open: ${(data.open || []).map(p => `${p}`).join(', ') || 'None'}
`; html += `
Closed: ${(data.closed || []).map(p => `${p}`).join(', ') || 'None'}
`; html += `
Filtered: ${(data.filtered || []).map(p => `${p}`).join(', ') || 'None'}
`; - 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() { From 0a3c3efe83f4eca0c4af9d29cd82e488fa78c571 Mon Sep 17 00:00:00 2001 From: "github-agentic-autofix[bot]" <314318003+GitHub Agentic Autofix@users.noreply.github.com> Date: Wed, 28 Oct 2026 05:44:55 +0000 Subject: [PATCH 3/3] Pin scraper requests to validated public IPs Co-authored-by: technoporada <171401859+technoporada@users.noreply.github.com> --- services/network.py | 181 +++++++++++++++++++++++++++--------------- tests/test_network.py | 74 ++++++++++++++++- 2 files changed, 188 insertions(+), 67 deletions(-) diff --git a/services/network.py b/services/network.py index 4e3b57c..d320e5e 100644 --- a/services/network.py +++ b/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, 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 class NetworkRecon: @@ -217,76 +260,82 @@ 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). - from urllib.parse import urljoin, urlparse + # pin each request to the address checked before connecting. 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): + 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: 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 "" - 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)} diff --git a/tests/test_network.py b/tests/test_network.py index 3e828ed..656a4fa 100644 --- a/tests/test_network.py +++ b/tests/test_network.py @@ -1,10 +1,12 @@ import sys import os +import socket sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import pytest import asyncio -from services.network import NetworkRecon +import httpx +from services.network import NetworkRecon, _prepare_public_request @pytest.fixture @@ -38,3 +40,73 @@ def test_tech_signatures_categories(network): assert "CMS" in categories assert "Framework" in categories assert "Server" in categories + + +def test_prepare_public_request_pins_resolved_address(monkeypatch): + def getaddrinfo(host, port, type): + assert host == "example.com" + assert port == 443 + assert type == socket.SOCK_STREAM + return [ + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("8.8.8.8", port)) + ] + + monkeypatch.setattr("services.network.socket.getaddrinfo", getaddrinfo) + + prepared = _prepare_public_request("https://example.com/path?q=1") + + assert prepared == ( + "https://8.8.8.8/path?q=1", + "example.com", + "example.com", + ) + + +def test_prepare_public_request_rejects_unsafe_urls(monkeypatch): + assert _prepare_public_request("http://127.0.0.1/") is None + assert _prepare_public_request("file:///etc/passwd") is None + assert _prepare_public_request("https://[invalid/") is None + + monkeypatch.setattr( + "services.network.socket.getaddrinfo", + lambda host, port, type: [ + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("8.8.8.8", port)), + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", port)), + ], + ) + assert _prepare_public_request("https://example.com/") is None + + +@pytest.mark.asyncio +async def test_web_scraper_sends_pinned_request(monkeypatch): + sent_requests = [] + + class FakeClient: + def __init__(self, **kwargs): + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, *args): + pass + + def build_request(self, method, url, headers): + return httpx.Request(method, url, headers=headers) + + async def send(self, request): + sent_requests.append(request) + return httpx.Response(200, text="Example", request=request) + + monkeypatch.setattr( + "services.network._prepare_public_request", + lambda url: ("https://8.8.8.8/path", "example.com", "example.com"), + ) + monkeypatch.setattr("services.network.httpx.AsyncClient", FakeClient) + + result = await NetworkRecon().web_scraper("https://example.com/path") + + assert result["title"] == "Example" + assert str(sent_requests[0].url) == "https://8.8.8.8/path" + assert sent_requests[0].headers["host"] == "example.com" + assert sent_requests[0].extensions["sni_hostname"] == "example.com"