From b6f099915a76875888148b6a9369a5b4ba73488e Mon Sep 17 00:00:00 2001 From: paulkarayan Date: Wed, 8 Jul 2026 12:29:41 -0700 Subject: [PATCH 1/9] feat(connectors): add url source connector (spike) A `url` source connector: ingest a literal list of {url, filename}. Replaces a bespoke indexer+downloader pair with a normal Indexer/Downloader. UrlIndexer yields FileData per {url, filename}; UrlDownloader does an SSRF-safe fetch with the private-IP guard as connector config (allow_private_ips) and no TOCTOU (resolve once, validate all records, pin the socket to the validated IP, revalidate redirect hops). Not yet registered in the source registry / utic_types enum / plugin manifests (follow-ups). scripts/url_connector_spike.py proves it end-to-end. Co-Authored-By: Claude Opus 4.8 --- scripts/url_connector_spike.py | 123 +++++++++ .../processes/connectors/url.py | 234 ++++++++++++++++++ 2 files changed, 357 insertions(+) create mode 100644 scripts/url_connector_spike.py create mode 100644 unstructured_ingest/processes/connectors/url.py diff --git a/scripts/url_connector_spike.py b/scripts/url_connector_spike.py new file mode 100644 index 000000000..b372931c3 --- /dev/null +++ b/scripts/url_connector_spike.py @@ -0,0 +1,123 @@ +"""Spike runner for the `url` source connector — proves it end-to-end with no +platform stack. Serves files on 127.0.0.1, then index -> download, and exercises +the SSRF guard (validator on public/private, private blocked by default, redirect +followed with per-hop revalidation). + +Run: python scripts/url_connector_spike.py +""" + +import http.server +import socketserver +import tempfile +import threading +from pathlib import Path + +from unstructured_ingest.processes.connectors.url import ( + CONNECTOR_TYPE, + FileReference, + UrlDownloaderConfig, + UrlIndexerConfig, + _validate_and_pin, + url_source_entry, +) + + +class _Handler(http.server.SimpleHTTPRequestHandler): + def do_GET(self): # noqa: N802 + if self.path == "/redirect": + self.send_response(302) + self.send_header("Location", "/a.txt") + self.end_headers() + return + super().do_GET() + + def log_message(self, *a): + pass + + +def _serve(directory: Path) -> tuple[socketserver.TCPServer, int]: + handler = lambda *a, **k: _Handler(*a, directory=str(directory), **k) # noqa: E731 + httpd = socketserver.TCPServer(("127.0.0.1", 0), handler) + threading.Thread(target=httpd.serve_forever, daemon=True).start() + return httpd, httpd.server_address[1] + + +def main() -> None: + # validator: no network, proves per-hop check on public vs private + assert _validate_and_pin("8.8.8.8", allow_private=False) == "8.8.8.8" + for bad in ("127.0.0.1", "10.0.0.1", "169.254.1.1", "0.0.0.0"): + try: + _validate_and_pin(bad, allow_private=False) + raise AssertionError(f"validator failed to block {bad}") + except Exception as e: # noqa: BLE001 + assert "refusing" in str(e).lower(), (bad, e) + + work = Path(tempfile.mkdtemp()) + served = work / "served" + served.mkdir() + (served / "a.txt").write_text("hello alpha") + (served / "b.txt").write_text("hello beta") + + httpd, port = _serve(served) + base = f"http://127.0.0.1:{port}" + dl_dir = work / "downloads" + + try: + conn = url_source_entry.connection_config() + indexer = url_source_entry.indexer( + connection_config=conn, + index_config=UrlIndexerConfig( + files=[ + FileReference(url=f"{base}/a.txt", filename="a.txt"), + FileReference(url=f"{base}/b.txt", filename="b.txt"), + ] + ), + ) + # allow_private_ips=True because the test server is on 127.0.0.1 + downloader = url_source_entry.downloader( + connection_config=conn, + download_config=UrlDownloaderConfig(download_dir=dl_dir, allow_private_ips=True), + ) + + file_datas = list(indexer.run()) + assert len(file_datas) == 2, file_datas + assert all(fd.connector_type == CONNECTOR_TYPE for fd in file_datas) + + contents = {} + for fd in file_datas: + resp = downloader.run(file_data=fd) + path = Path(resp["path"]) + assert path.exists(), path + contents[fd.source_identifiers.filename] = path.read_text() + assert resp["file_data"].local_download_path == str(path.resolve()) + assert contents == {"a.txt": "hello alpha", "b.txt": "hello beta"}, contents + + # redirect is followed (each hop revalidated) + redirect_fd = next(iter(indexer.run())) + redirect_fd.metadata.url = f"{base}/redirect" + redirect_fd.source_identifiers.filename = "redir.txt" + redirect_fd.source_identifiers.fullpath = "redir.txt" + redirect_fd.source_identifiers.rel_path = "redir.txt" + resp = downloader.run(file_data=redirect_fd) + assert Path(resp["path"]).read_text() == "hello alpha" + + # private blocked by default + guarded = url_source_entry.downloader( + connection_config=conn, + download_config=UrlDownloaderConfig(download_dir=dl_dir), # allow_private_ips=False + ) + blocked = False + try: + guarded.run(file_data=file_datas[0]) + except Exception as e: # noqa: BLE001 + blocked = "refusing" in str(e).lower() + assert blocked, "SSRF guard did NOT block a private IP by default" + + finally: + httpd.shutdown() + + print("SPIKE OK — conforms to interfaces; SSRF guard is config; no TOCTOU") + + +if __name__ == "__main__": + main() diff --git a/unstructured_ingest/processes/connectors/url.py b/unstructured_ingest/processes/connectors/url.py new file mode 100644 index 000000000..fbd4954a4 --- /dev/null +++ b/unstructured_ingest/processes/connectors/url.py @@ -0,0 +1,234 @@ +"""A `url` source connector: ingest a literal list of {url, filename}. + +It replaces a bespoke indexer + downloader pair with a normal +Indexer/Downloader pair: + + - UrlIndexer -- yields FileData for each configured {url, filename} + - UrlDownloader -- SSRF-safe HTTP(S) fetch to a local path + +The SSRF guard that lived in the playground downloader as an operator env var +(`environment != "dev"`) is expressed here as connector config +(`UrlDownloaderConfig.allow_private_ips`), and the TOCTOU in that original guard +is closed (see `_ssrf_safe_get`). + +NOT YET REGISTERED. `add_source_entry("url", url_source_entry)` in +`processes/connectors/__init__.py` + the utic_types enum + plugin manifests are +follow-ups. Importing this module has no effect on the shipped registry. +""" + +import http.client +import ipaddress +import socket +import ssl +import urllib.parse +import uuid +from dataclasses import dataclass +from typing import Any, Generator + +from pydantic import BaseModel, Field, Secret + +from unstructured_ingest.data_types.file_data import ( + FileData, + FileDataSourceMetadata, + SourceIdentifiers, +) +from unstructured_ingest.error import ValueError as IngestValueError +from unstructured_ingest.interfaces import ( + AccessConfig, + ConnectionConfig, + Downloader, + DownloaderConfig, + DownloadResponse, + Indexer, + IndexerConfig, +) +from unstructured_ingest.processes.connector_registry import SourceRegistryEntry + +CONNECTOR_TYPE = "url" + + +# --- connection (no auth: these are references we minted) ------------------ +class UrlAccessConfig(AccessConfig): + pass + + +class UrlConnectionConfig(ConnectionConfig): + access_config: Secret[UrlAccessConfig] = Field( + default_factory=UrlAccessConfig, validate_default=True + ) + + +# --- indexer: "here is a literal list of {url, filename}" ------------------ +class FileReference(BaseModel): + url: str + filename: str + + +class UrlIndexerConfig(IndexerConfig): + files: list[FileReference] = Field( + default_factory=list, + description="Explicit list of files to ingest, each a {url, filename}.", + ) + + +@dataclass +class UrlIndexer(Indexer): + connection_config: UrlConnectionConfig + index_config: UrlIndexerConfig + + def precheck(self) -> None: + if not self.index_config.files: + raise IngestValueError("No files provided") + + def run(self, **kwargs: Any) -> Generator[FileData, None, None]: + for ref in self.index_config.files: + filename = ref.filename + yield FileData( + identifier=uuid.uuid5(uuid.NAMESPACE_URL, ref.url).hex, + connector_type=CONNECTOR_TYPE, + source_identifiers=SourceIdentifiers( + filename=filename, + fullpath=filename, + rel_path=filename, + ), + metadata=FileDataSourceMetadata(url=ref.url), + display_name=filename, + ) + + +# --- downloader: HTTP fetch with SSRF guard as *config* ------------------- +class UrlDownloaderConfig(DownloaderConfig): + allow_private_ips: bool = Field( + default=False, + description="Permit fetching URLs that resolve to private IPs. " + "Replaces the playground downloader's `environment != 'dev'` env check.", + ) + timeout_seconds: int = Field(default=120) + + +# --- SSRF-safe fetch: no TOCTOU -------------------------------------------- +# The naive pattern (resolve -> validate -> urlopen) re-resolves DNS inside +# urlopen, so a hostile server can return a public IP to the validating lookup +# and a private one to the fetch. We close that by resolving ONCE, validating +# EVERY returned address, then connecting the socket to the pinned IP (keeping +# the original hostname for Host header + TLS SNI/cert). Redirects are followed +# manually so each hop is revalidated instead of trusting urllib's re-resolve. + + +class _PinnedHTTPConnection(http.client.HTTPConnection): + def __init__(self, host: str, pinned_ip: str, **kw: Any): + super().__init__(host, **kw) + self._pinned_ip = pinned_ip + + def connect(self) -> None: + self.sock = socket.create_connection((self._pinned_ip, self.port), self.timeout) + + +class _PinnedHTTPSConnection(http.client.HTTPSConnection): + def __init__(self, host: str, pinned_ip: str, **kw: Any): + super().__init__(host, **kw) + self._pinned_ip = pinned_ip + + def connect(self) -> None: + sock = socket.create_connection((self._pinned_ip, self.port), self.timeout) + # server_hostname=self.host -> SNI + cert validated against the real + # hostname, while the socket is pinned to the validated IP. + self.sock = self._context.wrap_socket(sock, server_hostname=self.host) + + +def _validate_and_pin(host: str, allow_private: bool) -> str: + """Resolve host, reject if ANY address is non-public, return one pinned IP.""" + try: + infos = socket.getaddrinfo(host, None, proto=socket.IPPROTO_TCP) + except socket.gaierror as e: + raise IngestValueError(f"Failed to resolve host: {host}") from e + ips = sorted({info[4][0] for info in infos}) + if not ips: + raise IngestValueError(f"No addresses for host: {host}") + for ip in ips: + addr = ipaddress.ip_address(ip) + if not allow_private and ( + addr.is_private + or addr.is_loopback + or addr.is_link_local + or addr.is_reserved + or addr.is_multicast + or addr.is_unspecified + ): + raise IngestValueError(f"Refusing non-public address {ip} for host {host}") + return ips[0] # deterministic pin; all addresses already validated + + +def _ssrf_safe_get(url: str, allow_private: bool, timeout: int, max_redirects: int = 5) -> bytes: + for _ in range(max_redirects + 1): + parts = urllib.parse.urlsplit(url) + if parts.scheme not in ("http", "https"): + raise IngestValueError(f"Unsupported scheme: {parts.scheme!r}") + host = parts.hostname + if not host: + raise IngestValueError(f"No host in url: {url}") + port = parts.port or (443 if parts.scheme == "https" else 80) + pinned = _validate_and_pin(host, allow_private) + path = parts.path or "/" + if parts.query: + path = f"{path}?{parts.query}" + + if parts.scheme == "https": + conn: http.client.HTTPConnection = _PinnedHTTPSConnection( + host, pinned, port=port, timeout=timeout, context=ssl.create_default_context() + ) + else: + conn = _PinnedHTTPConnection(host, pinned, port=port, timeout=timeout) + try: + conn.request("GET", path, headers={"Host": host}) + resp = conn.getresponse() + if resp.status in (301, 302, 303, 307, 308): + location = resp.getheader("Location") + resp.read() + if not location: + raise IngestValueError("Redirect without Location header") + url = urllib.parse.urljoin(url, location) # revalidated next iteration + continue + if resp.status != 200: + raise RuntimeError(f"GET {url} -> {resp.status}") + return resp.read() + finally: + conn.close() + raise IngestValueError(f"Too many redirects for url: {url}") + + +@dataclass +class UrlDownloader(Downloader): + connection_config: UrlConnectionConfig + download_config: UrlDownloaderConfig + connector_type: str = CONNECTOR_TYPE + + def is_async(self) -> bool: + return False + + def run(self, file_data: FileData, **kwargs: Any) -> DownloadResponse: + url = file_data.metadata.url + if not url: + raise IngestValueError(f"No url on file_data: {file_data.identifier}") + + data = _ssrf_safe_get( + url, + allow_private=self.download_config.allow_private_ips, + timeout=self.download_config.timeout_seconds, + ) + + download_path = self.get_download_path(file_data=file_data) + download_path.parent.mkdir(parents=True, exist_ok=True) + with open(download_path, "wb") as f: + f.write(data) + + return self.generate_download_response(file_data=file_data, download_path=download_path) + + +url_source_entry = SourceRegistryEntry( + indexer=UrlIndexer, + indexer_config=UrlIndexerConfig, + downloader=UrlDownloader, + downloader_config=UrlDownloaderConfig, + connection_config=UrlConnectionConfig, +) From 168091a7b47b8584dccef17e8543f5b56178ad6d Mon Sep 17 00:00:00 2001 From: paulkarayan Date: Wed, 8 Jul 2026 13:05:10 -0700 Subject: [PATCH 2/9] feat(connectors): register url source connector + httpx pinned transport - Port the SSRF-safe fetch from stdlib http.client to an httpx pinned transport (matches repo convention; httpx is a lazily-imported `url` extra). Resolves and validates ALL records once, then pins the httpx transport's TCP connect to the validated IP while httpcore keeps TLS SNI/cert on the real hostname -> no TOCTOU. Redirects disabled + followed manually so each hop is revalidated. - Register `url` in the source registry (URL_CONNECTOR_TYPE) and add the `url` extra. - Unit tests: indexer, download happy path, redirect-follow, validator (public/private/metadata/multi-record), redirect-target revalidation, and a guard test asserting the httpx pinning seam actually connects to the pinned IP. Co-Authored-By: Claude Opus 4.8 --- pyproject.toml | 1 + test/unit/connectors/test_url.py | 172 ++++++++++++++++++ .../processes/connectors/__init__.py | 3 + .../processes/connectors/url.py | 113 ++++++------ 4 files changed, 231 insertions(+), 58 deletions(-) create mode 100644 test/unit/connectors/test_url.py diff --git a/pyproject.toml b/pyproject.toml index dfa07b24e..1aed85991 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -87,6 +87,7 @@ singlestore = ["pandas", "singlestoredb"] slack = ["slack_sdk[optional]"] snowflake = ["pandas", "snowflake-connector-python", "psycopg2-binary"] teradata = ["pandas", "teradatasql"] +url = ["httpx"] vastdb = ["pandas", "pyarrow", "vastdb", "ibis"] vectara = ["requests", "aiofiles", "httpx"] weaviate = ["weaviate-client"] diff --git a/test/unit/connectors/test_url.py b/test/unit/connectors/test_url.py new file mode 100644 index 000000000..f2505566e --- /dev/null +++ b/test/unit/connectors/test_url.py @@ -0,0 +1,172 @@ +import http.server +import socketserver +import threading +from pathlib import Path + +import httpx +import pytest + +from unstructured_ingest.error import ValueError as IngestValueError +from unstructured_ingest.processes.connectors import url as url_mod +from unstructured_ingest.processes.connectors.url import ( + CONNECTOR_TYPE, + FileReference, + UrlConnectionConfig, + UrlDownloaderConfig, + UrlIndexer, + UrlIndexerConfig, + _pinned_transport, + _ssrf_safe_get, + _validate_and_pin, + url_source_entry, +) + + +class _Handler(http.server.SimpleHTTPRequestHandler): + redirect_to = "/a.txt" + + def do_GET(self): # noqa: N802 + if self.path == "/redirect": + self.send_response(302) + self.send_header("Location", self.redirect_to) + self.end_headers() + return + super().do_GET() + + def log_message(self, *a): + pass + + +@pytest.fixture +def server(tmp_path): + served = tmp_path / "served" + served.mkdir() + (served / "a.txt").write_text("hello alpha") + handler = lambda *a, **k: _Handler(*a, directory=str(served), **k) # noqa: E731 + httpd = socketserver.TCPServer(("127.0.0.1", 0), handler) + threading.Thread(target=httpd.serve_forever, daemon=True).start() + try: + yield httpd, f"http://127.0.0.1:{httpd.server_address[1]}" + finally: + httpd.shutdown() + + +# --- indexer --------------------------------------------------------------- +def test_indexer_yields_filedata(): + idx = UrlIndexer( + connection_config=UrlConnectionConfig(), + index_config=UrlIndexerConfig( + files=[FileReference(url="https://x/y.pdf", filename="y.pdf")] + ), + ) + fds = list(idx.run()) + assert len(fds) == 1 + assert fds[0].connector_type == CONNECTOR_TYPE + assert fds[0].metadata.url == "https://x/y.pdf" + assert fds[0].source_identifiers.filename == "y.pdf" + + +def test_indexer_precheck_rejects_empty(): + idx = UrlIndexer(connection_config=UrlConnectionConfig(), index_config=UrlIndexerConfig()) + with pytest.raises(IngestValueError): + idx.precheck() + + +# --- downloader happy path ------------------------------------------------- +def _downloader(tmp_path, allow_private=True): + return url_source_entry.downloader( + connection_config=UrlConnectionConfig(), + download_config=UrlDownloaderConfig(download_dir=tmp_path, allow_private_ips=allow_private), + ) + + +def test_download_happy_path(tmp_path, server): + _, base = server + fd = next( + iter( + UrlIndexer( + connection_config=UrlConnectionConfig(), + index_config=UrlIndexerConfig( + files=[FileReference(url=f"{base}/a.txt", filename="a.txt")] + ), + ).run() + ) + ) + resp = _downloader(tmp_path).run(file_data=fd) + assert Path(resp["path"]).read_text() == "hello alpha" + assert resp["file_data"].local_download_path == str(Path(resp["path"]).resolve()) + + +def test_download_follows_redirect(tmp_path, server): + _, base = server + fd = next( + iter( + UrlIndexer( + connection_config=UrlConnectionConfig(), + index_config=UrlIndexerConfig( + files=[FileReference(url=f"{base}/redirect", filename="r.txt")] + ), + ).run() + ) + ) + resp = _downloader(tmp_path).run(file_data=fd) + assert Path(resp["path"]).read_text() == "hello alpha" + + +# --- SSRF validator -------------------------------------------------------- +def test_validate_and_pin_allows_public(): + assert _validate_and_pin("8.8.8.8", allow_private=False) == "8.8.8.8" + + +@pytest.mark.parametrize( + "bad", ["127.0.0.1", "10.0.0.1", "192.168.1.1", "169.254.169.254", "0.0.0.0"] +) +def test_validate_and_pin_blocks_nonpublic(bad): + with pytest.raises(IngestValueError, match="Refusing non-public"): + _validate_and_pin(bad, allow_private=False) + + +def test_validate_and_pin_blocks_when_any_record_private(monkeypatch): + # DNS returns one public + one private -> reject (no multi-record bypass). + monkeypatch.setattr( + url_mod.socket, + "getaddrinfo", + lambda *a, **k: [(0, 0, 0, "", ("64.99.218.207", 0)), (0, 0, 0, "", ("10.1.2.3", 0))], + ) + with pytest.raises(IngestValueError, match="Refusing non-public"): + _validate_and_pin("evil.example.com", allow_private=False) + + +def test_redirect_target_is_revalidated(tmp_path, server, monkeypatch): + # The post-redirect host must be validated, not just the first hop. + _Handler.redirect_to = "http://evil.internal/secret" + _, base = server + seen = [] + + def fake_pin(host, allow_private): + seen.append(host) + if host == "evil.internal": + raise IngestValueError("Refusing non-public address 10.0.0.1 for host evil.internal") + return "127.0.0.1" # first hop -> reach the local test server + + monkeypatch.setattr(url_mod, "_validate_and_pin", fake_pin) + try: + with pytest.raises(IngestValueError, match="Refusing non-public"): + _ssrf_safe_get(f"{base}/redirect", allow_private=False, timeout=5) + assert "evil.internal" in seen # proves the redirect hop was revalidated + finally: + _Handler.redirect_to = "/a.txt" + + +def test_pinned_transport_connects_to_pinned_ip(server): + # Guards the private httpx seam: the socket must target the pinned IP, not the + # URL host. Requesting an unresolvable host but pinning to 127.0.0.1 must still + # reach the local server. If httpcore renames the backend seam, this fails loudly. + _, base = server + port = base.rsplit(":", 1)[1] + with httpx.Client( + transport=_pinned_transport("127.0.0.1"), follow_redirects=False, timeout=5 + ) as client: + resp = client.get(f"http://does-not-resolve.invalid:{port}/a.txt") + assert resp.status_code == 200 + assert resp.text == "hello alpha" diff --git a/unstructured_ingest/processes/connectors/__init__.py b/unstructured_ingest/processes/connectors/__init__.py index 1f492e0c9..4bec146c0 100644 --- a/unstructured_ingest/processes/connectors/__init__.py +++ b/unstructured_ingest/processes/connectors/__init__.py @@ -65,6 +65,8 @@ from .sharepoint import sharepoint_source_entry from .slack import CONNECTOR_TYPE as SLACK_CONNECTOR_TYPE from .slack import slack_source_entry +from .url import CONNECTOR_TYPE as URL_CONNECTOR_TYPE +from .url import url_source_entry from .vectara import CONNECTOR_TYPE as VECTARA_CONNECTOR_TYPE from .vectara import vectara_destination_entry from .zendesk.zendesk import CONNECTOR_TYPE as ZENDESK_CONNECTOR_TYPE @@ -87,6 +89,7 @@ add_source_entry(source_type=LOCAL_CONNECTOR_TYPE, entry=local_source_entry) add_destination_entry(destination_type=LOCAL_CONNECTOR_TYPE, entry=local_destination_entry) +add_source_entry(source_type=URL_CONNECTOR_TYPE, entry=url_source_entry) add_source_entry(source_type=ONEDRIVE_CONNECTOR_TYPE, entry=onedrive_source_entry) add_destination_entry(destination_type=ONEDRIVE_CONNECTOR_TYPE, entry=onedrive_destination_entry) diff --git a/unstructured_ingest/processes/connectors/url.py b/unstructured_ingest/processes/connectors/url.py index fbd4954a4..9e98a1ce9 100644 --- a/unstructured_ingest/processes/connectors/url.py +++ b/unstructured_ingest/processes/connectors/url.py @@ -16,14 +16,12 @@ follow-ups. Importing this module has no effect on the shipped registry. """ -import http.client import ipaddress import socket -import ssl import urllib.parse import uuid from dataclasses import dataclass -from typing import Any, Generator +from typing import TYPE_CHECKING, Any, Generator from pydantic import BaseModel, Field, Secret @@ -43,6 +41,10 @@ IndexerConfig, ) from unstructured_ingest.processes.connector_registry import SourceRegistryEntry +from unstructured_ingest.utils.dep_check import requires_dependencies + +if TYPE_CHECKING: + import httpx CONNECTOR_TYPE = "url" @@ -107,33 +109,13 @@ class UrlDownloaderConfig(DownloaderConfig): # --- SSRF-safe fetch: no TOCTOU -------------------------------------------- -# The naive pattern (resolve -> validate -> urlopen) re-resolves DNS inside -# urlopen, so a hostile server can return a public IP to the validating lookup -# and a private one to the fetch. We close that by resolving ONCE, validating -# EVERY returned address, then connecting the socket to the pinned IP (keeping -# the original hostname for Host header + TLS SNI/cert). Redirects are followed -# manually so each hop is revalidated instead of trusting urllib's re-resolve. - - -class _PinnedHTTPConnection(http.client.HTTPConnection): - def __init__(self, host: str, pinned_ip: str, **kw: Any): - super().__init__(host, **kw) - self._pinned_ip = pinned_ip - - def connect(self) -> None: - self.sock = socket.create_connection((self._pinned_ip, self.port), self.timeout) - - -class _PinnedHTTPSConnection(http.client.HTTPSConnection): - def __init__(self, host: str, pinned_ip: str, **kw: Any): - super().__init__(host, **kw) - self._pinned_ip = pinned_ip - - def connect(self) -> None: - sock = socket.create_connection((self._pinned_ip, self.port), self.timeout) - # server_hostname=self.host -> SNI + cert validated against the real - # hostname, while the socket is pinned to the validated IP. - self.sock = self._context.wrap_socket(sock, server_hostname=self.host) +# The naive pattern (resolve -> validate -> get) re-resolves DNS at fetch time, +# so a hostile/rebinding resolver can answer public to the validating lookup and +# private to the fetch. We close that by resolving ONCE, validating EVERY +# returned address, then pinning the httpx transport's socket to the validated +# IP -- httpcore still drives TLS SNI + cert verification off the real hostname, +# so pinning cannot be bypassed. Redirects are disabled and followed manually so +# each hop is revalidated. def _validate_and_pin(host: str, allow_private: bool) -> str: @@ -159,41 +141,55 @@ def _validate_and_pin(host: str, allow_private: bool) -> str: return ips[0] # deterministic pin; all addresses already validated +def _pinned_transport(pinned_ip: str) -> "httpx.HTTPTransport": + """An httpx transport whose TCP connect targets `pinned_ip`, while httpcore + keeps TLS SNI/cert bound to the request's real hostname.""" + import httpx + from httpcore._backends.sync import SyncBackend + + class _PinnedBackend(SyncBackend): + def connect_tcp(self, host, port, timeout=None, local_address=None, socket_options=None): + return super().connect_tcp( + pinned_ip, + port, + timeout=timeout, + local_address=local_address, + socket_options=socket_options, + ) + + transport = httpx.HTTPTransport(retries=0) + # Private seam (asserted by test_pinned_transport_connects_to_pinned_ip); if a + # future httpcore renames it, that test fails loudly rather than silently + # un-pinning the SSRF guard. + transport._pool._network_backend = _PinnedBackend() + return transport + + def _ssrf_safe_get(url: str, allow_private: bool, timeout: int, max_redirects: int = 5) -> bytes: + import httpx + + current = url for _ in range(max_redirects + 1): - parts = urllib.parse.urlsplit(url) + parts = urllib.parse.urlsplit(current) if parts.scheme not in ("http", "https"): raise IngestValueError(f"Unsupported scheme: {parts.scheme!r}") host = parts.hostname if not host: - raise IngestValueError(f"No host in url: {url}") - port = parts.port or (443 if parts.scheme == "https" else 80) + raise IngestValueError(f"No host in url: {current}") pinned = _validate_and_pin(host, allow_private) - path = parts.path or "/" - if parts.query: - path = f"{path}?{parts.query}" - - if parts.scheme == "https": - conn: http.client.HTTPConnection = _PinnedHTTPSConnection( - host, pinned, port=port, timeout=timeout, context=ssl.create_default_context() - ) - else: - conn = _PinnedHTTPConnection(host, pinned, port=port, timeout=timeout) - try: - conn.request("GET", path, headers={"Host": host}) - resp = conn.getresponse() - if resp.status in (301, 302, 303, 307, 308): - location = resp.getheader("Location") - resp.read() - if not location: - raise IngestValueError("Redirect without Location header") - url = urllib.parse.urljoin(url, location) # revalidated next iteration - continue - if resp.status != 200: - raise RuntimeError(f"GET {url} -> {resp.status}") - return resp.read() - finally: - conn.close() + with httpx.Client( + transport=_pinned_transport(pinned), timeout=timeout, follow_redirects=False + ) as client: + resp = client.get(current) + if resp.is_redirect: + location = resp.headers.get("location") + if not location: + raise IngestValueError("Redirect without Location header") + current = str(httpx.URL(current).join(location)) # revalidated next iteration + continue + if resp.status_code != 200: + raise RuntimeError(f"GET {current} -> {resp.status_code}") + return resp.content raise IngestValueError(f"Too many redirects for url: {url}") @@ -206,6 +202,7 @@ class UrlDownloader(Downloader): def is_async(self) -> bool: return False + @requires_dependencies(["httpx"], extras="url") def run(self, file_data: FileData, **kwargs: Any) -> DownloadResponse: url = file_data.metadata.url if not url: From 7b1e76d71e9615f07970e3e0ece93ad23aeb3193 Mon Sep 17 00:00:00 2001 From: paulkarayan Date: Wed, 8 Jul 2026 16:56:09 -0700 Subject: [PATCH 3/9] fix(connectors): use is_global allowlist for url SSRF check Mirrors the platform-plugins SSRF fix (cubic P1): `not addr.is_global` is a stricter allowlist than the private/loopback/... denylist and also rejects non-globally-routable ranges like CGNAT 100.64.0.0/10. Adds a 100.64.0.1 test. Co-Authored-By: Claude Opus 4.8 --- test/unit/connectors/test_url.py | 2 +- unstructured_ingest/processes/connectors/url.py | 13 ++++--------- 2 files changed, 5 insertions(+), 10 deletions(-) diff --git a/test/unit/connectors/test_url.py b/test/unit/connectors/test_url.py index f2505566e..0e36a4918 100644 --- a/test/unit/connectors/test_url.py +++ b/test/unit/connectors/test_url.py @@ -119,7 +119,7 @@ def test_validate_and_pin_allows_public(): @pytest.mark.parametrize( - "bad", ["127.0.0.1", "10.0.0.1", "192.168.1.1", "169.254.169.254", "0.0.0.0"] + "bad", ["127.0.0.1", "10.0.0.1", "192.168.1.1", "169.254.169.254", "0.0.0.0", "100.64.0.1"] ) def test_validate_and_pin_blocks_nonpublic(bad): with pytest.raises(IngestValueError, match="Refusing non-public"): diff --git a/unstructured_ingest/processes/connectors/url.py b/unstructured_ingest/processes/connectors/url.py index 9e98a1ce9..dfa7284d7 100644 --- a/unstructured_ingest/processes/connectors/url.py +++ b/unstructured_ingest/processes/connectors/url.py @@ -128,15 +128,10 @@ def _validate_and_pin(host: str, allow_private: bool) -> str: if not ips: raise IngestValueError(f"No addresses for host: {host}") for ip in ips: - addr = ipaddress.ip_address(ip) - if not allow_private and ( - addr.is_private - or addr.is_loopback - or addr.is_link_local - or addr.is_reserved - or addr.is_multicast - or addr.is_unspecified - ): + # is_global is an allowlist: rejects private/loopback/link-local/reserved/ + # multicast/unspecified AND non-globally-routable ranges a denylist misses + # (e.g. CGNAT 100.64.0.0/10). Skipped when allow_private is set. + if not allow_private and not ipaddress.ip_address(ip).is_global: raise IngestValueError(f"Refusing non-public address {ip} for host {host}") return ips[0] # deterministic pin; all addresses already validated From be4757baf5f6a0e1af2c1e7b9aa36d16efa5222a Mon Sep 17 00:00:00 2001 From: paulkarayan Date: Wed, 8 Jul 2026 20:15:44 -0700 Subject: [PATCH 4/9] fix(connectors): sanitize url filenames + reject duplicates (cubic) - P1 (path traversal): the download path derives from the caller-supplied filename, so `../` could escape the download dir. Reduce to a safe basename in the indexer (`_safe_filename`) and reject pure-traversal names. - P2 (collisions): reject duplicate basenames in precheck (would overwrite). - P2 (coverage): add a full-pipeline test that the default guard (allow_private_ips=False) blocks a private target through UrlDownloader. Co-Authored-By: Claude Opus 4.8 --- test/unit/connectors/test_url.py | 57 +++++++++++++++++++ .../processes/connectors/url.py | 16 +++++- 2 files changed, 72 insertions(+), 1 deletion(-) diff --git a/test/unit/connectors/test_url.py b/test/unit/connectors/test_url.py index 0e36a4918..8608d0c1c 100644 --- a/test/unit/connectors/test_url.py +++ b/test/unit/connectors/test_url.py @@ -16,6 +16,7 @@ UrlIndexer, UrlIndexerConfig, _pinned_transport, + _safe_filename, _ssrf_safe_get, _validate_and_pin, url_source_entry, @@ -72,6 +73,62 @@ def test_indexer_precheck_rejects_empty(): idx.precheck() +def test_indexer_sanitizes_path_traversal_filename(): + # a filename with path components must reduce to a safe basename (no dir escape) + idx = UrlIndexer( + connection_config=UrlConnectionConfig(), + index_config=UrlIndexerConfig( + files=[FileReference(url="https://x/y", filename="../../etc/passwd")] + ), + ) + fd = next(iter(idx.run())) + assert fd.source_identifiers.filename == "passwd" + assert fd.source_identifiers.rel_path == "passwd" + + +@pytest.mark.parametrize("bad", ["..", "../", "/", ""]) +def test_safe_filename_rejects_pure_traversal(bad): + with pytest.raises(IngestValueError): + _safe_filename(bad) + + +def test_indexer_precheck_rejects_duplicate_filenames(): + # two entries collapsing to the same basename would overwrite each other + idx = UrlIndexer( + connection_config=UrlConnectionConfig(), + index_config=UrlIndexerConfig( + files=[ + FileReference(url="https://x/a", filename="dup.txt"), + FileReference(url="https://x/b", filename="sub/dup.txt"), + ] + ), + ) + with pytest.raises(IngestValueError, match="Duplicate"): + idx.precheck() + + +def test_download_blocks_private_by_default(tmp_path, server): + # full pipeline with the production default (allow_private_ips=False) must block the + # loopback test server — exercises the guard through UrlDownloader, not just _validate_and_pin + _, base = server + fd = next( + iter( + UrlIndexer( + connection_config=UrlConnectionConfig(), + index_config=UrlIndexerConfig( + files=[FileReference(url=f"{base}/a.txt", filename="a.txt")] + ), + ).run() + ) + ) + dl = url_source_entry.downloader( + connection_config=UrlConnectionConfig(), + download_config=UrlDownloaderConfig(download_dir=tmp_path), # allow_private_ips=False + ) + with pytest.raises(IngestValueError, match="Refusing non-public"): + dl.run(file_data=fd) + + # --- downloader happy path ------------------------------------------------- def _downloader(tmp_path, allow_private=True): return url_source_entry.downloader( diff --git a/unstructured_ingest/processes/connectors/url.py b/unstructured_ingest/processes/connectors/url.py index dfa7284d7..b6eaec438 100644 --- a/unstructured_ingest/processes/connectors/url.py +++ b/unstructured_ingest/processes/connectors/url.py @@ -21,6 +21,7 @@ import urllib.parse import uuid from dataclasses import dataclass +from pathlib import Path from typing import TYPE_CHECKING, Any, Generator from pydantic import BaseModel, Field, Secret @@ -73,6 +74,15 @@ class UrlIndexerConfig(IndexerConfig): ) +def _safe_filename(filename: str) -> str: + """Reduce a caller-supplied filename to a safe basename — the download path is + derived from it, so `../` etc. must not escape the download dir.""" + name = Path(filename).name + if not name or name in (".", ".."): + raise IngestValueError(f"Invalid filename: {filename!r}") + return name + + @dataclass class UrlIndexer(Indexer): connection_config: UrlConnectionConfig @@ -81,10 +91,14 @@ class UrlIndexer(Indexer): def precheck(self) -> None: if not self.index_config.files: raise IngestValueError("No files provided") + # reject path traversal + collisions up front (the download path derives from filename) + names = [_safe_filename(f.filename) for f in self.index_config.files] + if len(set(names)) != len(names): + raise IngestValueError("Duplicate filenames are not allowed") def run(self, **kwargs: Any) -> Generator[FileData, None, None]: for ref in self.index_config.files: - filename = ref.filename + filename = _safe_filename(ref.filename) yield FileData( identifier=uuid.uuid5(uuid.NAMESPACE_URL, ref.url).hex, connector_type=CONNECTOR_TYPE, From bef2b0710479594de88e87a184199f5b36b40aba Mon Sep 17 00:00:00 2001 From: paulkarayan Date: Thu, 9 Jul 2026 10:12:57 -0700 Subject: [PATCH 5/9] chore: bump to 1.6.29, changelog, sync uv.lock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Merge main (branch was 8 behind), bump version above main's 1.6.28, add CHANGELOG entry for the url source connector, and re-lock so the url extra (httpx) is in uv.lock — the stale lock was the lint --locked failure. Co-Authored-By: Claude Opus 4.8 --- CHANGELOG.md | 6 ++++++ unstructured_ingest/__version__.py | 2 +- uv.lock | 6 +++++- 3 files changed, 12 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 95b316dea..157fd0f92 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,9 @@ +## [1.6.29] + +### Enhancements + +- **feat(connectors): add a `url` source connector.** Fetches documents directly from HTTP(S) URLs, with an SSRF guard (private/link-local/loopback ranges rejected unless `allow_private_ips` is set, revalidated per redirect hop), download-directory-safe filename handling, and rejection of duplicate filenames. Registered as `url` in the source registry; enabled via the `url` extra (`httpx`). + ## [1.6.28] ### Fixes diff --git a/unstructured_ingest/__version__.py b/unstructured_ingest/__version__.py index 5a23218dd..d2e431858 100644 --- a/unstructured_ingest/__version__.py +++ b/unstructured_ingest/__version__.py @@ -1 +1 @@ -__version__ = "1.6.28" # pragma: no cover +__version__ = "1.6.29" # pragma: no cover diff --git a/uv.lock b/uv.lock index e39c2acc8..2c55f8ff3 100644 --- a/uv.lock +++ b/uv.lock @@ -7291,6 +7291,9 @@ togetherai = [ tsv = [ { name = "unstructured", extra = ["tsv"] }, ] +url = [ + { name = "httpx" }, +] vastdb = [ { name = "ibis" }, { name = "pandas" }, @@ -7406,6 +7409,7 @@ requires-dist = [ { name = "htmlbuilder", marker = "extra == 'notion'" }, { name = "httpx", marker = "extra == 'ibm-watsonx-s3'" }, { name = "httpx", marker = "extra == 'notion'" }, + { name = "httpx", marker = "extra == 'url'" }, { name = "httpx", marker = "extra == 'vectara'" }, { name = "httpx", marker = "extra == 'zendesk'" }, { name = "hubspot-api-client", marker = "extra == 'hubspot'" }, @@ -7501,7 +7505,7 @@ requires-dist = [ { name = "weaviate-client", marker = "extra == 'weaviate'" }, { name = "wikipedia", marker = "extra == 'wikipedia'" }, ] -provides-extras = ["airtable", "astradb", "azure", "azure-ai-search", "bedrock", "biomed", "box", "chroma", "clarifai", "confluence", "couchbase", "databricks-delta-tables", "databricks-volumes", "delta-table", "discord", "doc", "docx", "dropbox", "duckdb", "elasticsearch", "epub", "gcs", "github", "gitlab", "google-drive", "hubspot", "huggingface", "ibm-watsonx-s3", "image", "jira", "kafka", "kdbai", "lancedb", "md", "milvus", "mixedbreadai", "mongodb", "neo4j", "notion", "octoai", "odt", "onedrive", "openai", "opensearch", "org", "outlook", "pdf", "pinecone", "postgres", "ppt", "pptx", "qdrant", "reddit", "redis", "remote", "rst", "rtf", "s3", "salesforce", "sftp", "sharepoint", "singlestore", "slack", "snowflake", "teradata", "togetherai", "tsv", "vastdb", "vectara", "vertexai", "voyageai", "weaviate", "wikipedia", "xlsx", "zendesk"] +provides-extras = ["airtable", "astradb", "azure", "azure-ai-search", "bedrock", "biomed", "box", "chroma", "clarifai", "confluence", "couchbase", "databricks-delta-tables", "databricks-volumes", "delta-table", "discord", "doc", "docx", "dropbox", "duckdb", "elasticsearch", "epub", "gcs", "github", "gitlab", "google-drive", "hubspot", "huggingface", "ibm-watsonx-s3", "image", "jira", "kafka", "kdbai", "lancedb", "md", "milvus", "mixedbreadai", "mongodb", "neo4j", "notion", "octoai", "odt", "onedrive", "openai", "opensearch", "org", "outlook", "pdf", "pinecone", "postgres", "ppt", "pptx", "qdrant", "reddit", "redis", "remote", "rst", "rtf", "s3", "salesforce", "sftp", "sharepoint", "singlestore", "slack", "snowflake", "teradata", "togetherai", "tsv", "url", "vastdb", "vectara", "vertexai", "voyageai", "weaviate", "wikipedia", "xlsx", "zendesk"] [package.metadata.requires-dev] lint = [{ name = "ruff" }] From 05a4518647e1beb6eda1c085df1fd0610c43b82a Mon Sep 17 00:00:00 2001 From: paulkarayan Date: Thu, 9 Jul 2026 10:16:36 -0700 Subject: [PATCH 6/9] fix(connectors): stream url downloads + correct stale docstring (cubic) - Stream the validated response to disk (_ssrf_safe_get -> _ssrf_safe_download) instead of buffering the whole body in resp.content, so a large download does not spike worker memory. Per-redirect SSRF revalidation is unchanged. - Fix the module docstring that still claimed the connector was NOT REGISTERED; it is wired via add_source_entry in processes/connectors/__init__.py. Path-traversal and duplicate-filename rejection (other cubic findings) were already handled in _safe_filename / precheck; the default-config SSRF download path is covered by test_download_blocks_private_by_default. Co-Authored-By: Claude Opus 4.8 --- test/unit/connectors/test_url.py | 6 +- .../processes/connectors/url.py | 55 +++++++++++-------- 2 files changed, 35 insertions(+), 26 deletions(-) diff --git a/test/unit/connectors/test_url.py b/test/unit/connectors/test_url.py index 8608d0c1c..e2688c586 100644 --- a/test/unit/connectors/test_url.py +++ b/test/unit/connectors/test_url.py @@ -17,7 +17,7 @@ UrlIndexerConfig, _pinned_transport, _safe_filename, - _ssrf_safe_get, + _ssrf_safe_download, _validate_and_pin, url_source_entry, ) @@ -209,7 +209,9 @@ def fake_pin(host, allow_private): monkeypatch.setattr(url_mod, "_validate_and_pin", fake_pin) try: with pytest.raises(IngestValueError, match="Refusing non-public"): - _ssrf_safe_get(f"{base}/redirect", allow_private=False, timeout=5) + _ssrf_safe_download( + f"{base}/redirect", tmp_path / "out", allow_private=False, timeout=5 + ) assert "evil.internal" in seen # proves the redirect hop was revalidated finally: _Handler.redirect_to = "/a.txt" diff --git a/unstructured_ingest/processes/connectors/url.py b/unstructured_ingest/processes/connectors/url.py index b6eaec438..fac46122d 100644 --- a/unstructured_ingest/processes/connectors/url.py +++ b/unstructured_ingest/processes/connectors/url.py @@ -9,11 +9,10 @@ The SSRF guard that lived in the playground downloader as an operator env var (`environment != "dev"`) is expressed here as connector config (`UrlDownloaderConfig.allow_private_ips`), and the TOCTOU in that original guard -is closed (see `_ssrf_safe_get`). +is closed (see `_ssrf_safe_download`). -NOT YET REGISTERED. `add_source_entry("url", url_source_entry)` in -`processes/connectors/__init__.py` + the utic_types enum + plugin manifests are -follow-ups. Importing this module has no effect on the shipped registry. +Registered as `url` in the source registry (`url_source_entry` below, wired via +`add_source_entry` in `processes/connectors/__init__.py`). """ import ipaddress @@ -174,7 +173,12 @@ def connect_tcp(self, host, port, timeout=None, local_address=None, socket_optio return transport -def _ssrf_safe_get(url: str, allow_private: bool, timeout: int, max_redirects: int = 5) -> bytes: +def _ssrf_safe_download( + url: str, dest: Path, allow_private: bool, timeout: int, max_redirects: int = 5 +) -> None: + """Stream a validated GET to `dest`. Redirects are followed manually so every hop + is revalidated; the body is streamed to disk rather than buffered whole, so a large + download does not spike worker memory.""" import httpx current = url @@ -186,19 +190,24 @@ def _ssrf_safe_get(url: str, allow_private: bool, timeout: int, max_redirects: i if not host: raise IngestValueError(f"No host in url: {current}") pinned = _validate_and_pin(host, allow_private) - with httpx.Client( - transport=_pinned_transport(pinned), timeout=timeout, follow_redirects=False - ) as client: - resp = client.get(current) - if resp.is_redirect: - location = resp.headers.get("location") - if not location: - raise IngestValueError("Redirect without Location header") - current = str(httpx.URL(current).join(location)) # revalidated next iteration - continue - if resp.status_code != 200: - raise RuntimeError(f"GET {current} -> {resp.status_code}") - return resp.content + with ( + httpx.Client( + transport=_pinned_transport(pinned), timeout=timeout, follow_redirects=False + ) as client, + client.stream("GET", current) as resp, + ): + if resp.is_redirect: + location = resp.headers.get("location") + if not location: + raise IngestValueError("Redirect without Location header") + current = str(httpx.URL(current).join(location)) # revalidated next iteration + continue + if resp.status_code != 200: + raise RuntimeError(f"GET {current} -> {resp.status_code}") + with open(dest, "wb") as f: + for chunk in resp.iter_bytes(): + f.write(chunk) + return raise IngestValueError(f"Too many redirects for url: {url}") @@ -217,17 +226,15 @@ def run(self, file_data: FileData, **kwargs: Any) -> DownloadResponse: if not url: raise IngestValueError(f"No url on file_data: {file_data.identifier}") - data = _ssrf_safe_get( + download_path = self.get_download_path(file_data=file_data) + download_path.parent.mkdir(parents=True, exist_ok=True) + _ssrf_safe_download( url, + download_path, allow_private=self.download_config.allow_private_ips, timeout=self.download_config.timeout_seconds, ) - download_path = self.get_download_path(file_data=file_data) - download_path.parent.mkdir(parents=True, exist_ok=True) - with open(download_path, "wb") as f: - f.write(data) - return self.generate_download_response(file_data=file_data, download_path=download_path) From 5c0dbdcc10891eab816717fff967143cbb2859c5 Mon Sep 17 00:00:00 2001 From: paulkarayan Date: Thu, 9 Jul 2026 11:42:47 -0700 Subject: [PATCH 7/9] fix(cli): support list[BaseModel] connector fields in CLI generation The url connector's UrlIndexerConfig.files is list[FileReference]; building the CLI (unstructured-ingest --help) iterates every registered connector's config fields and had no mapping for a list of pydantic models, raising 'Unexpected field type' and breaking --help (test_ingest_help / test_install_cli). Map list[] to Dict(), which json-loads an inline JSON array (and rejects a non-JSON value cleanly) so pydantic can validate it into the model list. Additive: this case previously raised, so no existing field mapping changes. Co-Authored-By: Claude Opus 4.8 --- test/unit/cli/test_utils.py | 17 +++++++++++++++-- .../cli/utils/model_conversion.py | 10 ++++++++++ 2 files changed, 25 insertions(+), 2 deletions(-) diff --git a/test/unit/cli/test_utils.py b/test/unit/cli/test_utils.py index 2c92a626d..3e26f6f62 100644 --- a/test/unit/cli/test_utils.py +++ b/test/unit/cli/test_utils.py @@ -2,12 +2,25 @@ from typing import Optional import pytest -from pydantic import Secret, ValidationError +from pydantic import BaseModel, Secret, ValidationError -from unstructured_ingest.cli.utils.click import extract_config +from unstructured_ingest.cli.utils.click import Dict, extract_config +from unstructured_ingest.cli.utils.model_conversion import get_type_from_annotation from unstructured_ingest.interfaces import AccessConfig, ConnectionConfig +def test_list_of_models_field_parses_json_array_on_cli(): + # a list of pydantic models maps to Dict() (json-loads the inline JSON array), and + # the parsed value must validate back into the model list + class Item(BaseModel): + a: str + + param_type = get_type_from_annotation(list[Item]) + assert isinstance(param_type, Dict) + parsed = param_type.convert('[{"a": "x"}, {"a": "y"}]') + assert [Item(**i) for i in parsed] == [Item(a="x"), Item(a="y")] + + def test_extract_config_optional_access_config(): class MyAccessConfig(AccessConfig): host: str diff --git a/unstructured_ingest/cli/utils/model_conversion.py b/unstructured_ingest/cli/utils/model_conversion.py index 67072f847..8854ee164 100644 --- a/unstructured_ingest/cli/utils/model_conversion.py +++ b/unstructured_ingest/cli/utils/model_conversion.py @@ -128,6 +128,16 @@ def get_type_from_annotation(field_type: Any) -> click.ParamType: return get_type_from_annotation(field_type=field_type) if field_origin is list and len(field_args) == 1 and field_args[0] is str: return DelimitedString() + # a list of structured models (e.g. url connector's list[FileReference]) is passed + # on the CLI as an inline JSON array; Dict() json-loads the string (and rejects a + # non-JSON value cleanly) so pydantic can validate it into the model list + if ( + field_origin is list + and len(field_args) == 1 + and isinstance(field_args[0], type) + and issubclass(field_args[0], BaseModel) + ): + return Dict() if field_type is SecretStr: return click.STRING if dict in [field_type, field_origin]: From 17685cf884a74c76614347302e22a2062fd76b50 Mon Sep 17 00:00:00 2001 From: paulkarayan Date: Thu, 9 Jul 2026 13:01:07 -0700 Subject: [PATCH 8/9] chore(connectors): drop url_connector_spike.py leftover (review) It was a throwaway spike; the connector is covered by test/unit/connectors/test_url.py. Co-Authored-By: Claude Opus 4.8 --- scripts/url_connector_spike.py | 123 --------------------------------- 1 file changed, 123 deletions(-) delete mode 100644 scripts/url_connector_spike.py diff --git a/scripts/url_connector_spike.py b/scripts/url_connector_spike.py deleted file mode 100644 index b372931c3..000000000 --- a/scripts/url_connector_spike.py +++ /dev/null @@ -1,123 +0,0 @@ -"""Spike runner for the `url` source connector — proves it end-to-end with no -platform stack. Serves files on 127.0.0.1, then index -> download, and exercises -the SSRF guard (validator on public/private, private blocked by default, redirect -followed with per-hop revalidation). - -Run: python scripts/url_connector_spike.py -""" - -import http.server -import socketserver -import tempfile -import threading -from pathlib import Path - -from unstructured_ingest.processes.connectors.url import ( - CONNECTOR_TYPE, - FileReference, - UrlDownloaderConfig, - UrlIndexerConfig, - _validate_and_pin, - url_source_entry, -) - - -class _Handler(http.server.SimpleHTTPRequestHandler): - def do_GET(self): # noqa: N802 - if self.path == "/redirect": - self.send_response(302) - self.send_header("Location", "/a.txt") - self.end_headers() - return - super().do_GET() - - def log_message(self, *a): - pass - - -def _serve(directory: Path) -> tuple[socketserver.TCPServer, int]: - handler = lambda *a, **k: _Handler(*a, directory=str(directory), **k) # noqa: E731 - httpd = socketserver.TCPServer(("127.0.0.1", 0), handler) - threading.Thread(target=httpd.serve_forever, daemon=True).start() - return httpd, httpd.server_address[1] - - -def main() -> None: - # validator: no network, proves per-hop check on public vs private - assert _validate_and_pin("8.8.8.8", allow_private=False) == "8.8.8.8" - for bad in ("127.0.0.1", "10.0.0.1", "169.254.1.1", "0.0.0.0"): - try: - _validate_and_pin(bad, allow_private=False) - raise AssertionError(f"validator failed to block {bad}") - except Exception as e: # noqa: BLE001 - assert "refusing" in str(e).lower(), (bad, e) - - work = Path(tempfile.mkdtemp()) - served = work / "served" - served.mkdir() - (served / "a.txt").write_text("hello alpha") - (served / "b.txt").write_text("hello beta") - - httpd, port = _serve(served) - base = f"http://127.0.0.1:{port}" - dl_dir = work / "downloads" - - try: - conn = url_source_entry.connection_config() - indexer = url_source_entry.indexer( - connection_config=conn, - index_config=UrlIndexerConfig( - files=[ - FileReference(url=f"{base}/a.txt", filename="a.txt"), - FileReference(url=f"{base}/b.txt", filename="b.txt"), - ] - ), - ) - # allow_private_ips=True because the test server is on 127.0.0.1 - downloader = url_source_entry.downloader( - connection_config=conn, - download_config=UrlDownloaderConfig(download_dir=dl_dir, allow_private_ips=True), - ) - - file_datas = list(indexer.run()) - assert len(file_datas) == 2, file_datas - assert all(fd.connector_type == CONNECTOR_TYPE for fd in file_datas) - - contents = {} - for fd in file_datas: - resp = downloader.run(file_data=fd) - path = Path(resp["path"]) - assert path.exists(), path - contents[fd.source_identifiers.filename] = path.read_text() - assert resp["file_data"].local_download_path == str(path.resolve()) - assert contents == {"a.txt": "hello alpha", "b.txt": "hello beta"}, contents - - # redirect is followed (each hop revalidated) - redirect_fd = next(iter(indexer.run())) - redirect_fd.metadata.url = f"{base}/redirect" - redirect_fd.source_identifiers.filename = "redir.txt" - redirect_fd.source_identifiers.fullpath = "redir.txt" - redirect_fd.source_identifiers.rel_path = "redir.txt" - resp = downloader.run(file_data=redirect_fd) - assert Path(resp["path"]).read_text() == "hello alpha" - - # private blocked by default - guarded = url_source_entry.downloader( - connection_config=conn, - download_config=UrlDownloaderConfig(download_dir=dl_dir), # allow_private_ips=False - ) - blocked = False - try: - guarded.run(file_data=file_datas[0]) - except Exception as e: # noqa: BLE001 - blocked = "refusing" in str(e).lower() - assert blocked, "SSRF guard did NOT block a private IP by default" - - finally: - httpd.shutdown() - - print("SPIKE OK — conforms to interfaces; SSRF guard is config; no TOCTOU") - - -if __name__ == "__main__": - main() From 1fd48914567d58b55e763f5d167293a5a188a092 Mon Sep 17 00:00:00 2001 From: paulkarayan Date: Thu, 9 Jul 2026 13:22:05 -0700 Subject: [PATCH 9/9] fix(connectors): reject IPv6 transition addrs + CGNAT in url SSRF guard (review) Addresses Andrew's #6: is_global alone allows IPv6 transition addresses that embed a private/metadata IPv4 (NAT64 64:ff9b::/96, 6to4 2002::/16, IPv4-mapped/-compatible) and is CGNAT-version-dependent. _is_public_address now: for IPv4 rejects CGNAT explicitly (version-independent); for IPv6 requires is_global AND that any embedded IPv4 is public. Tests cover NAT64/6to4/mapped/compat + a public IPv6. Co-Authored-By: Claude Opus 4.8 --- CHANGELOG.md | 2 +- test/unit/connectors/test_url.py | 21 +++++++-- .../processes/connectors/url.py | 43 +++++++++++++++++-- 3 files changed, 58 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 157fd0f92..ee6e4f90e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,7 @@ ### Enhancements -- **feat(connectors): add a `url` source connector.** Fetches documents directly from HTTP(S) URLs, with an SSRF guard (private/link-local/loopback ranges rejected unless `allow_private_ips` is set, revalidated per redirect hop), download-directory-safe filename handling, and rejection of duplicate filenames. Registered as `url` in the source registry; enabled via the `url` extra (`httpx`). +- **feat(connectors): add a `url` source connector.** Fetches documents directly from HTTP(S) URLs, with an SSRF guard (private/link-local/loopback/reserved/CGNAT ranges and IPv6 transition addresses embedding a private/metadata IPv4 — NAT64/6to4/IPv4-mapped/-compatible — are rejected unless `allow_private_ips` is set; the socket is pinned to the validated IP and revalidated per redirect hop), download-directory-safe filename handling, and rejection of duplicate filenames. Registered as `url` in the source registry; enabled via the `url` extra (`httpx`). ## [1.6.28] diff --git a/test/unit/connectors/test_url.py b/test/unit/connectors/test_url.py index e2688c586..c91938cff 100644 --- a/test/unit/connectors/test_url.py +++ b/test/unit/connectors/test_url.py @@ -171,12 +171,27 @@ def test_download_follows_redirect(tmp_path, server): # --- SSRF validator -------------------------------------------------------- -def test_validate_and_pin_allows_public(): - assert _validate_and_pin("8.8.8.8", allow_private=False) == "8.8.8.8" +@pytest.mark.parametrize("good", ["8.8.8.8", "2606:4700:4700::1111"]) +def test_validate_and_pin_allows_public(good): + assert _validate_and_pin(good, allow_private=False) == good @pytest.mark.parametrize( - "bad", ["127.0.0.1", "10.0.0.1", "192.168.1.1", "169.254.169.254", "0.0.0.0", "100.64.0.1"] + "bad", + [ + # IPv4: private / loopback / link-local / unspecified / CGNAT + "127.0.0.1", + "10.0.0.1", + "192.168.1.1", + "169.254.169.254", + "0.0.0.0", + "100.64.0.1", + # IPv6 transition addrs that embed a private/metadata IPv4 (is_global misses these) + "64:ff9b::a9fe:a9fe", # NAT64 -> 169.254.169.254 + "2002:a9fe:a9fe::1", # 6to4 -> 169.254.169.254 + "::ffff:10.0.0.1", # IPv4-mapped private + "::7f00:1", # IPv4-compatible -> 127.0.0.1 + ], ) def test_validate_and_pin_blocks_nonpublic(bad): with pytest.raises(IngestValueError, match="Refusing non-public"): diff --git a/unstructured_ingest/processes/connectors/url.py b/unstructured_ingest/processes/connectors/url.py index fac46122d..a6906873f 100644 --- a/unstructured_ingest/processes/connectors/url.py +++ b/unstructured_ingest/processes/connectors/url.py @@ -131,6 +131,42 @@ class UrlDownloaderConfig(DownloaderConfig): # each hop is revalidated. +_NAT64_PREFIX = ipaddress.ip_network("64:ff9b::/96") +_CGNAT = ipaddress.ip_network("100.64.0.0/10") + + +def _embedded_ipv4(v6: ipaddress.IPv6Address) -> "ipaddress.IPv4Address | None": + """The IPv4 an IPv6 transition address embeds (mapped / 6to4 / NAT64 / compat), else None.""" + if v6.ipv4_mapped: + return v6.ipv4_mapped + if v6.sixtofour: + return v6.sixtofour + if v6 in _NAT64_PREFIX: + return ipaddress.IPv4Address(int(v6) & 0xFFFFFFFF) + # IPv4-compatible ::/96 (deprecated), e.g. ::7f00:1 -> 127.0.0.1 (skip :: and ::1) + if int(v6) >> 32 == 0 and (int(v6) & 0xFFFFFFFF) not in (0, 1): + return ipaddress.IPv4Address(int(v6) & 0xFFFFFFFF) + return None + + +def _is_public_ipv4(v4: ipaddress.IPv4Address) -> bool: + # is_global covers private/loopback/etc.; the explicit CGNAT reject makes it + # version-independent (100.64.0.0/10's is_global is only correct on py>=3.11.9/3.12.4). + return v4.is_global and v4 not in _CGNAT + + +def _is_public_address(ip: str) -> bool: + addr = ipaddress.ip_address(ip) + if isinstance(addr, ipaddress.IPv4Address): + return _is_public_ipv4(addr) + # IPv6 must be globally routable AND, if it embeds an IPv4 (NAT64/6to4/mapped/compat), + # that IPv4 must be public too — otherwise it can reach a private/metadata host. + if not addr.is_global: + return False + embedded = _embedded_ipv4(addr) + return embedded is None or _is_public_ipv4(embedded) + + def _validate_and_pin(host: str, allow_private: bool) -> str: """Resolve host, reject if ANY address is non-public, return one pinned IP.""" try: @@ -141,10 +177,9 @@ def _validate_and_pin(host: str, allow_private: bool) -> str: if not ips: raise IngestValueError(f"No addresses for host: {host}") for ip in ips: - # is_global is an allowlist: rejects private/loopback/link-local/reserved/ - # multicast/unspecified AND non-globally-routable ranges a denylist misses - # (e.g. CGNAT 100.64.0.0/10). Skipped when allow_private is set. - if not allow_private and not ipaddress.ip_address(ip).is_global: + # Allowlist: rejects private/loopback/link-local/reserved/multicast/CGNAT and + # IPv6 transition addrs that embed a private/metadata IPv4. Skipped in dev. + if not allow_private and not _is_public_address(ip): raise IngestValueError(f"Refusing non-public address {ip} for host {host}") return ips[0] # deterministic pin; all addresses already validated