feat(connectors): add url source connector

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/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]
 
 ### Fixes

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"]

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

test/unit/connectors/test_url.py

@@ -0,0 +1,246 @@
+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,
+    _safe_filename,
+    _ssrf_safe_download,
+    _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()
+
+
+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(
+        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 --------------------------------------------------------
+@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",
+    [
+        # 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"):
+        _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_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"
+
+
+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"

unstructured_ingest/__version__.py

@@ -1 +1 @@
-__version__ = "1.6.28"  # pragma: no cover
+__version__ = "1.6.29"  # pragma: no cover

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]:

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)