diff --git a/services/network.py b/services/network.py index a610709..29472ef 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 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: 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 = `