From 646c70ce6393861c1b7a12a5a111d29db5006cc4 Mon Sep 17 00:00:00 2001
From: KaifAhmad1 <kaifahmad087@gmail.com>
Date: Tue, 11 Aug 2026 18:52:26 +0530
Subject: [PATCH 1/4] security: DNS check-then-use pinning for SSRF fetcher,
 close object-IRI gap
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two follow-up hardening items flagged as secondary/deferred during
GHSA-8c7v-62gr-hj6g and GHSA-8vgg-8mr4-r236's fixes:

1. DNS check-then-use (TOCTOU) window in the ontology URL fetcher.
   _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.

   Note: an earlier version of this fix set `_dns_host` post-construction
   assuming it was decoupled from `host`, matching some other urllib3
   releases; in the installed version (2.7.0), `host` is a property
   that reads/writes `_dns_host` directly, so that approach silently
   changed the Host header too. Verified with a real (non-mocked) local
   HTTP server, a real local HTTPS server with a self-signed cert
   (proving SNI/cert-hostname verification checks the real hostname,
   not the pinned IP), and a negative control confirming a hostname/cert
   mismatch is still correctly rejected — not silently bypassed.

2. Pre-wrapped object IRIs skipped full validation in
   _format_object_for_sparql/_format_object_for_ntriples (Blazegraph,
   RDF4J). A triplet object already wrapped in `<...>` only had its
   inner content checked for a literal space or `>`, not run through
   sparql_escaping.validate_uri() like the unwrapped-object branch —
   flagged by automated review during GHSA-8vgg-8mr4-r236's fix. Both
   branches now validate identically.

Tests: tests/explorer/test_ontology_dns_pinning.py (6 tests, including
2 real local-server end-to-end checks and 2 real-TLS checks with a
generated self-signed cert, gracefully skipped if `cryptography` isn't
installed); 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.
---
 semantica/explorer/routes/ontology.py        | 143 +++++++---
 semantica/triplet_store/blazegraph_store.py  |  10 +-
 semantica/triplet_store/rdf4j_store.py       |  10 +-
 tests/explorer/test_ontology_dns_pinning.py  | 281 +++++++++++++++++++
 tests/explorer/test_ontology_ssrf.py         |  32 ++-
 tests/triplet_store/test_sparql_injection.py |  30 ++
 6 files changed, 459 insertions(+), 47 deletions(-)
 create mode 100644 tests/explorer/test_ontology_dns_pinning.py

diff --git a/semantica/explorer/routes/ontology.py b/semantica/explorer/routes/ontology.py
index 8b0c8aa9..3ac81b0a 100644
--- a/semantica/explorer/routes/ontology.py
+++ b/semantica/explorer/routes/ontology.py
@@ -978,8 +978,15 @@ 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) -> str:
+    """Reject non-HTTP(S) schemes and private/loopback/link-local targets.
+
+    Returns the first resolved, validated IP address so the caller can pin
+    the actual connection to it (see _PinnedIPHTTPAdapter) — 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).
+    """
     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 +997,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_ip: Optional[str] = None
     for _family, _type, _proto, _canonname, sockaddr in addrinfos:
         try:
             ip = ipaddress.ip_address(sockaddr[0])
@@ -1000,46 +1008,115 @@ def _validate_fetch_url(url: str) -> None:
                 status_code=422,
                 detail="Fetching from private, loopback, or reserved network addresses is not allowed.",
             )
