Merge pull request #916 from semantica-agi/security/ssrf-dns-pinning-and-object-iri

security: DNS check-then-use pinning for SSRF fetcher, close object-IRI gap

CHANGELOG.md

@@ -244,6 +244,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Security
 
+- **DNS check-then-use hardening for the ontology URL fetcher, and a remaining object-IRI validation gap** (#916, follow-up to GHSA-8c7v-62gr-hj6g and GHSA-8vgg-8mr4-r236) by @KaifAhmad1
+  - **DNS check-then-use (TOCTOU) window**: GHSA-8c7v-62gr-hj6g's own fix description flagged this as a secondary gap — `_validate_fetch_url()` resolved and validated a hostname once, but `_fetch_url_sync()` then let `requests` resolve the same hostname again independently at connect time. A low-TTL or rebinding DNS answer could differ between the two lookups, reopening the SSRF window the validation exists to close
+  - `_validate_fetch_url()` now returns the validated IP, and a new `_make_pinned_session()` builds a per-hop `requests.Session` whose connection pool is pinned directly to that IP — bypassing DNS resolution for the connection entirely — while explicitly restoring the real hostname as the outgoing HTTP `Host` header and, for HTTPS, the TLS SNI `server_hostname`/`assert_hostname`, so the connection reaches the validated IP but still presents (and is verified against) the real hostname's identity, keeping virtual hosting and certificate validation correct
+  - Caught during implementation: an earlier draft set urllib3's `_dns_host` post-construction, assuming (as in some urllib3 releases) that it was decoupled from `host`. In the version this project installs (2.7.0), `host` is a property that reads/writes `_dns_host` directly, so that approach would have silently changed the Host header too — caught by an end-to-end test against a real local server before landing, rather than shipping. Verified with real (non-mocked) local HTTP and HTTPS servers, the latter using a generated self-signed certificate to prove SNI/cert-hostname verification checks the real hostname rather than the pinned IP, plus a negative control confirming a hostname/cert mismatch is still correctly rejected, not silently bypassed
+  - **Object-IRI validation gap** (GHSA-8vgg-8mr4-r236 follow-up, distinct from the object-branch fix already shipped in #911): a triplet object already wrapped in `<...>` skipped `sparql_escaping.validate_uri()` in both `blazegraph_store.py` and `rdf4j_store.py`'s `_format_object_for_sparql`/`_format_object_for_ntriples`, only checking the inner content for a literal space or `>` — the pre-wrapped and unwrapped branches now validate identically
+  - New `tests/explorer/test_ontology_dns_pinning.py` (6 tests, 4 against real local servers including 2 real-TLS checks, gracefully skipped without the optional `cryptography` package); updated `tests/explorer/test_ontology_ssrf.py` for the new per-hop session construction; 4 new tests in `tests/triplet_store/test_sparql_injection.py` for the object-IRI fix. Full `explorer` + `triplet_store` suite: 566 passed
+
 - **SPARQL injection via unvalidated triplet IRIs** (#911, GHSA-8vgg-8mr4-r236) by @KaifAhmad1
   - `Triplet.subject`/`.predicate` (and, in some builders, `.object`) were interpolated directly into SPARQL update/query strings in the Blazegraph and RDF4J stores, and into a SELECT filter in the Jena store. A subject containing `>` closes the `<...>` IRI token early, so the rest of the value is parsed as more SPARQL. Entity names are document text in the normal ingest pipeline, so anyone whose content gets processed could append operations like `CLEAR ALL`, running with the application's store credentials
   - Applied the existing `sparql_escaping.validate_uri` (already used by `anzo_store.py`, the one backend that was already hardened — this generalizes its approach rather than inventing a new one) at every subject/predicate/object interpolation site: `blazegraph_store.py`'s `_build_insert_data`, `_triplets_to_rdf`, `bulk_load`'s `graph` option, `get_triplets`'s filter, and `delete_triplet`; `rdf4j_store.py`'s `_triplets_to_ntriples`, `get_triplets`'s filter, and `delete_triplet`; `jena_store.py`'s `get_triplets`'s filter (the only vulnerable site there — `add_triplets`/`delete_triplet` already use rdflib's native `Graph.add`/`.remove` with `URIRef` rather than building query strings)

semantica/explorer/routes/ontology.py

@@ -978,8 +978,18 @@ def _normalize_format(fmt: Optional[str]) -> str:
     return _FORMAT_ALIASES.get(lower, lower)
 
 
-def _validate_fetch_url(url: str) -> None:
-    """Reject non-HTTP(S) schemes and private/loopback/link-local targets."""
+def _validate_fetch_url(url: str) -> List[str]:
+    """Reject non-HTTP(S) schemes and private/loopback/link-local targets.
+
+    Returns every resolved, validated IP address (deduplicated, in
+    resolution order) so the caller can pin the actual connection to them
+    (see _make_pinned_session) with fallback across all of them — not just
+    the first — since a hostname can have multiple A/AAAA records and the
+    first one isn't guaranteed reachable. Resolving the hostname again at
+    connect time would open a DNS check-then-use window (a low-TTL or
+    rebinding DNS answer could differ between this check and the client's
+    own lookup), which is what pinning to these specific addresses avoids.
+    """
     parsed = urlparse(url)
     if parsed.scheme not in ("http", "https"):
         raise HTTPException(status_code=422, detail="Only http and https URLs are allowed.")
@@ -990,6 +1000,7 @@ def _validate_fetch_url(url: str) -> None:
         addrinfos = socket.getaddrinfo(hostname, None)
     except socket.gaierror as exc:
         raise HTTPException(status_code=422, detail=f"Cannot resolve hostname '{hostname}': {exc}") from exc
+    validated_ips: List[str] = []
     for _family, _type, _proto, _canonname, sockaddr in addrinfos:
         try:
             ip = ipaddress.ip_address(sockaddr[0])
@@ -1000,46 +1011,160 @@ def _validate_fetch_url(url: str) -> None:
                 status_code=422,
                 detail="Fetching from private, loopback, or reserved network addresses is not allowed.",
             )
+        if sockaddr[0] not in validated_ips:
+            validated_ips.append(sockaddr[0])
+    if not validated_ips:
+        raise HTTPException(status_code=422, detail=f"Cannot resolve hostname '{hostname}' to a usable address.")
+    return validated_ips
+
+
+def _make_pinned_session(pinned_ips: List[str], url: str):
+    """Build a requests.Session whose connection is pinned to pinned_ips
+    (tried in order, falling back on connection failure), regardless of
+    what url's hostname resolves to at connect time.
+
+    _validate_fetch_url() resolves and validates the hostname once; letting
+    the HTTP client resolve it again independently at connect time reopens
+    the exact gap that validation exists to close — a low-TTL or rebinding
+    DNS answer can differ between the two lookups. This pins the pool's
+    connect target to the already-validated addresses directly (bypassing
+    DNS resolution for the connection entirely), while keeping the original
+    hostname as the outgoing HTTP Host header and, for HTTPS, the TLS SNI
+    server_hostname / assert_hostname — otherwise the connection would
+    reach the right IP but present the wrong identity, breaking name-based
+    virtual hosting and (for HTTPS) certificate hostname verification.
+
+    Falls back across every validated address (not just the first) so a
+    hostname with multiple A/AAAA records doesn't fail outright just
+    because the first-returned address happens to be unreachable.
+
+    Note: urllib3's Connection.host is a property that reads/writes the
+    same underlying value as `_dns_host` in this version — it is NOT the
+    separate "presented identity" field it is in some older releases, so
+    overriding just `_dns_host` post-construction (as an earlier version of
+    this fix did) actually changes the Host header too. Pinning the pool's
+    `host` directly and restoring the real hostname via an explicit Host
+    header (+ SNI params for HTTPS) is the correct mechanism here.
+    """
+    import requests as _req
+    import urllib3.util.connection as _u3_connection
+    from urllib3.exceptions import NewConnectionError
+
+    parsed = urlparse(url)
+    hostname = parsed.hostname
+    port = parsed.port
+    default_port = 443 if parsed.scheme == "https" else 80
+    host_header = hostname if port in (None, default_port) else f"{hostname}:{port}"
+
+    class _MultiIPConnectionMixin:
+        """Overrides _new_conn to fall back across every pinned IP in
+        order, instead of urllib3's default single-host connect."""
+
+        def _new_conn(self):
+            last_exc: Optional[BaseException] = None
+            for ip in pinned_ips:
+                try:
+                    return _u3_connection.create_connection(
+                        (ip, self.port),
+                        self.timeout,
+                        source_address=self.source_address,
+                        socket_options=self.socket_options,
+                    )
+                except OSError as exc:
+                    last_exc = exc
+                    continue
+            raise NewConnectionError(
+                self, f"Failed to establish a connection to any of {pinned_ips}: {last_exc}"
+            )
+
+    class _PinnedIPHTTPAdapter(_req.adapters.HTTPAdapter):
+        def get_connection_with_tls_context(self, request, verify, proxies=None, cert=None):
+            # A proxy would perform its own DNS resolution of the target
+            # host on this process's behalf — a resolution outside this
+            # process's visibility or control, so there is no client-side
+            # pin that closes that race. Proxies are disabled outright for
+            # this SSRF-sensitive fetcher (session.trust_env=False below),
+            # so this should be unreachable via environment proxies; fail
+            # closed rather than silently skip pinning if a proxy is
+            # somehow still configured (e.g. passed explicitly in the
+            # future). _validate_fetch_url's destination classification is
+            # a separate, always-enforced check — this only guards the
+            # secondary DNS-pinning hardening.
+            if _req.utils.select_proxy(request.url, proxies):
+                raise HTTPException(
+                    status_code=502,
+                    detail="Proxied requests are not supported for ontology URL fetching.",
+                )
+            host_params, pool_kwargs = self.build_connection_pool_key_attributes(request, verify, cert)
+            if host_params.get("scheme") == "https":
+                pool_kwargs.setdefault("assert_hostname", hostname)
+                pool_kwargs.setdefault("server_hostname", hostname)
+            host_params["host"] = pinned_ips[0]
+            pool = self.poolmanager.connection_from_host(**host_params, pool_kwargs=pool_kwargs)
+            base_connection_cls = pool.ConnectionCls
+            if not issubclass(base_connection_cls, _MultiIPConnectionMixin):
+                pool.ConnectionCls = type(
+                    "_PinnedConnection", (_MultiIPConnectionMixin, base_connection_cls), {}
+                )
+            return pool
+
+    session = _req.Session()
+    # Never honor HTTP_PROXY/HTTPS_PROXY/NO_PROXY env vars for this
+    # SSRF-sensitive fetcher: a configured proxy would perform its own DNS
+    # resolution of the target host outside this process's control,
+    # silently reopening the DNS check-then-use race pinning exists to
+    # close. See _PinnedIPHTTPAdapter.get_connection_with_tls_context for
+    # the fail-closed backstop if a proxy is somehow still configured.
+    session.trust_env = False
+    session.headers["Host"] = host_header
+    adapter = _PinnedIPHTTPAdapter()
+    session.mount("http://", adapter)
+    session.mount("https://", adapter)
+    return session
 
 
 def _fetch_url_sync(url: str) -> bytes:
-    _validate_fetch_url(url)
-    import requests as _req
+    pinned_ips = _validate_fetch_url(url)
     _MAX_REDIRECTS = 5
     current_url = url
     try:
         for _ in range(_MAX_REDIRECTS + 1):
-            resp = _req.get(
-                current_url,
-                headers={"Accept": "text/turtle, application/rdf+xml, application/ld+json, */*;q=0.1"},
-                timeout=30,
-                stream=True,
-                allow_redirects=False,  # SECURITY: follow redirects manually
-            )
-            if resp.is_redirect or resp.is_permanent_redirect:
-                redirect_url = resp.headers.get("Location")
-                resp.close()  # Release the streamed connection before following the redirect
-                if not redirect_url:
-                    raise HTTPException(status_code=502, detail="Redirect without Location header.")
-                # Resolve relative redirects (e.g. /ontology.ttl) against the current URL
-                redirect_url = urljoin(current_url, redirect_url)
-                # Re-validate the redirect target to prevent SSRF via
-                # open-redirect to internal/cloud-metadata endpoints.
-                _validate_fetch_url(redirect_url)
-                current_url = redirect_url
-                continue
+            session = _make_pinned_session(pinned_ips, current_url)
             try:
-                resp.raise_for_status()
-                chunks: List[bytes] = []
-                total = 0
-                for chunk in resp.iter_content(65536):
-                    total += len(chunk)
-                    if total > _MAX_FETCH_BYTES:
-                        raise HTTPException(status_code=413, detail="Remote resource exceeds 20 MB limit.")
-                    chunks.append(chunk)
-                return b"".join(chunks)
+                resp = session.get(
+                    current_url,
+                    headers={"Accept": "text/turtle, application/rdf+xml, application/ld+json, */*;q=0.1"},
+                    timeout=30,
+                    stream=True,
+                    allow_redirects=False,  # SECURITY: follow redirects manually
+                )
+                if resp.is_redirect or resp.is_permanent_redirect:
+                    redirect_url = resp.headers.get("Location")
+                    resp.close()  # Release the streamed connection before following the redirect
+                    if not redirect_url:
+                        raise HTTPException(status_code=502, detail="Redirect without Location header.")
+                    # Resolve relative redirects (e.g. /ontology.ttl) against the current URL
+                    redirect_url = urljoin(current_url, redirect_url)
+                    # Re-validate the redirect target to prevent SSRF via
+                    # open-redirect to internal/cloud-metadata endpoints, and
+                    # get fresh pins for the new host.
+                    pinned_ips = _validate_fetch_url(redirect_url)
+                    current_url = redirect_url
+                    continue
+                try:
+                    resp.raise_for_status()
+                    chunks: List[bytes] = []
+                    total = 0
+                    for chunk in resp.iter_content(65536):
+                        total += len(chunk)
+                        if total > _MAX_FETCH_BYTES:
+                            raise HTTPException(status_code=413, detail="Remote resource exceeds 20 MB limit.")
+                        chunks.append(chunk)
+                    return b"".join(chunks)
+                finally:
+                    resp.close()  # Release the streamed connection once fully read (or on error)
             finally:
-                resp.close()  # Release the streamed connection once fully read (or on error)
+                session.close()
         raise HTTPException(status_code=502, detail=f"Too many redirects (max {_MAX_REDIRECTS}).")
     except HTTPException:
         raise

semantica/triplet_store/blazegraph_store.py

@@ -391,10 +391,12 @@ class BlazegraphStore:
 
         if self._is_uri_value(obj):
             if obj.startswith("<") and obj.endswith(">"):
-                inner = obj[1:-1]
-                if " " in inner or ">" in inner:
-                    raise ValueError(f"IRI contains invalid characters: {obj!r}")
-                return obj
+                # Validate the inner IRI with the same disallowed-character
+                # set as the unwrapped branch below — a narrower ad-hoc
+                # check here previously let a pre-wrapped object bypass
+                # validate_uri() entirely (GHSA-8vgg-8mr4-r236 follow-up).
+                inner = sparql_escaping.validate_uri(obj[1:-1])
+                return f"<{inner}>"
             validated_obj = sparql_escaping.validate_uri(obj)
             return f"<{validated_obj}>"
 

semantica/triplet_store/rdf4j_store.py

@@ -539,10 +539,12 @@ class RDF4JStore:
 
         if self._is_uri_value(obj):
             if obj.startswith("<") and obj.endswith(">"):
-                inner = obj[1:-1]
-                if " " in inner or ">" in inner:
-                    raise ValueError(f"IRI contains invalid characters: {obj!r}")
-                return obj
+                # Validate the inner IRI with the same disallowed-character
+                # set as the unwrapped branch below — a narrower ad-hoc
+                # check here previously let a pre-wrapped object bypass
+                # validate_uri() entirely (GHSA-8vgg-8mr4-r236 follow-up).
+                inner = sparql_escaping.validate_uri(obj[1:-1])
+                return f"<{inner}>"
             validated_obj = sparql_escaping.validate_uri(obj)
             return f"<{validated_obj}>"
 

tests/explorer/test_ontology_dns_pinning.py

@@ -0,0 +1,392 @@
+"""Regression tests for DNS check-then-use (TOCTOU) hardening in the
+ontology URL fetcher (GHSA-8c7v-62gr-hj6g's secondary "smaller" gap).
+
+`_validate_fetch_url` resolves and validates a hostname once; if the actual
+HTTP client resolved it again independently at connect time, a low-TTL or
+rebinding DNS answer could differ between the two lookups, reopening the
+SSRF window the validation exists to close. `_make_pinned_session` closes
+this by pinning the connection pool's `host` directly to the already-
+validated IP (bypassing DNS resolution for the connection entirely), while
+explicitly restoring the real hostname as the outgoing HTTP `Host` header
+and, for HTTPS, the TLS SNI `server_hostname` / `assert_hostname` — so the
+connection reaches the pinned IP but still presents (and verifies against)
+the original hostname's identity.
+
+test_ontology_ssrf.py covers the redirect-handling logic around this with
+mocks; this file proves the pinning mechanism itself works end-to-end
+against real local servers, with no DNS mocking at all — the test hostname
+is never resolved, which is exactly the property being verified. It also
+includes a negative control (mismatched cert hostname) proving TLS
+verification is genuinely enforced against the real hostname, not silently
+bypassed or checked against the pinned IP instead.
+"""
+
+import http.server
+import socket
+import threading
+
+import pytest
+
+from semantica.explorer.routes import ontology as ontology_mod
+
+
+def _start_local_server():
+    captured = {}
+
+    class Handler(http.server.BaseHTTPRequestHandler):
+        def do_GET(self):
+            captured["host_header"] = self.headers.get("Host")
+            body = b"pinned response"
+            self.send_response(200)
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def log_message(self, *_args):
+            pass
+
+    server = http.server.HTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    return server, thread, captured
+
+
+def test_pinned_session_connects_to_pinned_ip_without_resolving_hostname():
+    """A session built by _make_pinned_session must reach the pinned IP
+    directly. The request URL uses a hostname that cannot be resolved via
+    real DNS ('.invalid' is reserved by RFC 2606) — if pinning weren't
+    working, this request would fail with a name-resolution error instead
+    of reaching the local server, since nothing else could route it there.
+    """
+    server, thread, captured = _start_local_server()
+    port = server.server_address[1]
+    url = f"http://pinned-test.invalid:{port}/resource"
+    try:
+        session = ontology_mod._make_pinned_session(["127.0.0.1"], url)
+        try:
+            resp = session.get(url, timeout=5)
+            assert resp.status_code == 200
+            assert resp.content == b"pinned response"
+        finally:
+            session.close()
+    finally:
+        server.shutdown()
+        thread.join(timeout=2)
+
+    # Host header must still be the original hostname, not the pinned IP —
+    # proving connection target and presented identity are decoupled
+    # correctly (this is what keeps virtual hosting / TLS SNI correct).
+    assert captured["host_header"] == f"pinned-test.invalid:{port}"
+
+
+def test_pinned_session_ignores_a_different_real_resolution():
+    """Even if the hostname *does* resolve to something else via real DNS,
+    the pinned session must still go to the pinned IP — this is the actual
+    TOCTOU property: the connection uses what was validated, not whatever
+    a fresh lookup returns. 'localhost' reliably resolves to a loopback
+    address, which is deliberately NOT where our test server listens on
+    (127.0.0.1 specifically) — but since Windows/most stacks map
+    'localhost' to 127.0.0.1 too, use a distinct high loopback address
+    (127.0.0.2) for the server so a real 'localhost' resolution (127.0.0.1)
+    provably would NOT reach it, isolating the assertion to pinning alone.
+    """
+    captured = {}
+
+    class Handler(http.server.BaseHTTPRequestHandler):
+        def do_GET(self):
+            captured["host_header"] = self.headers.get("Host")
+            body = b"pinned via explicit ip"
+            self.send_response(200)
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def log_message(self, *_args):
+            pass
+
+    try:
+        server = http.server.HTTPServer(("127.0.0.2", 0), Handler)
+    except OSError:
+        # 127.0.0.2 isn't bindable in this environment (uncommon, but
+        # possible in some sandboxes) — skip rather than false-fail.
+        import pytest
+        pytest.skip("127.0.0.2 is not bindable in this environment")
+
+    port = server.server_address[1]
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    url = f"http://localhost:{port}/resource"
+    try:
+        session = ontology_mod._make_pinned_session(["127.0.0.2"], url)
+        try:
+            resp = session.get(url, timeout=5)
+            assert resp.status_code == 200
+            assert resp.content == b"pinned via explicit ip"
+        finally:
+            session.close()
+    finally:
+        server.shutdown()
+        thread.join(timeout=2)
+
+    assert captured["host_header"] == f"localhost:{port}"
+
+
+def test_pinned_session_falls_back_across_multiple_pinned_ips():
+    """A hostname can have multiple A/AAAA records; pinning to only the
+    first-returned address means a fetch fails outright if that specific
+    address happens to be unreachable even though a later one would work.
+    _make_pinned_session must fall back through every pinned IP in order.
+    """
+    server, thread, captured = _start_local_server()
+    port = server.server_address[1]
+    url = f"http://pinned-test.invalid:{port}/resource"
+    # 127.0.0.3 has nothing listening on this port — connection refused,
+    # forcing a fallback to the second (real) address.
+    unreachable_ip = "127.0.0.3"
+    try:
+        session = ontology_mod._make_pinned_session([unreachable_ip, "127.0.0.1"], url)
+        try:
+            resp = session.get(url, timeout=5)
+            assert resp.status_code == 200
+            assert resp.content == b"pinned response"
+        finally:
+            session.close()
+    finally:
+        server.shutdown()
+        thread.join(timeout=2)
+
+
+def test_pinned_session_raises_when_every_pinned_ip_is_unreachable():
+    """If none of the pinned IPs are reachable, the session must raise
+    rather than silently falling back to resolving the hostname itself
+    (which would reopen the exact TOCTOU window pinning exists to close)."""
+    import requests
+
+    url = "http://pinned-test.invalid:9/resource"  # port 9 (discard) — nothing listens
+    session = ontology_mod._make_pinned_session(["127.0.0.3", "127.0.0.4"], url)
+    try:
+        with pytest.raises(requests.exceptions.ConnectionError):
+            session.get(url, timeout=5)
+    finally:
+        session.close()
+
+
+def test_pinned_session_disables_environment_proxy_trust():
+    """A pinned session must never honor HTTP_PROXY/HTTPS_PROXY env vars —
+    a proxy would perform its own DNS resolution of the target host outside
+    this process's control, reopening the exact TOCTOU window pinning
+    exists to close."""
+    session = ontology_mod._make_pinned_session(["127.0.0.1"], "http://example.org/")
+    try:
+        assert session.trust_env is False
+    finally:
+        session.close()
+
+
+def test_pinned_session_ignores_env_proxy_and_connects_directly(monkeypatch):
+    """End-to-end: even with HTTP_PROXY pointed at an address that would
+    fail if contacted, a pinned session must reach the real local server
+    directly — proving the env var is genuinely not consulted, not just
+    that the trust_env flag is set."""
+    monkeypatch.setenv("HTTP_PROXY", "http://127.0.0.5:1/")  # would fail if ever used
+    server, thread, _captured = _start_local_server()
+    port = server.server_address[1]
+    url = f"http://pinned-test.invalid:{port}/resource"
+    try:
+        session = ontology_mod._make_pinned_session(["127.0.0.1"], url)
+        try:
+            resp = session.get(url, timeout=5)
+            assert resp.status_code == 200
+            assert resp.content == b"pinned response"
+        finally:
+            session.close()
+    finally:
+        server.shutdown()
+        thread.join(timeout=2)
+
+
+def test_pinned_session_fails_closed_if_a_proxy_is_explicitly_forced():
+    """Backstop: if a proxy is somehow still configured on the session
+    despite trust_env=False (e.g. set explicitly, as a future code path
+    might), the adapter must fail closed with a clear error rather than
+    silently connecting through the proxy unpinned."""
+    from fastapi import HTTPException
+
+    session = ontology_mod._make_pinned_session(["127.0.0.1"], "http://example.org/")
+    session.proxies = {"http": "http://127.0.0.5:1"}
+    try:
+        with pytest.raises(HTTPException) as exc_info:
+            session.get("http://example.org/", timeout=5)
+        assert exc_info.value.status_code == 502
+    finally:
+        session.close()
+
+
+def test_validate_fetch_url_returns_the_resolved_ip():
+    """_validate_fetch_url must return every IP it validated, so callers can
+    pin the connection to them (with fallback across all of them)."""
+    def fake_getaddrinfo(host, *_a, **_k):
+        return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0))]
+
+    import unittest.mock as mock
+    with mock.patch.object(ontology_mod.socket, "getaddrinfo", side_effect=fake_getaddrinfo):
+        resolved_ips = ontology_mod._validate_fetch_url("http://example.org/ontology.ttl")
+
+    assert resolved_ips == ["93.184.216.34"]
+
+
+def test_validate_fetch_url_returns_all_validated_ips_deduplicated():
+    """A hostname with multiple A/AAAA records must return every distinct
+    validated address, in resolution order, so the caller can fall back
+    across all of them rather than failing if only the first is
+    unreachable."""
+    def fake_getaddrinfo(host, *_a, **_k):
+        return [
+            (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0)),
+            (socket.AF_INET, socket.SOCK_DGRAM, 17, "", ("93.184.216.34", 0)),  # duplicate, different socktype
+            (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.35", 0)),
+        ]
+
+    import unittest.mock as mock
+    with mock.patch.object(ontology_mod.socket, "getaddrinfo", side_effect=fake_getaddrinfo):
+        resolved_ips = ontology_mod._validate_fetch_url("http://example.org/ontology.ttl")
+
+    assert resolved_ips == ["93.184.216.34", "93.184.216.35"]
+
+
+def test_validate_fetch_url_still_rejects_private_ip():
+    """Confirm the pinning refactor didn't loosen the original address
+    classification — a hostname resolving to a private/internal address
+    must still be rejected before any IP is returned."""
+    import ipaddress
+    import unittest.mock as mock
+
+    import pytest
+    from fastapi import HTTPException
+
+    def fake_getaddrinfo(host, *_a, **_k):
+        return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("169.254.169.254", 0))]
+
+    with mock.patch.object(ontology_mod.socket, "getaddrinfo", side_effect=fake_getaddrinfo):
+        with pytest.raises(HTTPException) as exc_info:
+            ontology_mod._validate_fetch_url("http://attacker.example/ontology.ttl")
+
+    assert exc_info.value.status_code == 422
+
+
+# ---------------------------------------------------------------------------
+# HTTPS: SNI + certificate hostname verification must use the real hostname,
+# not the pinned IP — this is the highest-risk part of pinning to get wrong,
+# since a mistake here could silently weaken TLS verification rather than
+# just breaking connectivity. Requires the optional `cryptography` package
+# to mint a throwaway self-signed cert; skipped gracefully without it.
+# ---------------------------------------------------------------------------
+
+def _make_self_signed_cert(hostname: str, tmp_path):
+    import datetime
+
+    pytest.importorskip("cryptography")
+    from cryptography import x509
+    from cryptography.hazmat.primitives import hashes, serialization
+    from cryptography.hazmat.primitives.asymmetric import rsa
+    from cryptography.x509.oid import NameOID
+
+    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, hostname)])
+    now = datetime.datetime.now(datetime.timezone.utc)
+    cert = (
+        x509.CertificateBuilder()
+        .subject_name(name)
+        .issuer_name(name)
+        .public_key(key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(now - datetime.timedelta(days=1))
+        .not_valid_after(now + datetime.timedelta(days=1))
+        .add_extension(x509.SubjectAlternativeName([x509.DNSName(hostname)]), critical=False)
+        .sign(key, hashes.SHA256())
+    )
+
+    cert_path = tmp_path / "cert.pem"
+    key_path = tmp_path / "key.pem"
+    cert_path.write_bytes(cert.public_bytes(serialization.Encoding.PEM))
+    key_path.write_bytes(
+        key.private_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PrivateFormat.TraditionalOpenSSL,
+            encryption_algorithm=serialization.NoEncryption(),
+        )
+    )
+    return str(cert_path), str(key_path)
+
+
+def _start_local_https_server(cert_path, key_path):
+    import ssl
+
+    captured = {}
+
+    class Handler(http.server.BaseHTTPRequestHandler):
+        def do_GET(self):
+            captured["host_header"] = self.headers.get("Host")
+            body = b"tls pinned response"
+            self.send_response(200)
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def log_message(self, *_args):
+            pass
+
+    server = http.server.HTTPServer(("127.0.0.1", 0), Handler)
+    ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+    ssl_ctx.minimum_version = ssl.TLSVersion.TLSv1_2
+    ssl_ctx.load_cert_chain(cert_path, key_path)
+    server.socket = ssl_ctx.wrap_socket(server.socket, server_side=True)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    return server, thread, captured
+
+
+def test_pinned_https_session_verifies_against_real_hostname_not_pinned_ip(tmp_path):
+    """A pinned HTTPS connection must present + verify SNI/cert against the
+    real hostname, even though the socket connects to the pinned IP. The
+    cert's SAN is the hostname, never '127.0.0.1' — if pinning verified
+    against the IP instead (or against nothing), this would either fail
+    for the wrong reason or silently succeed with no real verification."""
+    cert_path, key_path = _make_self_signed_cert("pinned-tls-test.invalid", tmp_path)
+    server, thread, captured = _start_local_https_server(cert_path, key_path)
+    port = server.server_address[1]
+    url = f"https://pinned-tls-test.invalid:{port}/resource"
+    try:
+        session = ontology_mod._make_pinned_session(["127.0.0.1"], url)
+        try:
+            resp = session.get(url, timeout=5, verify=cert_path)
+        finally:
+            session.close()
+    finally:
+        server.shutdown()
+        thread.join(timeout=2)
+
+    assert resp.status_code == 200
+    assert resp.content == b"tls pinned response"
+    assert captured["host_header"] == f"pinned-tls-test.invalid:{port}"
+
+
+def test_pinned_https_session_rejects_hostname_mismatch(tmp_path):
+    """Negative control: requesting a hostname that does NOT match the
+    cert's SAN must still fail verification — proving pinning doesn't
+    silently bypass or misdirect certificate hostname checking."""
+    cert_path, key_path = _make_self_signed_cert("pinned-tls-test.invalid", tmp_path)
+    server, thread, _captured = _start_local_https_server(cert_path, key_path)
+    port = server.server_address[1]
+    url = f"https://wrong-name.invalid:{port}/resource"
+    try:
+        session = ontology_mod._make_pinned_session(["127.0.0.1"], url)
+        try:
+            import requests
+            with pytest.raises(requests.exceptions.SSLError):
+                session.get(url, timeout=5, verify=cert_path)
+        finally:
+            session.close()
+    finally:
+        server.shutdown()
+        thread.join(timeout=2)

