diff --git a/services/network.py b/services/network.py index 4e3b57c..a610709 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 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)} diff --git a/tests/test_network.py b/tests/test_network.py index 3e828ed..4e8ea13 100644 --- a/tests/test_network.py +++ b/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="