a8595cc950
- AsyncFetcher: Sicheres HTTP-Fetching mit SSRF-Schutz, Connection Pooling (50/10), Timeout, Redirect Limit, Rate Limiting, User-Agent - Content-Extraction: trafilatura für HTML→Text, BeautifulSoup4 Fallback - PDF-Extraction: pdfminer.six mit Error-Handling - NormalizedDocument: Schema (url, title, text, metadata, links, content_hash, extraction_tool, extracted_at, word_count) - Crawler-Manager: SSRF-Check → robots.txt → HTTP-Fetch → Extraction → Normalization (Batch-fähig, Error-Isolation pro Fetch) - Security-Policy: SSRF-Schutz (RFC1918, Cloud Metadata, file://, ftp://, localhost), URL-Validation (nur http/https) - Crawler-Endpoints: POST /crawler/fetch, /crawler/fetch/batch, /crawler/validate-url - Test-Cases: SSRF-Schutz, Content Extraction, NormalizedDocument, Error Handling, Content Hash Determinismus
269 lines
8.5 KiB
Python
269 lines
8.5 KiB
Python
"""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,<h1>hello</h1>")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Content Extraction (HTML -> Text)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_extract_main_content_basic() -> None:
|
|
"""Basic HTML extraction must return readable text."""
|
|
html = "<html><body><p>Hello world</p></body></html>"
|
|
text = extract_main_content(html)
|
|
assert "Hello world" in text
|
|
|
|
|
|
def test_extract_main_content_strips_nav() -> None:
|
|
"""Navigation elements should be minimised."""
|
|
html = (
|
|
"<html><body>"
|
|
"<nav><a href='/a'>nav</a></nav>"
|
|
"<main><article><p>Main content here</p></article></main>"
|
|
"<footer>Footer</footer>"
|
|
"</body></html>"
|
|
)
|
|
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 = "<html><body><script>evil()</script><p>Clean</p></body></html>"
|
|
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 |