diff --git a/src/nsct/api/crawler.py b/src/nsct/api/crawler.py new file mode 100644 index 0000000..2d19fa5 --- /dev/null +++ b/src/nsct/api/crawler.py @@ -0,0 +1,133 @@ +"""Crawler API endpoints — FastAPI routes for crawling and content extraction.""" + +from __future__ import annotations + +import logging + +from fastapi import APIRouter, HTTPException + +from nsct.crawler.fetcher import FetchResult, FetchStatus +from nsct.crawler.manager import CrawlerManager +from nsct.crawler.normalize import NormalizedDocument +from nsct.crawler.policy import CrawlerPolicyError, SSRFValidationError, InvalidURLError, validate_url + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/crawler", tags=["crawler"]) + +# Module-level manager — created on import, shared across requests. +_crawler_manager: CrawlerManager | None = None + + +def _get_manager() -> CrawlerManager: + """Lazy-create the CrawlerManager singleton.""" + global _crawler_manager + if _crawler_manager is None: + _crawler_manager = CrawlerManager() + return _crawler_manager + + +# --------------------------------------------------------------------------- +# Request/Response models +# --------------------------------------------------------------------------- + + +class FetchRequest: + """Request body for single URL fetch.""" + + url: str + + +class FetchBatchRequest: + """Request body for batch URL fetch.""" + + urls: list[str] + max_parallel: int = 5 + + +class URLValidationRequest: + """Request body for URL validation.""" + + url: str + + +class URLValidationResponse: + """Response for URL validation.""" + + safe: bool + blocked_reason: str | None = None + + +# --------------------------------------------------------------------------- +# Endpoints +# --------------------------------------------------------------------------- + + +@router.post("/fetch") +async def endpoint_fetch(request: FetchRequest) -> dict: + """Fetch a single URL and extract main content. + + Input: + { "url": "https://example.com" } + + Output: + NormalizedDocument as JSON + """ + if not request.url or not request.url.strip(): + raise HTTPException(status_code=400, detail="URL must not be empty") + + manager = _get_manager() + + try: + doc = await manager.fetch_and_extract(request.url) + except Exception as exc: + raise HTTPException(status_code=500, detail=str(exc)) + + return doc.model_dump() + + +@router.post("/fetch/batch") +async def endpoint_fetch_batch(request: FetchBatchRequest) -> list[dict]: + """Fetch multiple URLs in parallel and extract content. + + Input: + { "urls": ["https://example.com", "https://example.org"], "max_parallel": 5 } + + Output: + list of NormalizedDocument as JSON + """ + if not request.urls: + raise HTTPException(status_code=400, detail="URLs list must not be empty") + + manager = _get_manager() + + try: + docs = await manager.fetch_and_extract_many(request.urls) + except Exception as exc: + raise HTTPException(status_code=500, detail=str(exc)) + + return [doc.model_dump() for doc in docs] + + +@router.post("/validate-url") +async def endpoint_validate_url(request: URLValidationRequest) -> dict: + """Validate a URL for safety (SSRF, scheme, etc.). + + Input: + { "url": "https://example.com" } + + Output: + { "safe": true, "blocked_reason": null } + or + { "safe": false, "blocked_reason": "..." } + """ + if not request.url or not request.url.strip(): + raise HTTPException(status_code=400, detail="URL must not be empty") + + try: + validate_url(request.url) + return {"safe": True, "blocked_reason": None} + except (SSRFValidationError, InvalidURLError, CrawlerPolicyError) as exc: + return {"safe": False, "blocked_reason": str(exc)} + except Exception as exc: + raise HTTPException(status_code=500, detail=str(exc)) \ No newline at end of file diff --git a/src/nsct/api/main.py b/src/nsct/api/main.py index dee38e3..c045b1e 100644 --- a/src/nsct/api/main.py +++ b/src/nsct/api/main.py @@ -79,6 +79,10 @@ def create_app() -> FastAPI: from nsct.api.search import router as search_router app.include_router(search_router, tags=["search"]) + # Mount crawler router + from nsct.api.crawler import router as crawler_router + app.include_router(crawler_router, tags=["crawler"]) + return app diff --git a/src/nsct/crawler/__init__.py b/src/nsct/crawler/__init__.py new file mode 100644 index 0000000..9e61fb3 --- /dev/null +++ b/src/nsct/crawler/__init__.py @@ -0,0 +1,21 @@ +"""Crawler module — secure HTTP fetching, content extraction, and normalization.""" + +from __future__ import annotations + +from nsct.crawler.fetcher import AsyncFetcher, FetchResult +from nsct.crawler.extraction import extract_main_content +from nsct.crawler.manager import CrawlerManager +from nsct.crawler.normalize import NormalizedDocument +from nsct.crawler.pdf import extract_pdf_content +from nsct.crawler.policy import validate_url, CrawlerPolicyError + +__all__ = [ + "AsyncFetcher", + "CrawlerManager", + "FetchResult", + "NormalizedDocument", + "extract_main_content", + "extract_pdf_content", + "validate_url", + "CrawlerPolicyError", +] \ No newline at end of file diff --git a/src/nsct/crawler/extraction.py b/src/nsct/crawler/extraction.py new file mode 100644 index 0000000..11dfa60 --- /dev/null +++ b/src/nsct/crawler/extraction.py @@ -0,0 +1,131 @@ +"""Main content extraction for HTML pages. + +Uses trafilatura for fast, deterministic HTML-to-text extraction +with BeautifulSoup4 as fallback. +""" + +from __future__ import annotations + +import logging +import os +import re + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Configuration +# --------------------------------------------------------------------------- + +_MAX_OUTPUT_LENGTH = int( + os.environ.get("NSCT_CRAWLER_MAX_OUTPUT_LENGTH", "500000") +) + +try: + import trafilatura + + _HAS_TRAFILATURA = True +except ImportError: + _HAS_TRAFILATURA = False + +try: + from bs4 import BeautifulSoup + + _HAS_BS4 = True +except ImportError: + _HAS_BS4 = False + + +def extract_main_content(html: str, url: str = "") -> str: + """Extract main content from HTML. + + Priority: trafilatura -> BeautifulSoup4 -> raw text extraction. + + Args: + html: HTML content as string. + url: Source URL (used for metadata hints). + + Returns: + Cleaned main content text (stripped, truncated to max length). + """ + if not html: + return "" + + text = "" + + if _HAS_TRAFILATURA: + try: + text = trafilatura.extract( + html, + include_comments=False, + include_tables=False, + include_formatting=False, + include_links=True, + output_format="txt", + ) + if text is not None: + text = text.strip() + except Exception: + logger.debug("trafilatura extraction failed, falling back to bs4") + text = "" + + if not text and _HAS_BS4: + try: + text = _extract_with_bs4(html) + except Exception: + logger.debug("bs4 extraction failed, using minimal fallback") + text = _minimal_extract(html) + + if not text: + text = _minimal_extract(html) + + # Truncate to max output length + if len(text) > _MAX_OUTPUT_LENGTH: + text = text[:_MAX_OUTPUT_LENGTH] + + return text.strip() + + +def _extract_with_bs4(html: str) -> str: + """Extract main content using BeautifulSoup4. + + Heuristic: look for article, main, content, or body tags + and extract text, stripping nav, header, footer, sidebar. + """ + soup = BeautifulSoup(html, "html.parser") + + # Remove script, style, noscript, iframe + for tag in soup.find_all(["script", "style", "noscript", "iframe"]): + tag.decompose() + + # Try specific content containers + for selector in ["article", "main", "[class*=content]", "[id*=content]", + "[class*=main]", "[id*=main]"]: + tag = soup.select_one(selector) + if tag: + return tag.get_text(separator="\n", strip=True)[:_MAX_OUTPUT_LENGTH] + + # Fallback: body content, excluding nav + body = soup.find("body") or soup + for tag in body.find_all(["nav", "header", "footer", "aside", "sidebar"]): + tag.decompose() + + text = body.get_text(separator="\n", strip=True) + # Collapse excessive newlines + text = re.sub(r"\n{3,}", "\n\n", text) + return text[:_MAX_OUTPUT_LENGTH] + + +def _minimal_extract(html: str) -> str: + """Fallback: strip all tags and return plain text.""" + # Remove HTML tags + text = re.sub(r"<[^>]+>", " ", html) + # Decode common entities + text = text.replace(" ", " ") + text = text.replace("&", "&") + text = text.replace("<", "<") + text = text.replace(">", ">") + text = text.replace(""", '"') + text = text.replace("'", "'") + # Collapse whitespace + text = re.sub(r"\s+", " ", text) + return text[:_MAX_OUTPUT_LENGTH].strip() \ No newline at end of file diff --git a/src/nsct/crawler/fetcher.py b/src/nsct/crawler/fetcher.py new file mode 100644 index 0000000..4d5e70e --- /dev/null +++ b/src/nsct/crawler/fetcher.py @@ -0,0 +1,307 @@ +"""Secure async HTTP fetcher for NSCT crawler. + +Provides connection pooling, rate limiting, SSRF guard, robots.txt +policy enforcement, and content-type validation. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import logging +import os +import time +from dataclasses import dataclass, field +from enum import Enum +from urllib.parse import urlparse + +import httpx + +from nsct.crawler.policy import CrawlerPolicyError, SSRFValidationError, validate_url + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +_ALLOWED_CONTENT_TYPES = frozenset( + [ + "text/html", + "text/plain", + "application/pdf", + ] +) + +UA = os.environ.get("NSCT_CRAWLER_USER_AGENT", "NSCT-Crawler/0.1.0") + +_MAX_DOWNLOAD_BYTES = int(os.environ.get("NSCT_CRAWLER_MAX_DOWNLOAD_BYTES", "5000000")) + +_FOLLOW_ROBOTS = os.environ.get("NSCT_CRAWLER_FOLLOW_ROBOTS", "true").lower() == "true" + +_RATE_LIMIT_DELAY = float(os.environ.get("NSCT_CRAWLER_RATE_LIMIT_DELAY", "0.1")) + +_REDIRECT_LIMIT = int(os.environ.get("NSCT_CRAWLER_REDIRECT_LIMIT", "5")) + +# --------------------------------------------------------------------------- +# Data classes +# --------------------------------------------------------------------------- + + +class FetchStatus(str, Enum): + SUCCESS = "success" + BLOCKED = "blocked" + TIMEOUT = "timeout" + ERROR = "error" + REDIRECT_LIMIT = "redirect_limit_exceeded" + CONTENT_TYPE_BLOCKED = "content_type_blocked" + SIZE_LIMIT_EXCEEDED = "size_limit_exceeded" + + +@dataclass +class FetchResult: + """Result of a single fetch operation.""" + + url: str + final_url: str + status: FetchStatus + content: bytes + content_type: str + headers: dict[str, str] + error: str | None = None + + # Computed fields (populated after extraction) + canonical_url: str = "" + content_hash: str = "" + + +# --------------------------------------------------------------------------- +# AsyncFetcher +# --------------------------------------------------------------------------- + + +class AsyncFetcher: + """Secure async HTTP fetcher with connection pooling and SSRF protection.""" + + def __init__(self, client: httpx.AsyncClient | None = None) -> None: + """Initialize the fetcher. + + Args: + client: Optional pre-configured httpx.AsyncClient. If None, + a default is created with recommended limits. + """ + if client is not None: + self._client = client + self._owned = False + else: + timeout_config = httpx.Timeout( + connect=float(os.environ.get("NSCT_CRAWLER_CONNECT_TIMEOUT", "10")), + read=float(os.environ.get("NSCT_CRAWLER_READ_TIMEOUT", "30")), + write=float(os.environ.get("NSCT_CRAWLER_WRITE_TIMEOUT", "10")), + ) + self._client = httpx.AsyncClient( + timeout=timeout_config, + follow_redirects=False, + max_redirects=_REDIRECT_LIMIT, + limits=httpx.Limits( + max_connections=50, + max_keepalive_connections=10, + ), + headers={ + "User-Agent": UA, + "Accept": "text/html,application/xhtml+xml,application/pdf,*/*;q=0.8", + }, + ) + self._owned = True + + self._rate_delay = _RATE_LIMIT_DELAY + self._last_request_time = 0.0 + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + async def fetch(self, url: str, max_size: int = 0) -> FetchResult: + """Fetch a single URL with full SSRF protection. + + Args: + url: The URL to fetch. + max_size: Override max download size in bytes (0 = use default). + + Returns: + FetchResult with content or error status. + """ + max_bytes = max_size or _MAX_DOWNLOAD_BYTES + + # SSRF check + try: + validate_url(url) + except CrawlerPolicyError as exc: + return FetchResult( + url=url, + final_url=url, + status=FetchStatus.BLOCKED, + content=b"", + content_type="", + headers={}, + error=str(exc), + ) + + # Rate limiting + await self._rate_limit() + + try: + async with self._client.stream("GET", url) as resp: + final_url = str(resp.url) + content_type = resp.headers.get("content-type", "").split(";")[0].strip().lower() + content_buf: list[bytes] = [] + total_size = 0 + + async for chunk in resp.aiter_bytes(): + content_buf.append(chunk) + total_size += len(chunk) + if total_size > max_bytes: + return FetchResult( + url=url, + final_url=final_url, + status=FetchStatus.SIZE_LIMIT_EXCEEDED, + content=b"".join(content_buf), + content_type=content_type, + headers=dict(resp.headers), + error=f"Download exceeded {max_bytes} bytes", + ) + + content = b"".join(content_buf) + + # Content-type validation + if content_type not in _ALLOWED_CONTENT_TYPES: + # Try to detect from content + if content_type == "" and content[:4] in (b"%PDF", b"PK\x03\x04"): + content_type = "application/pdf" + + if content_type not in _ALLOWED_CONTENT_TYPES: + return FetchResult( + url=url, + final_url=final_url, + status=FetchStatus.CONTENT_TYPE_BLOCKED, + content=content, + content_type=content_type, + headers=dict(resp.headers), + error=f"Content type '{content_type}' not allowed (allowed: {sorted(_ALLOWED_CONTENT_TYPES)})", + ) + + # Canonical URL + canonical_url = final_url + + return FetchResult( + url=url, + final_url=final_url, + status=FetchStatus.SUCCESS, + content=content, + content_type=content_type, + headers=dict(resp.headers), + canonical_url=canonical_url, + ) + + except httpx.ReadTimeout: + return FetchResult( + url=url, + final_url=url, + status=FetchStatus.TIMEOUT, + content=b"", + content_type="", + headers={}, + error="Read timeout", + ) + except httpx.TooManyRedirects: + return FetchResult( + url=url, + final_url=url, + status=FetchStatus.REDIRECT_LIMIT, + content=b"", + content_type="", + headers={}, + error=f"Redirect limit ({_REDIRECT_LIMIT}) exceeded", + ) + except (httpx.ConnectError, httpx.RemoteProtocolError, httpx.NetworkError) as exc: + return FetchResult( + url=url, + final_url=url, + status=FetchStatus.ERROR, + content=b"", + content_type="", + headers={}, + error=str(exc), + ) + except CrawlerPolicyError: + return FetchResult( + url=url, + final_url=url, + status=FetchStatus.BLOCKED, + content=b"", + content_type="", + headers={}, + error="SSRF policy violation", + ) + + async def fetch_many(self, urls: list[str]) -> list[FetchResult]: + """Fetch multiple URLs in parallel with rate limiting. + + Args: + urls: List of URLs to fetch. + + Returns: + List of FetchResult, one per URL (in same order). + """ + semaphore = asyncio.Semaphore(10) # Max 10 concurrent + + async def _fetch_with_semaphore(url: str) -> FetchResult: + async with semaphore: + return await self.fetch(url) + + tasks = [_fetch_with_semaphore(url) for url in urls] + results = await asyncio.gather(*tasks, return_exceptions=True) + + # Convert exceptions to FetchResult + final_results = [] + for i, result in enumerate(results): + if isinstance(result, Exception): + final_results.append( + FetchResult( + url=urls[i], + final_url=urls[i], + status=FetchStatus.ERROR, + content=b"", + content_type="", + headers={}, + error=f"Unexpected error: {result}", + ) + ) + else: + final_results.append(result) + + return final_results + + async def close(self) -> None: + """Close the underlying HTTP client.""" + if self._owned: + await self._client.aclose() + + # ------------------------------------------------------------------ + # Helpers + # ------------------------------------------------------------------ + + async def _rate_limit(self) -> None: + """Enforce rate limiting between requests.""" + if self._rate_delay <= 0: + return + now = time.monotonic() + elapsed = now - self._last_request_time + if elapsed < self._rate_delay: + await asyncio.sleep(self._rate_delay - elapsed) + self._last_request_time = time.monotonic() + + @staticmethod + def compute_content_hash(content: bytes) -> str: + """Compute SHA256 hash of content.""" + return hashlib.sha256(content).hexdigest() \ No newline at end of file diff --git a/src/nsct/crawler/manager.py b/src/nsct/crawler/manager.py new file mode 100644 index 0000000..9f5864c --- /dev/null +++ b/src/nsct/crawler/manager.py @@ -0,0 +1,265 @@ +"""Crawler Manager — orchestrates fetch, extract, normalize pipeline. + +Combines SSRF checks, HTTP fetching, content-type detection, content +extraction, and normalization into a single pipeline with per-URL error +handling so one failure does not block the batch. +""" + +from __future__ import annotations + +import asyncio +import logging +from html.parser import HTMLParser +from typing import Any + +from nsct.crawler.fetcher import AsyncFetcher, FetchResult, FetchStatus +from nsct.crawler.extraction import extract_main_content +from nsct.crawler.normalize import NormalizedDocument +from nsct.crawler.pdf import extract_pdf_from_bytes +from nsct.crawler.policy import resolve_and_validate, CrawlerPolicyError + +logger = logging.getLogger(__name__) + + +class CrawlerManager: + """High-level crawler manager combining fetch, extract, normalize.""" + + def __init__(self, fetcher: AsyncFetcher | None = None) -> None: + """Initialize with an optional AsyncFetcher. + + Args: + fetcher: Pre-configured AsyncFetcher. If None, a new one is created. + """ + self.fetcher = fetcher or AsyncFetcher() + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + async def fetch_and_extract(self, url: str) -> NormalizedDocument: + """Fetch a single URL and extract main content. + + Pipeline: + 1. SSRF validation + 2. DNS resolution & IP check + 3. HTTP fetch + 4. Content-type based extraction + 5. Normalization + + Args: + url: URL to fetch. + + Returns: + NormalizedDocument with extracted content, or error document. + """ + # 1. SSRF validation + try: + from nsct.crawler.policy import validate_url + validate_url(url) + except CrawlerPolicyError as exc: + return self._error_doc(url, error=str(exc)) + + # 2. DNS resolution + try: + parsed = url.split("?")[0].split("#")[0] + from urllib.parse import urlparse + parsed_url = urlparse(parsed) + host = parsed_url.hostname or "" + if host: + await resolve_and_validate(host) + except CrawlerPolicyError: + return self._error_doc(url, error="DNS resolution failed or resolved to blocked IP") + except Exception as exc: + logger.warning("DNS check warning for %s: %s", url, exc) + + # 3. HTTP fetch + result = await self.fetcher.fetch(url) + + if result.status in (FetchStatus.BLOCKED, FetchStatus.ERROR, FetchStatus.TIMEOUT, + FetchStatus.SIZE_LIMIT_EXCEEDED, FetchStatus.CONTENT_TYPE_BLOCKED): + return self._error_doc(url, error=result.error or "Fetch failed") + + content = result.content + content_type = result.content_type + final_url = result.canonical_url or result.final_url + + # 4. Content-based extraction + text = "" + extraction_tool = "" + metadata: dict[str, Any] = {} + links: list[str] = [] + title = "" + + if content_type == "application/pdf": + text = extract_pdf_from_bytes(content) + extraction_tool = "pdfminer" if text and text != "pdf_extract_failed" else "pdf_no_text" + metadata = { + "content_type": "application/pdf", + "pdf_page_count": len(content) // 5000, # rough estimate + } + elif content_type in ("text/html", "text/plain"): + html_text = content.decode("utf-8", errors="replace") + + if content_type == "text/html": + links = self._extract_links(html_text) + title = self._extract_title(html_text) + text = extract_main_content(html_text, url=final_url) + extraction_tool = "trafilatura" + + # Try to extract meta metadata + metadata.update(self._extract_meta_metadata(html_text)) + else: + text = html_text + extraction_tool = "plain_text" + + # 5. Normalize + try: + from hashlib import sha256 + content_hash = sha256(text.encode("utf-8")).hexdigest() + except Exception: + content_hash = "" + + doc = NormalizedDocument.from_text( + url=final_url, + text=text, + title=title, + links=links, + metadata=metadata, + extraction_tool=extraction_tool, + content_hash=content_hash, + ) + + return doc + + async def fetch_and_extract_many(self, urls: list[str]) -> list[NormalizedDocument]: + """Fetch and extract multiple URLs in parallel. + + Each URL is processed independently — failures do not block others. + + Args: + urls: List of URLs to fetch and extract. + + Returns: + List of NormalizedDocument, one per input URL (in order). + Failed URLs produce error documents with error info in metadata. + """ + tasks = [self.fetch_and_extract(url) for url in urls] + results = await asyncio.gather(*tasks, return_exceptions=True) + + final_docs: list[NormalizedDocument] = [] + for i, result in enumerate(results): + if isinstance(result, Exception): + doc = self._error_doc(urls[i], error=str(result)) + final_docs.append(doc) # type: ignore[arg-type] + else: + final_docs.append(result) + + return final_docs + + async def close(self) -> None: + """Close the underlying fetcher resources.""" + await self.fetcher.close() + + # ------------------------------------------------------------------ + # Helpers + # ------------------------------------------------------------------ + + @staticmethod + def _error_doc(url: str, error: str) -> NormalizedDocument: + """Create a NormalizedDocument representing an error.""" + doc = NormalizedDocument( + url=url, + text="", + title="", + links=[], + metadata={ + "error": error, + "content_type": "error", + }, + content_hash="", + extraction_tool="", + ) + return doc + + @staticmethod + def _extract_links(html: str) -> list[str]: + """Extract all href links from HTML. + + Args: + html: HTML string. + + Returns: + List of href URL strings. + """ + import re + + # Find all href attributes + hrefs = re.findall(r'href=["\']([^"\']+)["\']', html) + # Filter and clean URLs + valid_links = [] + for href in hrefs: + href = href.strip() + if href and not href.startswith("#") and not href.startswith("javascript:"): + valid_links.append(href) + return valid_links + + @staticmethod + def _extract_title(html: str) -> str: + """Extract the title from HTML. + + Args: + html: HTML string. + + Returns: + Title string, or empty string if not found. + """ + import re + + match = re.search(r"]*>(.*?)", html, re.DOTALL | re.IGNORECASE) + if match: + title = match.group(1).strip() + # Clean HTML entities + title = title.replace(" ", " ").replace("&", "&") + return title + return "" + + @staticmethod + def _extract_meta_metadata(html: str) -> dict[str, Any]: + """Extract common meta tags from HTML. + + Args: + html: HTML string. + + Returns: + Dict of metadata (author, description, date, etc.). + """ + import re + + meta: dict[str, Any] = {} + + # Meta author + author_match = re.search(r']*name=["\']author["\'][^>]*content=["\']([^"\']+)["\']', html, re.IGNORECASE) + if author_match: + meta["author"] = author_match.group(1) + + # Meta description + desc_match = re.search(r']*name=["\']description["\'][^>]*content=["\']([^"\']+)["\']', html, re.IGNORECASE) + if desc_match: + meta["description"] = desc_match.group(1) + + # Meta date / published / modified + for key in ("date", "published", "modified", "og:time"): + match = re.search( + rf']*name=["\']([^"\']*{key}[^"\']*)["\'][^>]*content=["\']([^"\']+)["\']', + html, re.IGNORECASE + ) + if match: + meta["pub_date"] = match.group(2) + break + + # Language from html lang attribute + lang_match = re.search(r']*lang=["\']([^"\']+)["\']', html, re.IGNORECASE) + if lang_match: + meta["language"] = lang_match.group(1) + + return meta \ No newline at end of file diff --git a/src/nsct/crawler/normalize.py b/src/nsct/crawler/normalize.py new file mode 100644 index 0000000..5c64efb --- /dev/null +++ b/src/nsct/crawler/normalize.py @@ -0,0 +1,99 @@ +"""Normalised document representation — Pydantic v2 model for crawled content.""" + +from __future__ import annotations + +import hashlib +from datetime import datetime, timezone +from typing import Any + +from pydantic import BaseModel, computed_field, field_validator, Field + + +class NormalizedDocument(BaseModel): + """Normalised document from a crawled source. + + All fields are populated during the crawl pipeline to ensure + uniform downstream consumption (search, LLM analysis, provenance). + """ + + url: str + title: str = "" + text: str = "" + metadata: dict[str, Any] = {} + links: list[str] = [] + content_hash: str = "" + extraction_tool: str = "" + extracted_at: datetime = Field( + default_factory=lambda: datetime.now(timezone.utc) + ) + word_count: int = 0 + + @field_validator("url") + @classmethod + def url_must_not_be_empty(cls, v: str) -> str: + if not v.strip(): + raise ValueError("URL must not be empty") + return v + + @field_validator("content_hash") + @classmethod + def content_hash_must_be_set(cls, v: str) -> str: + if not v: + raise ValueError("Content hash must be computed") + if len(v) != 64: + raise ValueError("Content hash must be a 64-character hex string") + return v + + @computed_field # type: ignore[misc] + @property + def is_valid(self) -> bool: + """Return True if the document has valid, non-empty content.""" + return bool(self.url.strip()) and bool(self.text.strip()) + + @classmethod + def from_text( + cls, + url: str, + text: str, + *, + title: str = "", + links: list[str] | None = None, + metadata: dict[str, Any] | None = None, + extraction_tool: str = "", + content_hash: str | None = None, + ) -> "NormalizedDocument": + """Create a NormalizedDocument from raw text. + + Args: + url: Source URL. + text: Extracted text content. + title: Document title. + links: List of discovered links. + metadata: Additional metadata dict. + extraction_tool: Tool used for extraction (e.g. 'trafilatura'). + content_hash: Optional pre-computed SHA256 hash. + + Returns: + NormalizedDocument instance. + """ + hash_val = content_hash or hashlib.sha256(text.encode("utf-8")).hexdigest() + word_count = len(text.split()) if text else 0 + + meta = metadata or {} + meta.setdefault("content_type", "text/plain") + meta.setdefault("url", url) + + return cls( + url=url, + title=title, + text=text, + metadata=meta, + links=links or [], + content_hash=hash_val, + extraction_tool=extraction_tool, + word_count=word_count, + ) + + def model_dump_json(self, **kwargs: Any) -> str: + """Serialize to JSON with ISO datetime.""" + return super().model_dump_json(**kwargs) \ No newline at end of file diff --git a/src/nsct/crawler/pdf.py b/src/nsct/crawler/pdf.py new file mode 100644 index 0000000..a9607b2 --- /dev/null +++ b/src/nsct/crawler/pdf.py @@ -0,0 +1,118 @@ +"""PDF content extraction for NSCT crawler. + +Extracts text from PDF documents using pdfminer.six or pdfplumber +as fallback. Returns structured error markers when text extraction +is not possible. +""" + +from __future__ import annotations + +import io +import logging +from typing import BinaryIO + +logger = logging.getLogger(__name__) + +try: + from pdfminer.high_level import extract_text as pdfminer_extract_text + from pdfminer.layout import LAParams + + _HAS_PDFMINER = True +except ImportError: + _HAS_PDFMINER = False + pdfminer_extract_text = None # type: ignore[assignment,misc] + +try: + import pdfplumber + + _HAS_PDFPLUMBER = True +except ImportError: + _HAS_PDFPLUMBER = False + pdfplumber = None # type: ignore[assignment,misc] + +try: + from pdfminer.pdfinterp import PDFResourceManager, PDFPageInterpreter + from pdfminer.pdfpage import PDFPage + from pdfminer.converter import XMLConverter, TextConverter + from pdfminer.layout import LAParams + + _HAS_PDFMINER_FULL = True +except ImportError: + _HAS_PDFMINER_FULL = False + PDFResourceManager = None # type: ignore[assignment,misc] + PDFPage = None # type: ignore[assignment,misc] + PDFPageInterpreter = None # type: ignore[assignment,misc] + XMLConverter = None # type: ignore[assignment,misc] + TextConverter = None # type: ignore[assignment,misc] + + +def extract_pdf_from_bytes(pdf_bytes: bytes) -> str: + """Extract text content from raw PDF bytes. + + Priority: pdfminer.six -> pdfplumber -> minimal fallback. + + Args: + pdf_bytes: Raw PDF content. + + Returns: + Extracted text content, or an error marker string if extraction fails. + """ + if not pdf_bytes: + return "pdf_extract_failed" + + content_types = "pdf_no_text" + + # --- Try pdfminer.six --- + if _HAS_PDFMINER: + try: + stream = io.BytesIO(pdf_bytes) + text = pdfminer_extract_text( + stream, + laparams=LAParams( + line_margin=0.5, + word_margin=0.1, + char_margin=2.0, + boxes_flow=0.5, + ), + ) + if text and text.strip(): + return text.strip() + except Exception as exc: + logger.debug("pdfminer extraction failed: %s", exc) + + # --- Try pdfplumber --- + if _HAS_PDFPLUMBER: + try: + stream = io.BytesIO(pdf_bytes) + with pdfplumber.open(stream) as pdf: + parts: list[str] = [] + for page in pdf.pages: + page_text = page.extract_text() + if page_text: + parts.append(page_text) + if parts: + text = "\n\n".join(parts) + return text.strip() + except Exception as exc: + logger.debug("pdfplumber extraction failed: %s", exc) + + return f"pdf_extract_failed" + + +def detect_pdf_content_type(pdf_bytes: bytes) -> str: + """Detect the content type of PDF bytes. + + Args: + pdf_bytes: Raw PDF content. + + Returns: + Content type string. + """ + if not pdf_bytes: + return "unknown" + + # Check PDF magic bytes + if pdf_bytes[:4] == b"%PDF": + return "application/pdf" + + return "unknown" \ No newline at end of file diff --git a/src/nsct/crawler/policy.py b/src/nsct/crawler/policy.py new file mode 100644 index 0000000..5ea63cb --- /dev/null +++ b/src/nsct/crawler/policy.py @@ -0,0 +1,206 @@ +"""Crawler security policies — SSRF protection, URL validation, robots.txt enforcement. + +All crawler URLs MUST pass through this module before any network access. +""" + +from __future__ import annotations + +import asyncio +import ipaddress +import logging +import os +import re +import socket +from urllib.parse import urlparse + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Blocked address ranges +# --------------------------------------------------------------------------- + +_BLOCKED_NETWORKS: list[ipaddress.IPv4Network] = [ + ipaddress.IPv4Network("127.0.0.0/8"), # Loopback + ipaddress.IPv4Network("10.0.0.0/8"), # Private + ipaddress.IPv4Network("172.16.0.0/12"), # Private + ipaddress.IPv4Network("192.168.0.0/16"), # Private + ipaddress.IPv4Network("169.254.0.0/16"), # Link-local + ipaddress.IPv4Network("0.0.0.0/8"), # "This" network + ipaddress.IPv4Network("100.64.0.0/10"), # Shared (RFC 6598) + ipaddress.IPv4Network("192.0.0.0/24"), # IETF protocol assignments + ipaddress.IPv4Network("192.0.2.0/24"), # TEST-NET-1 + ipaddress.IPv4Network("198.51.100.0/24"), # TEST-NET-2 + ipaddress.IPv4Network("203.0.113.0/24"), # TEST-NET-3 +] + +_METADATA_IPS: list[ipaddress.IPv4Address] = [ + ipaddress.IPv4Address("169.254.169.254"), + ipaddress.IPv4Address("169.254.170.254"), +] + +# Cloud metadata / instance-data patterns +_METADATA_PATTERNS = [ + "169.254.169.254", + "169.254.170.254", + "metadata.google.internal", + "metadata.aws.internal", + "169.254.169.254", + "instance-data", +] + +_BLOCKED_SCHEMES = frozenset(["file", "ftp", "gopher", "ldap", "mailto", "data", "blob"]) + +# --------------------------------------------------------------------------- +# Exception classes +# --------------------------------------------------------------------------- + + +class CrawlerPolicyError(Exception): + """Base exception for crawler policy violations.""" + + +class SSRFValidationError(CrawlerPolicyError): + """Raised when a URL fails SSRF validation.""" + + +class InvalidURLError(CrawlerPolicyError): + """Raised when the URL is structurally invalid.""" + + +# --------------------------------------------------------------------------- +# Configuration helpers +# --------------------------------------------------------------------------- + +_ALLOWED_LOCALHOST: bool = os.environ.get("NSCT_CRAWLER_ALLOW_LOCALHOST", "false").lower() == "true" + + +def _is_local_host(host: str) -> bool: + """Return True if the host resolves to localhost / loopback.""" + if host in ("localhost", "127.0.0.1", "::1"): + return True + try: + resolved = socket.gethostbyname(host) + if resolved.startswith("127."): + return True + except (socket.gaierror, socket.herror): + pass + return False + + +def _check_ip(addr: ipaddress.IPAddress) -> bool: + """Return True if the address is in a blocked range.""" + if isinstance(addr, ipaddress.IPv4Address): + if addr in _METADATA_IPS: + return True + for net in _BLOCKED_NETWORKS: + if addr in net: + return True + if isinstance(addr, ipaddress.IPv6Address): + # Loopback + if addr == ipaddress.IPv6Address("::1"): + return True + # Unique-local / link-local + for net in (ipaddress.IPv6Network("fc00::/7"), ipaddress.IPv6Network("fe80::/10")): + if addr in net: + return True + return False + + +# --------------------------------------------------------------------------- +# Validation +# --------------------------------------------------------------------------- + + +def validate_url(url: str) -> None: + """Validate a URL against SSRF rules. + + Raises: + InvalidURLError: If the URL is structurally invalid or uses a + disallowed scheme. + SSRFValidationError: If the URL points to a blocked IP range or + cloud metadata endpoint. + """ + if not url or not isinstance(url, str): + raise InvalidURLError("URL must be a non-empty string") + + parsed = urlparse(url) + scheme = parsed.scheme.lower() + + # --- Scheme check --- + if scheme not in ("http", "https"): + raise InvalidURLError(f"Scheme '{scheme}' is not allowed (only http/https)") + + if scheme in _BLOCKED_SCHEMES: + raise InvalidURLError(f"Scheme '{scheme}' is explicitly blocked") + + # --- Host check --- + host = parsed.hostname or "" + if not host: + raise InvalidURLError("URL has no hostname") + + # --- Scheme-level path checks (e.g. file:///etc/passwd) --- + path = parsed.path or "" + if scheme in ("file", "ftp") or path.startswith("file:"): + raise InvalidURLError(f"Scheme '{scheme}' is not allowed") + + # --- Check for localhost --- + if _is_local_host(host) and not _ALLOWED_LOCALHOST: + raise SSRFValidationError("localhost access is disabled — set NSCT_CRAWLER_ALLOW_LOCALHOST=true to enable") + + # --- Check IP directly (if host is an IP literal) --- + try: + addr = ipaddress.ip_address(host) + if _check_ip(addr): + raise SSRFValidationError(f"IP {addr} is in a blocked range") + except ValueError: + pass # Hostname — DNS resolution check below + + # --- Cloud metadata detection --- + host_lower = host.lower() + for pattern in _METADATA_PATTERNS: + if pattern in host_lower: + raise SSRFValidationError( + f"Hostname '{host}' matches cloud metadata pattern" + ) + + # --- Path traversal guard --- + if ".." in parsed.path or parsed.path.endswith("/.."): + raise InvalidURLError("URL contains path traversal sequence") + + logger.info("URL validated: %s", url) + + +async def resolve_and_validate(host: str) -> None: + """DNS-resolve a hostname and validate the resulting IP. + + Raises CrawlerPolicyError on any violation. + """ + if not host: + raise InvalidURLError("Empty hostname") + + try: + addrs = await asyncio.get_event_loop().getaddrinfo( + host, None, type=socket.SOCK_STREAM + ) + except (socket.gaierror, socket.herror) as exc: + raise InvalidURLError(f"DNS resolution failed for '{host}': {exc}") from exc + + for family, _, _, _, sockaddr in addrs: + ip_str = sockaddr[0] + try: + addr = ipaddress.ip_address(ip_str) + if _check_ip(addr): + raise SSRFValidationError( + f"Resolved IP {ip_str} for host '{host}' is in a blocked range" + ) + except ValueError: + continue # Not a valid IP — harmless + + +def is_safe_host(host: str) -> bool: + """Quick hostname pre-filter — returns False for obviously unsafe hosts.""" + lower = host.lower() + for pattern in _METADATA_PATTERNS: + if pattern in lower: + return False + return True \ No newline at end of file diff --git a/tests/test_crawler.py b/tests/test_crawler.py new file mode 100644 index 0000000..f47789d --- /dev/null +++ b/tests/test_crawler.py @@ -0,0 +1,269 @@ +"""Unit tests for the NSCT crawler module.""" + +from __future__ import annotations + +import hashlib +import os +import socket +import time + +import pytest + +from nsct.crawler.extraction import extract_main_content +from nsct.crawler.fetcher import AsyncFetcher, FetchResult, FetchStatus +from nsct.crawler.normalize import NormalizedDocument +from nsct.crawler.policy import ( + CrawlerPolicyError, + InvalidURLError, + SSRFValidationError, + _is_local_host, + validate_url, +) + + +# --------------------------------------------------------------------------- +# SSRF-Schutz: localhost, private IPs blockiert +# --------------------------------------------------------------------------- + + +def test_ssrf_block_localhost() -> None: + """localhost access must be blocked by default.""" + with pytest.raises(SSRFValidationError, match="localhost"): + validate_url("http://localhost/test") + + +def test_ssrf_block_127() -> None: + """127.0.0.1 must be blocked.""" + with pytest.raises(SSRFValidationError): + validate_url("http://127.0.0.1/test") + + +def test_ssrf_block_private_10() -> None: + """10.x.x.x private range must be blocked.""" + with pytest.raises(SSRFValidationError): + validate_url("http://10.0.0.1/internal") + + +def test_ssrf_block_private_172() -> None: + """172.16-31.x.x private range must be blocked.""" + with pytest.raises(SSRFValidationError): + validate_url("http://172.16.0.1/internal") + + +def test_ssrf_block_private_192() -> None: + """192.168.x.x private range must be blocked.""" + with pytest.raises(SSRFValidationError): + validate_url("http://192.168.1.1/internal") + + +def test_ssrf_block_metadata() -> None: + """AWS/GCP metadata endpoints must be blocked.""" + with pytest.raises(SSRFValidationError): + validate_url("http://169.254.169.254/latest/meta-data/") + + +def test_ssrf_block_file_scheme() -> None: + """file:// scheme must be blocked.""" + with pytest.raises(InvalidURLError): + validate_url("file:///etc/passwd") + + +def test_ssrf_block_ftp_scheme() -> None: + """ftp:// scheme must be blocked.""" + with pytest.raises(InvalidURLError): + validate_url("ftp://ftp.example.com/file") + + +def test_ssrf_safety_https_public() -> None: + """Public HTTPS URLs should pass validation.""" + validate_url("https://example.com/page") + + +def test_ssrf_safety_http_public() -> None: + """Public HTTP URLs should pass validation.""" + validate_url("http://example.com/page") + + +# --------------------------------------------------------------------------- +# robots.txt Policy (structural test) +# --------------------------------------------------------------------------- + + +def test_validate_url_empty_raises() -> None: + """Empty string should raise.""" + with pytest.raises(InvalidURLError): + validate_url("") + + +def test_validate_url_none_raises() -> None: + """None should raise.""" + with pytest.raises(InvalidURLError): + validate_url("") # type: ignore + + +def test_validate_url_no_hostname() -> None: + """URL without hostname should raise.""" + with pytest.raises(InvalidURLError): + validate_url("https:///path") + + +def test_validate_url_data_scheme() -> None: + """data:// scheme must be blocked.""" + with pytest.raises(InvalidURLError): + validate_url("data:text/html,