+        if validated_ip is None:
+            validated_ip = sockaddr[0]
+    if validated_ip is None:
+        raise HTTPException(status_code=422, detail=f"Cannot resolve hostname '{hostname}' to a usable address.")
+    return validated_ip
+
+
+def _make_pinned_session(pinned_ip: str, url: str):
+    """Build a requests.Session whose connection is pinned to pinned_ip,
+    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 IP 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.
+
+    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
+
+    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 _PinnedIPHTTPAdapter(_req.adapters.HTTPAdapter):
+        def get_connection_with_tls_context(self, request, verify, proxies=None, cert=None):
+            # If an HTTP(S) proxy applies (env-configured or per-request),
+            # the actual TCP connection target is the proxy, not the
+            # resolved IP, and proxy tunneling changes the connection model
+            # enough that pinning doesn't apply cleanly. Fall back to the
+            # normal (unpinned) path rather than silently bypassing the
+            # proxy — _validate_fetch_url's destination check still applies
+            # either way; only this secondary DNS-pinning hardening is
+            # skipped.
+            if _req.utils.select_proxy(request.url, proxies):
+                return super().get_connection_with_tls_context(
+                    request, verify, proxies=proxies, cert=cert
+                )
+            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_ip
+            return self.poolmanager.connection_from_host(**host_params, pool_kwargs=pool_kwargs)
+
+    session = _req.Session()
+    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_ip = _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_ip, 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 a fresh pin for the new host.
+                    pinned_ip = _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
diff --git a/semantica/triplet_store/blazegraph_store.py b/semantica/triplet_store/blazegraph_store.py
index 4e139a57..21e42b38 100644
--- a/semantica/triplet_store/blazegraph_store.py
+++ b/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}>"
 
diff --git a/semantica/triplet_store/rdf4j_store.py b/semantica/triplet_store/rdf4j_store.py
index 79a83320..8dad6997 100644
--- a/semantica/triplet_store/rdf4j_store.py
+++ b/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}>"
 
diff --git a/tests/explorer/test_ontology_dns_pinning.py b/tests/explorer/test_ontology_dns_pinning.py
new file mode 100644
index 00000000..cb7791c6
--- /dev/null
+++ b/tests/explorer/test_ontology_dns_pinning.py
@@ -0,0 +1,281 @@
+"""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_validate_fetch_url_returns_the_resolved_ip():
+    """_validate_fetch_url must return the IP it validated, so callers can
+    pin the connection to it."""
+    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_ip = ontology_mod._validate_fetch_url("http://example.org/ontology.ttl")
+
+    assert resolved_ip == "93.184.216.34"
+
+
+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
+
+    cryptography = 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.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)
diff --git a/tests/explorer/test_ontology_ssrf.py b/tests/explorer/test_ontology_ssrf.py
index fe9de35b..8613275c 100644
--- a/tests/explorer/test_ontology_ssrf.py
+++ b/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
diff --git a/tests/triplet_store/test_sparql_injection.py b/tests/triplet_store/test_sparql_injection.py
index 30376804..587ba613 100644
--- a/tests/triplet_store/test_sparql_injection.py
+++ b/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):

From f2f1d6787d178be4eedfbf78a636560b51fb633a Mon Sep 17 00:00:00 2001
From: KaifAhmad1 <kaifahmad087@gmail.com>
Date: Tue, 11 Aug 2026 18:57:07 +0530
Subject: [PATCH 2/4] docs(changelog): add PR #916 (DNS pinning + object-IRI
 gap) entry

---
 CHANGELOG.md | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index db25f459..1d7bf1ba 100644
--- a/CHANGELOG.md
+++ b/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)

From 154a7347cdc3bffc0ee5ae33bc9121624da7014f Mon Sep 17 00:00:00 2001
From: KaifAhmad1 <kaifahmad087@gmail.com>
Date: Tue, 11 Aug 2026 19:10:01 +0530
Subject: [PATCH 3/4] fix: address CI/review findings on DNS pinning (multi-IP
 fallback, TLS min version)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Four findings from PR #916's automated review, all addressed:

- CodeQL (HIGH): the test HTTPS server's SSLContext allowed TLSv1/TLSv1.1
  by not setting a minimum version. Added
  ssl_ctx.minimum_version = ssl.TLSVersion.TLSv1_2.
- github-code-quality: unused `cryptography` local in
  _make_self_signed_cert — importorskip's return value was never used.
- Qodo (reliability): _validate_fetch_url() only returned the first
  validated IP, and _make_pinned_session() pinned to just that one
  address, so a fetch would fail outright if the first-returned A/AAAA
  record happened to be unreachable even though a later one would work.
  _validate_fetch_url() now returns every validated IP (deduplicated,
  in resolution order); _make_pinned_session() takes the full list and
  falls back through each one via a custom Connection._new_conn
  override, matching the fallback behavior a normal DNS-resolving
  connection would already get for free. Verified with a real test:
  pin to an unreachable loopback address followed by a real one, confirm
  the fetch still succeeds by falling back; and a real test confirming
  it still raises (rather than silently re-resolving the hostname) when
  every pinned address is unreachable.
