"""Network source adapter for deepweb/clearnet content ingestion. Fetches content from approved URLs, extracts text, and registers it in the knowledge plane. All network access goes through the EgressGate (fail-closed). """ from __future__ import annotations import hashlib import ipaddress import mimetypes import re from dataclasses import dataclass, field from typing import Any from urllib.parse import urlsplit import httpx from cyberbrain26.knowledge_plane.contracts import ( DocumentRegistration, DocumentTrust, KnowledgeDocument, KnowledgeSource, Sensitivity, ) from cyberbrain26.knowledge_plane.repository import SQLiteKnowledgeRepository from cyberbrain26.security_plane.audit import AuditEventType, AuditLog from cyberbrain26.security_plane.egress import ( DataClassification, EgressGate, EgressRequest, ) @dataclass(frozen=True) class FetchResult: """Result of a single URL fetch.""" url: str content: bytes media_type: str status_code: int content_sha256: str = field(init=False) def __post_init__(self) -> None: object.__setattr__(self, "content_sha256", hashlib.sha256(self.content).hexdigest()) @dataclass(frozen=True) class NetworkSourceConfig: """Configuration for a network source.""" source_id: str allowed_hosts: frozenset[str] = field(default_factory=frozenset) trust: DocumentTrust = DocumentTrust.EXTERNAL sensitivity: Sensitivity = Sensitivity.PERSONAL timeout: float = 30.0 max_content_bytes: int = 10_000_000 # 10 MB class NetworkSourceAdapter: """Fetch and register content from approved network URLs. All outbound requests are authorized through the EgressGate before any bytes are transferred. The adapter never persists raw bytes to the knowledge plane — only metadata and a SHA-256 digest are stored. """ def __init__( self, repository: SQLiteKnowledgeRepository, *, egress_gate: EgressGate, audit_log: AuditLog | None = None, config: NetworkSourceConfig | None = None, ) -> None: self.repository = repository self.egress_gate = egress_gate self.audit_log = audit_log self.config = config or NetworkSourceConfig(source_id="network") @staticmethod def _is_private_host(host: str) -> bool: """Return True if host is a private/reserved/link-local IP address. This prevents SSRF via redirects to cloud metadata endpoints, localhost, or internal RFC1918 addresses. """ try: ip = ipaddress.ip_address(host) return ( ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved or ip.is_multicast ) except ValueError: return False def fetch(self, url: str, *, agent_id: str = "knowledge-ingester") -> FetchResult: """Fetch content from a URL through the egress gate. Automatic redirects are disabled. Every redirect target is re-validated against the allowlist and blocked if it resolves to a private address. """ self._validate_url(url) parsed = urlsplit(url) host = (parsed.hostname or "").lower() if self.config.allowed_hosts and host not in self.config.allowed_hosts: raise PermissionError(f"host '{host}' is not in the allowed hosts list") if self._is_private_host(host): raise PermissionError(f"host '{host}' is a private/internal address; access denied") # Build egress request payload_sha256 = hashlib.sha256(url.encode()).hexdigest() classification = ( DataClassification.PUBLIC if self.config.sensitivity == Sensitivity.PUBLIC else DataClassification.PERSONAL ) request = EgressRequest( agent_id=agent_id, provider_id="network-source", payload_sha256=payload_sha256, classifications=frozenset({classification}), destination=url, ) # Authorize through egress gate (fail-closed) self.egress_gate.authorize(request) # Fetch content with manual redirect validation (no auto-follow) current_url = url max_redirects = 5 response = None for _ in range(max_redirects + 1): with httpx.Client(timeout=self.config.timeout, follow_redirects=False) as client: response = client.get(current_url) response.raise_for_status() if response.is_redirect and response.headers.get("location"): # Validate redirect target before following redirect_url = response.headers["location"] self._validate_url(redirect_url) r_parsed = urlsplit(redirect_url) r_host = (r_parsed.hostname or "").lower() if self.config.allowed_hosts and r_host not in self.config.allowed_hosts: raise PermissionError(f"redirect target '{r_host}' is not in the allowed hosts list") if self._is_private_host(r_host): raise PermissionError(f"redirect target '{r_host}' is a private/internal address; access denied") # Re-authorize for the new destination payload_sha256 = hashlib.sha256(redirect_url.encode()).hexdigest() request = EgressRequest( agent_id=agent_id, provider_id="network-source", payload_sha256=payload_sha256, classifications=frozenset({classification}), destination=redirect_url, ) self.egress_gate.authorize(request) current_url = redirect_url else: break else: raise RuntimeError(f"too many redirects (max {max_redirects})") content = response.content if len(content) > self.config.max_content_bytes: raise ValueError( f"content size {len(content)} exceeds max {self.config.max_content_bytes}" ) media_type = response.headers.get("content-type", "application/octet-stream") media_type = media_type.split(";")[0].strip() if not media_type or media_type == "application/octet-stream": guessed, _ = mimetypes.guess_type(url) if guessed: media_type = guessed result = FetchResult(url=current_url, content=content, media_type=media_type, status_code=response.status_code) if self.audit_log is not None: self.audit_log.record( AuditEventType.DATA_ACCESS, actor=agent_id, agent_id=agent_id, action="network_source_fetched", resource=current_url, result="allowed", metadata={ "content_sha256": result.content_sha256, "media_type": media_type, "status_code": response.status_code, "content_length": len(content), }, ) return result def register_url( self, url: str, *, agent_id: str = "knowledge-ingester", metadata: dict[str, Any] | None = None, ) -> tuple[KnowledgeDocument, bool]: """Fetch a URL and register it in the knowledge plane.""" result = self.fetch(url, agent_id=agent_id) source = KnowledgeSource( source_id=self.config.source_id, source_uri=url, trust=self.config.trust, sensitivity=self.config.sensitivity, approved=True, metadata={"url": url, "host": urlsplit(url).hostname or ""}, ) registration = DocumentRegistration( source=source, content=result.content, media_type=result.media_type, metadata={ "url": url, "fetched_at": __import__("time").time(), "status_code": result.status_code, "content_sha256": result.content_sha256, **(metadata or {}), }, ) return self.repository.register(registration) def extract_text(self, result: FetchResult) -> str: """Extract plain text from fetched content based on media type.""" if "text/html" in result.media_type: return self._extract_html(result.content) if "application/json" in result.media_type: return result.content.decode("utf-8", errors="replace") if "text/plain" in result.media_type: return result.content.decode("utf-8", errors="replace") # Fallback: try UTF-8 decode return result.content.decode("utf-8", errors="replace") @staticmethod def _extract_html(content: bytes) -> str: """Simple HTML to text extraction (no external dependencies).""" text = content.decode("utf-8", errors="replace") # Remove script and style tags text = re.sub(r"]*>.*?", " ", text, flags=re.DOTALL | re.IGNORECASE) text = re.sub(r"]*>.*?", " ", text, flags=re.DOTALL | re.IGNORECASE) # Remove HTML tags text = re.sub(r"<[^>]+>", " ", text) # Collapse whitespace text = re.sub(r"\s+", " ", text).strip() return text @staticmethod def _validate_url(url: str) -> None: parsed = urlsplit(url) if parsed.scheme not in ("http", "https"): raise ValueError(f"unsupported URL scheme: {parsed.scheme}") if not parsed.hostname: raise ValueError("URL must have a hostname") __all__ = ["FetchResult", "NetworkSourceAdapter", "NetworkSourceConfig"]