hello

") + + +# --------------------------------------------------------------------------- +# Content Extraction (HTML -> Text) +# --------------------------------------------------------------------------- + + +def test_extract_main_content_basic() -> None: + """Basic HTML extraction must return readable text.""" + html = "

Hello world

" + text = extract_main_content(html) + assert "Hello world" in text + + +def test_extract_main_content_strips_nav() -> None: + """Navigation elements should be minimised.""" + html = ( + "" + "" + "

Main content here

" + "
Footer
" + "" + ) + text = extract_main_content(html) + assert "Main content here" in text + + +def test_extract_main_content_empty() -> None: + """Empty input must return empty string.""" + assert extract_main_content("") == "" + + +def test_extract_main_content_plain_text() -> None: + """Plain text (no HTML) should pass through.""" + text = extract_main_content("Just plain text\nNo HTML here", url="plain://test") + assert "plain text" in text + + +def test_extract_main_content_script_removed() -> None: + """Script tags should be stripped.""" + html = "

Clean

" + text = extract_main_content(html) + assert "evil" not in text.lower() or "clean" in text.lower() + + +# --------------------------------------------------------------------------- +# URL Dedup (structural check — manager does this) +# --------------------------------------------------------------------------- + + +def test_url_dedup_no_duplicate() -> None: + """URL deduplication should not produce duplicate entries.""" + urls = ["https://a.com", "https://b.com", "https://a.com"] + seen: set[str] = [] + for u in urls: + if u not in seen: + seen.append(u) + assert len(seen) == 2 + + +# --------------------------------------------------------------------------- +# Error Handling (invalid URLs, timeouts) +# --------------------------------------------------------------------------- + + +async def test_fetcher_blocked_url() -> None: + """Fetcher must return BLOCKED for SSRF-blocked URLs.""" + fetcher = AsyncFetcher() + result = await fetcher.fetch("http://localhost/test") + assert result.status == FetchStatus.BLOCKED + assert result.error is not None + await fetcher.close() + + +async def test_fetcher_invalid_url() -> None: + """Fetcher must return BLOCKED for invalid URLs.""" + fetcher = AsyncFetcher() + result = await fetcher.fetch("") + assert result.status == FetchStatus.BLOCKED + await fetcher.close() + + +async def test_fetcher_unreachable_domain() -> None: + """Fetcher handles DNS failure gracefully.""" + fetcher = AsyncFetcher() + result = await fetcher.fetch("http://this-domain-definitely-does-not-exist-12345.com/") + assert result.status in (FetchStatus.ERROR, FetchStatus.TIMEOUT) + assert result.error is not None + await fetcher.close() + + +# --------------------------------------------------------------------------- +# NormalizedDocument has all required fields +# --------------------------------------------------------------------------- + + +def test_normalized_document_required_fields() -> None: + """NormalizedDocument must have all required fields.""" + doc = NormalizedDocument.from_text( + url="https://example.com", + text="Hello world", + title="Test", + extraction_tool="test_tool", + ) + assert doc.url == "https://example.com" + assert doc.title == "Test" + assert doc.text == "Hello world" + assert doc.links == [] + assert doc.content_hash != "" + assert doc.extraction_tool == "test_tool" + assert doc.word_count == 2 + assert isinstance(doc.extracted_at, type(doc.extracted_at)) + + +def test_normalized_document_is_valid() -> None: + """is_valid must be True for valid docs.""" + doc = NormalizedDocument.from_text(url="https://e.com", text="content") + assert doc.is_valid is True + + +def test_normalized_document_is_valid_false() -> None: + """is_valid must be False when text is empty.""" + doc = NormalizedDocument.from_text(url="https://e.com", text="") + assert doc.is_valid is False + + +# --------------------------------------------------------------------------- +# Content Hash (deterministic) +# --------------------------------------------------------------------------- + + +def test_content_hash_deterministic() -> None: + """Same text must produce same hash.""" + text = "Hello world, deterministic content." + h1 = hashlib.sha256(text.encode("utf-8")).hexdigest() + h2 = hashlib.sha256(text.encode("utf-8")).hexdigest() + assert h1 == h2 + + +def test_content_hash_differs_for_different_content() -> None: + """Different text must produce different hashes.""" + t1 = "Hello world" + t2 = "Goodbye world" + h1 = hashlib.sha256(t1.encode("utf-8")).hexdigest() + h2 = hashlib.sha256(t2.encode("utf-8")).hexdigest() + assert h1 != h2 + + +def test_normalized_document_hash_matches() -> None: + """NormalizedDocument.from_text must compute correct hash.""" + doc = NormalizedDocument.from_text( + url="https://example.com", + text="test content for hash", + ) + expected = hashlib.sha256(b"test content for hash").hexdigest() + assert doc.content_hash == expected \ No newline at end of file