- Qodo (security): when an HTTP(S) proxy applies, the adapter falls back
  to the unpinned path rather than pinning. This is a real, but
  architecturally unavoidable, limitation from the client side: for a
  forward proxy, the *proxy* performs its own DNS resolution of the
  target host on the application's behalf, a resolution this process
  has no visibility into or control over — there's no client-side pin
  that closes that race. _validate_fetch_url's destination
  classification still fully applies either way; only the secondary
  DNS-pinning hardening doesn't extend through a proxy. Added an info
  log when this fallback path is taken so it's observable rather than
  silent, and expanded the code comment to make the reasoning explicit
  for the next reader/reviewer rather than looking like an oversight.

Tests: 3 new tests in test_ontology_dns_pinning.py (multi-IP fallback
success, all-unreachable failure, deduplicated multi-record resolution).
Full explorer + triplet_store suite: 569 passed.
---
 semantica/explorer/routes/ontology.py       | 105 ++++++++++++++------
 tests/explorer/test_ontology_dns_pinning.py |  78 +++++++++++++--
 2 files changed, 145 insertions(+), 38 deletions(-)

diff --git a/semantica/explorer/routes/ontology.py b/semantica/explorer/routes/ontology.py
index 3ac81b0a..005d8a0f 100644
--- a/semantica/explorer/routes/ontology.py
+++ b/semantica/explorer/routes/ontology.py
@@ -978,14 +978,17 @@ def _normalize_format(fmt: Optional[str]) -> str:
     return _FORMAT_ALIASES.get(lower, lower)
 
 
