mengzhx/semantica
1
0
Fork 0
mirror of https://github.com/semantica-agi/semantica.git synced 2026-08-29 04:26:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: address CI/review findings on DNS pinning (multi-IP fallback, TLS min version)

Browse Source 
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.
This commit is contained in:
KaifAhmad1
2026-08-11 19:10:01 +05:30
parent f2f1d6787d
commit 154a7347cd
2 changed files with 145 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
semantica/explorer/routes/ontology.py
+76 -29
View File
@@ -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:
tests/explorer/test_ontology_dns_pinning.py
+69 -9
View File
@@ -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):