tests/explorer/test_ontology_ssrf.py

@@ -3,12 +3,17 @@
 `_fetch_url_sync` disables `requests`' automatic redirect following and
 re-validates every hop with `_validate_fetch_url` (see GHSA-8c7v-62gr-hj6g:
 unvalidated redirect targets previously let a public first hop 302 the
-server into fetching cloud metadata / loopback services).
+server into fetching cloud metadata / loopback services). It also pins each
+hop's connection to the IP `_validate_fetch_url` already resolved and
+validated, via `_make_pinned_session`, closing the DNS check-then-use gap
+between that validation and the client's own (potentially different) lookup.
 
 These tests cover the redirect-handling logic itself: relative `Location`
 headers must resolve correctly instead of being rejected outright, redirect
 targets that resolve to private/loopback addresses must still be blocked,
 and every response must be closed (no leaked connections across hops).
+`test_ontology_dns_pinning.py` covers the pinning mechanism
+(`_make_pinned_session`, `_validate_fetch_url`'s returned IP) directly.
 """
 
 import socket
@@ -37,6 +42,17 @@ def _make_response(is_redirect=False, is_permanent=False, location=None, body=b"
     return resp
 
 
+def _patch_session(responses):
+    """Patch _make_pinned_session so _fetch_url_sync's session.get(...)
+    calls return the given responses in order, without touching the real
+    requests.Session/pinning machinery (that's covered by test_pinning.py).
+    """
+    fake_session = MagicMock()
+    fake_session.get = MagicMock(side_effect=responses)
+    fake_session.close = MagicMock()
+    return patch.object(ontology_mod, "_make_pinned_session", return_value=fake_session), fake_session
+
+
 @patch.object(ontology_mod.socket, "getaddrinfo", side_effect=_fake_getaddrinfo)
 def test_relative_redirect_location_is_resolved(mock_getaddrinfo):
     """A relative Location header (e.g. '/ontology.ttl') must resolve against
@@ -44,11 +60,12 @@ def test_relative_redirect_location_is_resolved(mock_getaddrinfo):
     redirect_resp = _make_response(is_redirect=True, location="/ontology.ttl")
     final_resp = _make_response(body=b"final content")
 
-    with patch("requests.get", side_effect=[redirect_resp, final_resp]) as mock_get:
+    patcher, fake_session = _patch_session([redirect_resp, final_resp])
+    with patcher:
         result = ontology_mod._fetch_url_sync("http://example.org/start")
 
     assert result == b"final content"
-    second_call_url = mock_get.call_args_list[1].args[0]
+    second_call_url = fake_session.get.call_args_list[1].args[0]
     assert second_call_url == "http://example.org/ontology.ttl"
     redirect_resp.close.assert_called_once()
     final_resp.close.assert_called_once()
@@ -67,7 +84,8 @@ def test_redirect_to_private_ip_is_rejected(mock_getaddrinfo):
     mock_getaddrinfo.side_effect = getaddrinfo_side_effect
     redirect_resp = _make_response(is_redirect=True, location="http://internal.example/latest/meta-data/")
 
-    with patch("requests.get", side_effect=[redirect_resp]):
+    patcher, fake_session = _patch_session([redirect_resp])
+    with patcher:
         with pytest.raises(ontology_mod.HTTPException) as exc_info:
             ontology_mod._fetch_url_sync("http://example.org/start")
 
@@ -78,7 +96,8 @@ def test_redirect_to_private_ip_is_rejected(mock_getaddrinfo):
 @patch.object(ontology_mod.socket, "getaddrinfo", side_effect=_fake_getaddrinfo)
 def test_final_response_is_closed(mock_getaddrinfo):
     final_resp = _make_response(body=b"content")
-    with patch("requests.get", side_effect=[final_resp]):
+    patcher, _fake_session = _patch_session([final_resp])
+    with patcher:
         ontology_mod._fetch_url_sync("http://example.org/start")
     final_resp.close.assert_called_once()
 
@@ -86,7 +105,8 @@ def test_final_response_is_closed(mock_getaddrinfo):
 @patch.object(ontology_mod.socket, "getaddrinfo", side_effect=_fake_getaddrinfo)
 def test_redirect_chain_exceeding_cap_is_rejected(mock_getaddrinfo):
     responses = [_make_response(is_redirect=True, location=f"/hop{i}") for i in range(10)]
-    with patch("requests.get", side_effect=responses):
+    patcher, _fake_session = _patch_session(responses)
+    with patcher:
         with pytest.raises(ontology_mod.HTTPException) as exc_info:
             ontology_mod._fetch_url_sync("http://example.org/start")
     assert exc_info.value.status_code == 502

tests/triplet_store/test_sparql_injection.py

@@ -82,6 +82,22 @@ class TestBlazegraphSparqlInjection(unittest.TestCase):
         self.assertIn("<http://s> <http://p>", insert_data)
         self.assertNotIn("CLEAR ALL", insert_data)
 
+    def test_format_object_rejects_malicious_pre_wrapped_iri(self):
+        """A caller-supplied object already wrapped in '<...>' must still be
+        fully validated, not just checked for a literal space/'>' — a
+        narrower ad-hoc check here previously let this branch bypass
+        validate_uri() entirely (Codex-flagged follow-up to GHSA-8vgg)."""
+        store = self._make_store()
+        evil_object = f"<{EVIL_SUBJECT}>"
+        triplet = Triplet(subject="http://s", predicate="http://p", object=evil_object)
+        with self.assertRaises(ValidationError):
+            store._format_object_for_sparql(triplet)
+
+    def test_format_object_accepts_legitimate_pre_wrapped_iri(self):
+        store = self._make_store()
+        triplet = Triplet(subject="http://s", predicate="http://p", object="<http://o>")
+        self.assertEqual(store._format_object_for_sparql(triplet), "<http://o>")
+
 
 class TestRDF4JSparqlInjection(unittest.TestCase):
     @patch.object(RDF4JStore, "_connect", autospec=True)
@@ -207,6 +223,20 @@ class TestRDF4JSparqlInjection(unittest.TestCase):
         self.assertIn("<http://s> <http://p>", ntriples)
         self.assertNotIn("CLEAR ALL", ntriples)
 
+    def test_format_object_rejects_malicious_pre_wrapped_iri(self):
+        """Same pre-wrapped-object bypass as Blazegraph, fixed in
+        _format_object_for_ntriples."""
+        store = self._make_store()
+        evil_object = f"<{EVIL_SUBJECT}>"
+        triplet = Triplet(subject="http://s", predicate="http://p", object=evil_object)
+        with self.assertRaises(ValidationError):
+            store._format_object_for_ntriples(triplet)
+
+    def test_format_object_accepts_legitimate_pre_wrapped_iri(self):
+        store = self._make_store()
+        triplet = Triplet(subject="http://s", predicate="http://p", object="<http://o>")
+        self.assertEqual(store._format_object_for_ntriples(triplet), "<http://o>")
+
 
 class TestJenaSparqlInjection(unittest.TestCase):
     def setUp(self):