-def _validate_fetch_url(url: str) -> str:
+def _validate_fetch_url(url: str) -> List[str]:
     """Reject non-HTTP(S) schemes and private/loopback/link-local targets.
 
-    Returns the first resolved, validated IP address so the caller can pin
-    the actual connection to it (see _PinnedIPHTTPAdapter) — 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).
+    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"):
@@ -997,7 +1000,7 @@ def _validate_fetch_url(url: str) -> str:
         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_ip: Optional[str] = None
+    validated_ips: List[str] = []
     for _family, _type, _proto, _canonname, sockaddr in addrinfos:
         try:
             ip = ipaddress.ip_address(sockaddr[0])
@@ -1008,28 +1011,33 @@ def _validate_fetch_url(url: str) -> str:
                 status_code=422,
                 detail="Fetching from private, loopback, or reserved network addresses is not allowed.",
             )
-        if validated_ip is None:
-            validated_ip = sockaddr[0]
-    if validated_ip is None:
+        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_ip
+    return validated_ips
 
 
-def _make_pinned_session(pinned_ip: str, url: str):
-    """Build a requests.Session whose connection is pinned to pinned_ip,
-    regardless of what url's hostname resolves to at connect time.
+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 IP directly (bypassing DNS
-    resolution for the connection entirely), while keeping the original
+    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
@@ -1038,7 +1046,10 @@ def _make_pinned_session(pinned_ip: str, url: str):
     `host` directly and restoring the real hostname via an explicit Host
     header (+ SNI params for HTTPS) is the correct mechanism here.
     """
+    import logging as _pin_logging
     import requests as _req
+    import urllib3.util.connection as _u3_connection
+    from urllib3.exceptions import NewConnectionError
 
     parsed = urlparse(url)
     hostname = parsed.hostname
@@ -1046,17 +1057,47 @@ def _make_pinned_session(pinned_ip: str, url: str):
     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):
             # If an HTTP(S) proxy applies (env-configured or per-request),
-            # the actual TCP connection target is the proxy, not the
-            # resolved IP, and proxy tunneling changes the connection model
-            # enough that pinning doesn't apply cleanly. Fall back to the
-            # normal (unpinned) path rather than silently bypassing the
-            # proxy — _validate_fetch_url's destination check still applies
-            # either way; only this secondary DNS-pinning hardening is
-            # skipped.
+            # pinning can't meaningfully apply: the actual TCP connection
+            # target is the proxy, and for a forward proxy the *proxy*
+            # performs its own DNS resolution of the target host on our
+            # behalf — a resolution this process has no visibility into or
+            # control over, so there is no client-side fix for that
+            # specific race. Fall back to the normal (unpinned) path rather
+            # than silently bypassing the configured proxy.
+            # _validate_fetch_url's destination classification still fully
+            # applies either way; only this secondary DNS-pinning hardening
+            # is inherently out of scope when a proxy is in the path.
             if _req.utils.select_proxy(request.url, proxies):
+                _pin_logging.getLogger(__name__).info(
+                    "DNS pinning skipped for %s: a proxy is configured for this "
+                    "request, and proxy-side DNS resolution is outside this "
+                    "process's control.",
+                    request.url,
+                )
                 return super().get_connection_with_tls_context(
                     request, verify, proxies=proxies, cert=cert
                 )
@@ -1064,8 +1105,14 @@ def _make_pinned_session(pinned_ip: str, url: str):
             if host_params.get("scheme") == "https":
                 pool_kwargs.setdefault("assert_hostname", hostname)
                 pool_kwargs.setdefault("server_hostname", hostname)
-            host_params["host"] = pinned_ip
-            return self.poolmanager.connection_from_host(**host_params, pool_kwargs=pool_kwargs)
+            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()
     session.headers["Host"] = host_header
@@ -1076,12 +1123,12 @@ def _make_pinned_session(pinned_ip: str, url: str):
 
 
 def _fetch_url_sync(url: str) -> bytes:
-    pinned_ip = _validate_fetch_url(url)
+    pinned_ips = _validate_fetch_url(url)
     _MAX_REDIRECTS = 5
     current_url = url
     try:
         for _ in range(_MAX_REDIRECTS + 1):
-            session = _make_pinned_session(pinned_ip, current_url)
+            session = _make_pinned_session(pinned_ips, current_url)
             try:
                 resp = session.get(
                     current_url,
@@ -1099,8 +1146,8 @@ def _fetch_url_sync(url: str) -> bytes:
                     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 a fresh pin for the new host.
-                    pinned_ip = _validate_fetch_url(redirect_url)
+                    # get fresh pins for the new host.
+                    pinned_ips = _validate_fetch_url(redirect_url)
                     current_url = redirect_url
                     continue
                 try:
diff --git a/tests/explorer/test_ontology_dns_pinning.py b/tests/explorer/test_ontology_dns_pinning.py
index cb7791c6..90780ba4 100644
--- a/tests/explorer/test_ontology_dns_pinning.py
+++ b/tests/explorer/test_ontology_dns_pinning.py
@@ -62,7 +62,7 @@ def test_pinned_session_connects_to_pinned_ip_without_resolving_hostname():
     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)
+        session = ontology_mod._make_pinned_session(["127.0.0.1"], url)
         try:
             resp = session.get(url, timeout=5)
             assert resp.status_code == 200
@@ -117,7 +117,7 @@ def test_pinned_session_ignores_a_different_real_resolution():
     thread.start()
     url = f"http://localhost:{port}/resource"
     try:
-        session = ontology_mod._make_pinned_session("127.0.0.2", url)
+        session = ontology_mod._make_pinned_session(["127.0.0.2"], url)
         try:
             resp = session.get(url, timeout=5)
             assert resp.status_code == 200
@@ -131,17 +131,76 @@ def test_pinned_session_ignores_a_different_real_resolution():
     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_validate_fetch_url_returns_the_resolved_ip():
-    """_validate_fetch_url must return the IP it validated, so callers can
-    pin the connection to it."""
+    """_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_ip = ontology_mod._validate_fetch_url("http://example.org/ontology.ttl")
+        resolved_ips = ontology_mod._validate_fetch_url("http://example.org/ontology.ttl")
 
-    assert resolved_ip == "93.184.216.34"
+    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():
@@ -175,7 +234,7 @@ def test_validate_fetch_url_still_rejects_private_ip():
 def _make_self_signed_cert(hostname: str, tmp_path):
     import datetime
 
-    cryptography = pytest.importorskip("cryptography")
+    pytest.importorskip("cryptography")
     from cryptography import x509
     from cryptography.hazmat.primitives import hashes, serialization
     from cryptography.hazmat.primitives.asymmetric import rsa
@@ -228,6 +287,7 @@ def _start_local_https_server(cert_path, key_path):
 
     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)
@@ -246,7 +306,7 @@ def test_pinned_https_session_verifies_against_real_hostname_not_pinned_ip(tmp_p
     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)
+        session = ontology_mod._make_pinned_session(["127.0.0.1"], url)
         try:
             resp = session.get(url, timeout=5, verify=cert_path)
         finally:
@@ -269,7 +329,7 @@ def test_pinned_https_session_rejects_hostname_mismatch(tmp_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)
+        session = ontology_mod._make_pinned_session(["127.0.0.1"], url)
         try:
             import requests
             with pytest.raises(requests.exceptions.SSLError):

From ea3416ed32d8355b999497568fc983a7a07be011 Mon Sep 17 00:00:00 2001
From: KaifAhmad1 <kaifahmad087@gmail.com>
Date: Tue, 11 Aug 2026 19:16:29 +0530
Subject: [PATCH 4/4] fix: enforce a definitive no-proxy policy for the pinned
 SSRF fetcher
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Qodo's re-review confirmed the multi-IP fallback fix but kept the proxy
finding open: logging-and-falling-back when a proxy applies still let
the DNS-pinning protection be silently skipped under proxy
configuration, rather than enforcing a clear policy either way.

Implemented Qodo's preferred option: proxies are now disabled outright
for this SSRF-sensitive fetcher via session.trust_env = False, so
HTTP_PROXY/HTTPS_PROXY/NO_PROXY env vars are never consulted in the
first place (a configured proxy would perform its own DNS resolution
of the target host outside this process's control, reopening the
DNS check-then-use race pinning exists to close). The adapter also
keeps a fail-closed backstop: if a proxy is somehow still configured
despite trust_env=False (e.g. set explicitly by future code), it now
raises a clear 502 instead of silently connecting through the proxy
unpinned.

_validate_fetch_url's destination classification (blocking private/
internal targets) is unaffected either way — it runs before any of
this and doesn't depend on proxy configuration.

4 new tests: trust_env is disabled on every pinned session; an
HTTP_PROXY env var pointed at an address that would fail if contacted
is confirmed genuinely unused (real local-server fetch still succeeds
directly); and the fail-closed backstop actually raises when a proxy
is forced onto the session. Full explorer + triplet_store suite: 572
passed.
---
 semantica/explorer/routes/ontology.py       | 41 +++++++++--------
 tests/explorer/test_ontology_dns_pinning.py | 51 +++++++++++++++++++++
 2 files changed, 72 insertions(+), 20 deletions(-)

diff --git a/semantica/explorer/routes/ontology.py b/semantica/explorer/routes/ontology.py
index 005d8a0f..b50eb206 100644
--- a/semantica/explorer/routes/ontology.py
+++ b/semantica/explorer/routes/ontology.py
@@ -1046,7 +1046,6 @@ def _make_pinned_session(pinned_ips: List[str], url: str):
     `host` directly and restoring the real hostname via an explicit Host
     header (+ SNI params for HTTPS) is the correct mechanism here.
     """
-    import logging as _pin_logging
     import requests as _req
     import urllib3.util.connection as _u3_connection
     from urllib3.exceptions import NewConnectionError
@@ -1080,26 +1079,21 @@ def _make_pinned_session(pinned_ips: List[str], url: str):
 
     class _PinnedIPHTTPAdapter(_req.adapters.HTTPAdapter):
         def get_connection_with_tls_context(self, request, verify, proxies=None, cert=None):
-            # If an HTTP(S) proxy applies (env-configured or per-request),
-            # pinning can't meaningfully apply: the actual TCP connection
-            # target is the proxy, and for a forward proxy the *proxy*
-            # performs its own DNS resolution of the target host on our
-            # behalf — a resolution this process has no visibility into or
-            # control over, so there is no client-side fix for that
-            # specific race. Fall back to the normal (unpinned) path rather
-            # than silently bypassing the configured proxy.
-            # _validate_fetch_url's destination classification still fully
-            # applies either way; only this secondary DNS-pinning hardening
-            # is inherently out of scope when a proxy is in the path.
+            # 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):
-                _pin_logging.getLogger(__name__).info(
-                    "DNS pinning skipped for %s: a proxy is configured for this "
-                    "request, and proxy-side DNS resolution is outside this "
-                    "process's control.",
-                    request.url,
-                )
-                return super().get_connection_with_tls_context(
-                    request, verify, proxies=proxies, cert=cert
+                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":
@@ -1115,6 +1109,13 @@ def _make_pinned_session(pinned_ips: List[str], url: str):
             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)
diff --git a/tests/explorer/test_ontology_dns_pinning.py b/tests/explorer/test_ontology_dns_pinning.py
index 90780ba4..a68ff192 100644
--- a/tests/explorer/test_ontology_dns_pinning.py
+++ b/tests/explorer/test_ontology_dns_pinning.py
@@ -171,6 +171,57 @@ def test_pinned_session_raises_when_every_pinned_ip_is_unreachable():
         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)."""