From 111307d53f4f21c80d6521c8bede4b029c6bdbda Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 21 Oct 2026 07:13:44 +0000 Subject: [PATCH 1/2] Fix redirect host checks, DNS pinning, and private-target redirects Validate each redirect against the URL requests will connect to, dial the addresses that check vetted, resolve names with the system resolver under one deadline, and follow same-host redirects for private literals. Co-authored-by: Kamal Choubey --- gambit/http.py | 375 ++++++++++++++++++++++++++++++++--- gambit/net.py | 71 +++++-- tests/test_http.py | 475 ++++++++++++++++++++++++++++++++++++++++++++- tests/test_net.py | 73 +++++-- 4 files changed, 932 insertions(+), 62 deletions(-) diff --git a/gambit/http.py b/gambit/http.py index ca8e2c2..a823c33 100644 --- a/gambit/http.py +++ b/gambit/http.py @@ -12,17 +12,54 @@ the process, so only the first module to touch an HTTP-only host pays for the failed HTTPS attempt. -Redirects are followed only when the next URL stays on the same host and -every resolved address is globally routable, and only up to -:data:`MAX_REDIRECTS` hops. Response bodies are streamed and capped at -:data:`MAX_BODY` bytes unless the caller is reading the stream itself. +Redirects are followed only up to :data:`MAX_REDIRECTS` hops, and only +when the host *requests will connect to* stays the same. That host is +taken from ``requests``' prepared URL parsed with ``urllib3``, because +``urllib.parse`` and urllib3 disagree on some Location values (a +backslash in the authority is the one that turns into an SSRF). The +TCP connection is opened to the addresses vetted for that host, not to +whatever a second lookup would return. A redirect whose addresses are +not all public is refused, except when the original target is itself a +non-public address literal — an internal scan of 10.x, 192.168.x, or +127.0.0.1 must still be able to follow HTTP to HTTPS. A name that +merely resolves to a private address is a pivot and is not followed, +including names that exist only in the hosts file. + +``HTTP(S)_PROXY`` / ``proxies=`` skips the pin: a proxy resolves the +name itself. Redirects are not followed on that path. A lookup that +outruns its deadline leaves a daemon thread blocked in ``getaddrinfo`` +until the libc call returns; the scan does not wait. + +Response bodies are streamed and capped at :data:`MAX_BODY` bytes unless +the caller is reading the stream itself. """ -from urllib.parse import urljoin, urlsplit +import ipaddress +import os +import socket +from urllib.parse import urljoin import requests - -from gambit.net import is_public_ip, resolve_ips +import urllib3.connection +import urllib3.exceptions +from urllib3.util.url import parse_url + +from gambit.net import is_ip_address, is_public_ip, resolve_ips + +# Identity of the real requests entry point. Tests replace +# ``gambit.http.requests.request``; when they have, pinning is skipped +# and that double is called instead. +_ORIGINAL_REQUEST = requests.request + +_REDIRECT_STATUSES = (301, 302, 303, 307, 308) +_PROXY_ENV_VARS = ( + "http_proxy", + "https_proxy", + "all_proxy", + "HTTP_PROXY", + "HTTPS_PROXY", + "ALL_PROXY", +) USER_AGENT = ( "GAMBIT/0.1 (+https://github.com/Phantom9869/gambit; authorized testing only)" @@ -51,9 +88,13 @@ def request( other keyword is forwarded to :func:`requests.request`, except: * ``allow_redirects`` (default True) follows only same-host redirects - whose resolved addresses are all public, and at most - :data:`MAX_REDIRECTS` of them. A cross-host or private redirect is - returned as the redirect response and is not fetched. + whose vetted addresses are all public — or whose original target + is a non-public address literal — and at most :data:`MAX_REDIRECTS` + of them. The host compared is the one the next request connects + to. A cross-host redirect, a public-name-to-private pivot, or a + Location containing a backslash, whitespace, or a control + character is returned as the redirect response and is not fetched. + Each connection uses the addresses that decision vetted. * ``max_bytes`` (default :data:`MAX_BODY`) caps how much of the final body is buffered. Ignored when ``stream=True``; the caller then reads the body itself. @@ -71,28 +112,69 @@ def request( max_bytes = _as_limit(kwargs.pop("max_bytes", MAX_BODY)) kwargs["allow_redirects"] = False kwargs["stream"] = True + # A proxy resolves the name itself, so the vetted address would not + # be the address that is connected. One direct request is sent and + # redirects are not followed. + proxied = bool(kwargs.get("proxies")) or _env_proxy_configured() + use_pin = requests.request is _ORIGINAL_REQUEST and not proxied + follow = allow_redirects and not proxied current = url current_method = method + original_host: str | None = None + vetted: list[str] | None = None hops = 0 while True: - response = requests.request( - current_method, current, headers=merged, timeout=timeout, **kwargs - ) - _stamp_redirect(response, hops, current) - if not allow_redirects or hops >= MAX_REDIRECTS: + send_url = current + if use_pin: + if vetted is None: + target = _prepare_target(current) + if target is None: + raise requests.InvalidURL(f"refusing ambiguous URL {current!r}") + send_url, host = target + if original_host is None: + original_host = host + vetted = resolve_ips(_dns_name(host), _timeout_seconds(timeout)) + if not vetted: + raise requests.ConnectionError(f"could not resolve {host}") + response = _request_pinned( + current_method, + send_url, + headers=merged, + timeout=timeout, + ips=vetted, + **kwargs, + ) + else: + if original_host is None: + target = _prepare_target(current) + original_host = target[1] if target else "" + response = requests.request( + current_method, current, headers=merged, timeout=timeout, **kwargs + ) + send_url = current + + _stamp_redirect(response, hops, send_url) + vetted = None + if not follow or hops >= MAX_REDIRECTS: break status = getattr(response, "status_code", 0) or 0 location = _location_header(response) - if status not in (301, 302, 303, 307, 308) or not location: + if status not in _REDIRECT_STATUSES or not _location_is_acceptable(location): break - nxt = urljoin(current, location.strip()) - if not _redirect_is_safe(current, nxt, timeout): + target = _prepare_target(urljoin(send_url, location.strip())) + if target is None or not original_host: + break + next_url, next_host = target + vetted = _vetted_redirect_ips(original_host, next_host, timeout) + if vetted is None: break _close_quietly(response) if status == 303: current_method = "GET" - current = nxt + current = next_url + if not use_pin: + vetted = None hops += 1 if not caller_streams: _buffer_limited(response, max_bytes) @@ -296,21 +378,254 @@ def _location_header(response) -> str: return "" -def _redirect_is_safe(current: str, nxt: str, timeout) -> bool: - """Same host, http(s) only, and every resolved address globally routable.""" - new = urlsplit(nxt) - old = urlsplit(current) - if new.scheme not in ("http", "https") or not new.hostname: +def _env_proxy_configured() -> bool: + return any(os.environ.get(name) for name in _PROXY_ENV_VARS) + + +def _location_is_acceptable(location: str) -> bool: + """Reject Location values whose bytes parsers disagree on. + + A raw backslash ends the authority for urllib3 and is only userinfo + for ``urllib.parse``, so ``http://127.0.0.1\\@example.com/`` is + checked as example.com and fetched as 127.0.0.1. Whitespace and + controls are stripped by one parser and kept by the other. Outer + header whitespace is ignored; anything inside the value is refused. + """ + if not location: return False - current_host = (old.hostname or "").lower().rstrip(".") - next_host = new.hostname.lower().rstrip(".") - if not current_host or current_host != next_host: + value = location.strip() + if not value: return False + for char in value: + code = ord(char) + if char == "\\" or char.isspace() or code < 32 or code == 127: + return False + return True + + +def _prepare_target(url: str) -> tuple[str, str] | None: + """Return ``(prepared_url, host)`` for the URL requests would connect to. + + ``PreparedRequest.url`` is what the adapter sends, and urllib3 parses + that string again with :func:`urllib3.util.parse_url`. The host of + that parse is the connection host. A URL that does not survive a + second prepare unchanged is ambiguous and is refused. + """ + try: + prepared_url = requests.Request("GET", url).prepare().url + if not isinstance(prepared_url, str) or not prepared_url: + return None + if requests.Request("GET", prepared_url).prepare().url != prepared_url: + return None + parsed = parse_url(prepared_url) + except ( + requests.RequestException, + urllib3.exceptions.LocationParseError, + ValueError, + ): + return None + if parsed.scheme not in ("http", "https") or not parsed.host: + return None + return prepared_url, parsed.host + + +def _host_key(host: str) -> str: + """Comparable host: lower-case, no trailing dot, no IPv6 brackets.""" + name = host.strip().lower().rstrip(".") + if len(name) >= 2 and name.startswith("[") and name.endswith("]"): + name = name[1:-1] try: - ips = resolve_ips(next_host, _timeout_seconds(timeout)) + return ipaddress.ip_address(name).compressed + except ValueError: + return name + + +def _dns_name(host: str) -> str: + """Host to resolve. Brackets are stripped; a trailing dot stays on names. + + The dot marks an absolute DNS name (no search-list suffix). It is + removed only when what remains is an address literal. + """ + name = host.strip() + if len(name) >= 2 and name.startswith("[") and name.endswith("]"): + name = name[1:-1] + stripped = name.rstrip(".") + if is_ip_address(stripped): + return stripped + return name + + +def _policy_ip(ip: str) -> str: + """Address without an IPv6 zone id, for :func:`is_public_ip`.""" + return ip.split("%", 1)[0] + + +def _is_private_literal(host: str) -> bool: + """True when ``host`` is an address literal that is not globally routable.""" + key = _host_key(host) + return is_ip_address(key) and not is_public_ip(key) + + +def _vetted_redirect_ips(original_host: str, next_host: str, timeout) -> list[str] | None: + """Addresses the redirect may be connected to, or None to stop. + + The next host must be the original host (the one the scan was aimed + at). Every address must be public, unless that original host is + itself a non-public literal — then a same-host hop such as HTTP to + HTTPS is allowed, and a mix of public and private addresses is not. + """ + if not original_host or not next_host: + return None + if _host_key(original_host) != _host_key(next_host): + return None + try: + ips = resolve_ips(_dns_name(next_host), _timeout_seconds(timeout)) except Exception: - return False - return bool(ips) and all(is_public_ip(ip) for ip in ips) + return None + if not ips: + return None + public = [is_public_ip(_policy_ip(ip)) for ip in ips] + if all(public): + return list(ips) + if _is_private_literal(original_host) and not any(public): + return list(ips) + return None + + +def _open_tcp_socket(ip: str, port: int, timeout, source_address, socket_options): + """Connect to ``ip`` without another name lookup.""" + raw = ip + scope = 0 + if "%" in ip: + raw, zone = ip.split("%", 1) + if zone.isdigit(): + scope = int(zone) + else: + try: + scope = socket.if_nametoindex(zone) + except OSError: + scope = 0 + addr = ipaddress.ip_address(raw) + family = socket.AF_INET6 if addr.version == 6 else socket.AF_INET + sock = socket.socket(family, socket.SOCK_STREAM) + try: + if socket_options: + for level, opt, value in socket_options: + sock.setsockopt(level, opt, value) + if timeout is not None: + sock.settimeout(float(timeout)) + if source_address: + sock.bind(source_address) + if addr.version == 6: + sock.connect((raw, int(port), 0, scope)) + else: + sock.connect((raw, int(port))) + return sock + except BaseException: + sock.close() + raise + + +def _pinned_connection_class(base, ips: tuple[str, ...]): + """Connection class that dials ``ips`` and keeps ``host`` for Host/SNI.""" + + class Pinned(base): + def _new_conn(self): + last: Exception | None = None + for ip in ips: + try: + return _open_tcp_socket( + ip, + self.port, + self.timeout, + self.source_address, + self.socket_options, + ) + except TimeoutError as exc: + last = urllib3.exceptions.ConnectTimeoutError( + self, + f"Connection to {ip} timed out. (connect timeout={self.timeout}): {exc}", + ) + except OSError as exc: + last = urllib3.exceptions.NewConnectionError( + self, f"Failed to establish a new connection: {exc}" + ) + if last is None: + raise urllib3.exceptions.NewConnectionError( + self, "no vetted address to connect to" + ) + raise last + + Pinned.__name__ = f"Pinned{base.__name__}" + return Pinned + + +class _PinnedHTTPConnectionPool(urllib3.HTTPConnectionPool): + def __init__(self, host, port=None, *, gambit_pinned_ips=None, **kwargs): + super().__init__(host, port, **kwargs) + if gambit_pinned_ips: + self.ConnectionCls = _pinned_connection_class( + urllib3.connection.HTTPConnection, tuple(gambit_pinned_ips) + ) + + +class _PinnedHTTPSConnectionPool(urllib3.HTTPSConnectionPool): + def __init__(self, host, port=None, *, gambit_pinned_ips=None, **kwargs): + super().__init__(host, port, **kwargs) + if gambit_pinned_ips: + self.ConnectionCls = _pinned_connection_class( + urllib3.connection.HTTPSConnection, tuple(gambit_pinned_ips) + ) + + +class _PinnedPoolManager(urllib3.PoolManager): + def __init__(self, pinned_ips, **kwargs): + super().__init__(**kwargs) + self._gambit_pinned_ips = tuple(pinned_ips) + self.pool_classes_by_scheme = { + "http": _PinnedHTTPConnectionPool, + "https": _PinnedHTTPSConnectionPool, + } + + def _new_pool(self, scheme, host, port, request_context=None): + context = ( + self.connection_pool_kw.copy() + if request_context is None + else request_context.copy() + ) + context["gambit_pinned_ips"] = self._gambit_pinned_ips + return super()._new_pool(scheme, host, port, context) + + +class _PinnedAdapter(requests.adapters.HTTPAdapter): + def __init__(self, pinned_ips, **kwargs): + self._pinned_ips = tuple(pinned_ips) + super().__init__(**kwargs) + + def init_poolmanager(self, connections, maxsize, block=False, **pool_kwargs): + self._pool_connections = connections + self._pool_maxsize = maxsize + self._pool_block = block + self.poolmanager = _PinnedPoolManager( + self._pinned_ips, + num_pools=connections, + maxsize=maxsize, + block=block, + **pool_kwargs, + ) + + +def _request_pinned(method, url, *, headers, timeout, ips, **kwargs): + """Send one request, dialing ``ips`` instead of resolving ``url`` again.""" + session = requests.Session() + session.trust_env = False + adapter = _PinnedAdapter(ips) + session.mount("http://", adapter) + session.mount("https://", adapter) + try: + return session.request(method, url, headers=headers, timeout=timeout, **kwargs) + finally: + session.close() def _close_quietly(response) -> None: diff --git a/gambit/net.py b/gambit/net.py index 81a3ac3..c0053ab 100644 --- a/gambit/net.py +++ b/gambit/net.py @@ -2,15 +2,19 @@ ``socket.create_connection`` resolves names with ``getaddrinfo`` before it starts the connect timer, so a stuck resolver can hang a scan forever. -Callers here resolve with dnspython's ``lifetime`` first, then connect to -the resulting address. +:func:`resolve_ips` asks the system resolver (so ``/etc/hosts``, mDNS, and +NSS apply) inside a daemon thread and stops waiting at ``timeout``. One +lookup covers both address families; the deadline is not spent twice. + +A lookup that outruns the deadline is abandoned. The thread keeps running +until the libc call returns — Python cannot cancel ``getaddrinfo`` — but +the scan does not wait for it, and the thread is a daemon so it will not +hold the process open. """ import ipaddress import socket - -import dns.exception -import dns.resolver +import threading def is_ip_address(hostname: str) -> bool: @@ -39,23 +43,56 @@ def is_public_ip(ip: str) -> bool: def resolve_ips(hostname: str, timeout: float) -> list[str]: """Resolve ``hostname`` to IP strings within ``timeout`` seconds. - An address literal is returned as itself, with no DNS query. DNS - failures and an empty answer both yield ``[]``; they never raise. + An address literal is returned as itself, with no name lookup. The + system resolver (``getaddrinfo``) is used for every other name, so + hosts-file and NSS answers are visible. Failures, an empty answer, + and a lookup that exceeds ``timeout`` all yield ``[]``; they never + raise. The whole lookup shares one deadline. """ if is_ip_address(hostname): return [hostname] + if not hostname or not str(hostname).strip(): + return [] + try: + budget = float(timeout) + except (TypeError, ValueError): + return [] + if budget <= 0: + return [] + + box: dict = {} + + def worker() -> None: + try: + box["ips"] = _resolve_system(hostname) + except Exception as exc: + box["error"] = exc + + thread = threading.Thread(target=worker, name="gambit-resolve", daemon=True) + thread.start() + thread.join(budget) + if thread.is_alive() or "ips" not in box: + return [] + return list(box["ips"]) + + +def _resolve_system(hostname: str) -> list[str]: + """One ``getaddrinfo`` call: hosts file, then DNS, A and AAAA together.""" + try: + infos = socket.getaddrinfo( + hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM + ) + except OSError: + return [] found: list[str] = [] seen: set[str] = set() - for rdtype in ("A", "AAAA"): - try: - answers = dns.resolver.resolve(hostname, rdtype, lifetime=timeout) - except dns.exception.DNSException: + for _family, socktype, _proto, _canon, sockaddr in infos: + if socktype != socket.SOCK_STREAM: continue - for rdata in answers: - ip = getattr(rdata, "address", None) or str(rdata).strip() - if ip and ip not in seen: - seen.add(ip) - found.append(ip) + ip = sockaddr[0] + if ip and ip not in seen: + seen.add(ip) + found.append(ip) return found @@ -63,7 +100,7 @@ def connect(hostname: str, port: int, timeout: float): """TCP-connect to ``hostname``:``port`` with DNS bounded by ``timeout``. The socket is connected to a resolved address, never to a name, so - ``getaddrinfo`` is only asked to parse an IP literal. Ports outside + the second ``getaddrinfo`` only parses an IP literal. Ports outside 1–65535 raise ``OSError`` instead of being truncated by the resolver. Raises ``OSError`` when the name does not resolve or every address refuses the connection. diff --git a/tests/test_http.py b/tests/test_http.py index c7f8ef7..fa7102a 100644 --- a/tests/test_http.py +++ b/tests/test_http.py @@ -1,5 +1,18 @@ +import shutil +import socket +import ssl +import subprocess +import tempfile +import threading +import time +import warnings +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + import pytest import requests +from urllib3.exceptions import InsecureRequestWarning +from urllib3.util.url import parse_url import gambit.http as http @@ -12,8 +25,19 @@ def __init__(self, url): @pytest.fixture(autouse=True) -def clean_cache(): +def clean_cache(monkeypatch): http.clear_scheme_cache() + # Real-socket tests pin to a vetted address. A proxy in the environment + # would skip that pin and send the request somewhere else. + for name in ( + "http_proxy", + "https_proxy", + "all_proxy", + "HTTP_PROXY", + "HTTPS_PROXY", + "ALL_PROXY", + ): + monkeypatch.delenv(name, raising=False) yield http.clear_scheme_cache() @@ -221,3 +245,452 @@ def close(self): monkeypatch.setattr(http.requests, "request", lambda *args, **kwargs: Response()) response = http.get("https://example.com/", max_bytes=150) assert response._content == (b"A" * 100) + (b"B" * 50) + + +class _RecordingHandler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def do_GET(self): + self._respond() + + def do_HEAD(self): + self._respond(write_body=False) + + def _respond(self, write_body=True): + self.server.seen.append((self.path, self.headers.get("Host", ""))) + route = self.server.routes.get(self.path) + if route is None: + status, headers, body = 404, {}, b"missing" + else: + status, headers, body = route + if isinstance(body, str): + body = body.encode() + self.send_response(status) + for key, value in headers.items(): + self.send_header(key, value) + self.send_header("Content-Length", str(len(body))) + self.send_header("Connection", "close") + self.end_headers() + if write_body: + self.wfile.write(body) + + def log_message(self, format, *args): + return + + +def _serve(routes): + ThreadingHTTPServer.allow_reuse_address = True + server = ThreadingHTTPServer(("127.0.0.1", 0), _RecordingHandler) + server.routes = routes + server.seen = [] + server.daemon_threads = True + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + return server + + +def _close_server(server): + server.shutdown() + server.server_close() + + +class _Listener(threading.Thread): + def __init__(self): + super().__init__(daemon=True) + self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self.sock.bind(("127.0.0.1", 0)) + self.sock.listen(5) + self.sock.settimeout(0.2) + self.port = self.sock.getsockname()[1] + self.hits = [] + self._done = threading.Event() + + def run(self): + while not self._done.is_set(): + try: + conn, _addr = self.sock.accept() + except socket.timeout: + continue + self.hits.append(conn.getpeername()) + conn.close() + + def close(self): + self._done.set() + self.join(timeout=1) + self.sock.close() + + +def _redirect_response(location, calls): + class Response: + def __init__(self): + self.url = calls[-1] + self.status_code = 302 if len(calls) == 1 else 200 + self.headers = {"Location": location} if self.status_code == 302 else {} + + def close(self): + return None + + def iter_content(self, chunk_size=8192): + yield b"" + + return Response() + + +def test_backslash_location_is_not_fetched_from_the_real_connection_target(monkeypatch): + """urlsplit sees example.com; requests connects to 127.0.0.1. Do not follow. + + resolve_ips is the only stand-in, and it reports a public address, which + is the condition that used to let the check pass. The transport still + parses the URL the way requests does and dials that host for real. + """ + listener = _Listener() + listener.start() + calls = [] + attempted = [] + location = f"http://127.0.0.1:{listener.port}\\@example.com/secret" + + def fake_request(method, url, headers=None, timeout=None, **kwargs): + calls.append(url) + prepared = requests.Request("GET", url).prepare().url + parsed = parse_url(prepared) + if parsed.host == "127.0.0.1" and parsed.port == listener.port: + attempted.append(prepared) + sock = socket.create_connection(("127.0.0.1", listener.port), timeout=1) + sock.close() + return _redirect_response(location, calls) + + monkeypatch.setattr(http.requests, "request", fake_request) + monkeypatch.setattr(http, "resolve_ips", lambda host, timeout: ["93.184.216.34"]) + try: + response = http.get("http://example.com/") + time.sleep(0.3) + finally: + listener.close() + assert response.status_code == 302 + assert response.gambit_redirects == 0 + assert calls == ["http://example.com/"] + assert attempted == [] + assert listener.hits == [] + + +def test_parser_differential_is_blocked_without_the_character_filter(monkeypatch): + """The prepared-URL host, not the backslash filter, has to stop this.""" + monkeypatch.setattr(http, "_location_is_acceptable", lambda location: bool(location)) + listener = _Listener() + listener.start() + calls = [] + attempted = [] + location = f"http://127.0.0.1:{listener.port}\\@example.com/secret" + + def fake_request(method, url, headers=None, timeout=None, **kwargs): + calls.append(url) + prepared = requests.Request("GET", url).prepare().url + parsed = parse_url(prepared) + if parsed.host == "127.0.0.1" and parsed.port == listener.port: + attempted.append(prepared) + sock = socket.create_connection(("127.0.0.1", listener.port), timeout=1) + sock.close() + return _redirect_response(location, calls) + + monkeypatch.setattr(http.requests, "request", fake_request) + monkeypatch.setattr(http, "resolve_ips", lambda host, timeout: ["93.184.216.34"]) + try: + response = http.get("http://example.com/start") + time.sleep(0.3) + finally: + listener.close() + assert response.gambit_redirects == 0 + assert len(calls) == 1 + assert attempted == [] + assert listener.hits == [] + prepared = requests.Request("GET", location).prepare().url + assert parse_url(prepared).host == "127.0.0.1" + assert parse_url(prepared).port == listener.port + + +def test_percent_encoded_backslash_and_at_sign_do_not_reach_a_private_host(monkeypatch): + listener = _Listener() + listener.start() + cases = [ + f"http://127.0.0.1:{listener.port}%5C@example.com/secret", + f"http://127.0.0.1:{listener.port}%5c@example.com/secret", + f"http://example.com@127.0.0.1:{listener.port}/secret", + ] + try: + for location in cases: + calls = [] + attempted = [] + + def fake_request(method, url, headers=None, timeout=None, location=location, **kwargs): + calls.append(url) + prepared = requests.Request("GET", url).prepare().url + parsed = parse_url(prepared) + if parsed.host == "127.0.0.1" and parsed.port == listener.port: + attempted.append(prepared) + sock = socket.create_connection(("127.0.0.1", listener.port), timeout=1) + sock.close() + return _redirect_response(location, calls) + + monkeypatch.setattr(http.requests, "request", fake_request) + monkeypatch.setattr(http, "resolve_ips", lambda host, timeout: ["93.184.216.34"]) + response = http.get("http://example.com/") + assert response.gambit_redirects == 0 or all( + parse_url(requests.Request("GET", url).prepare().url).host == "example.com" + for url in calls + ) + assert attempted == [] + time.sleep(0.3) + assert listener.hits == [] + finally: + listener.close() + + +def test_whitespace_and_control_characters_in_location_are_not_followed(monkeypatch): + calls = [] + + def fake_request(method, url, headers=None, timeout=None, **kwargs): + calls.append(url) + + class Response: + status_code = 302 + headers = {"Location": "http://127.0.0.1\t@example.com/"} + + def __init__(self): + self.url = url + + def close(self): + return None + + return Response() + + monkeypatch.setattr(http.requests, "request", fake_request) + monkeypatch.setattr(http, "resolve_ips", lambda host, timeout: ["93.184.216.34"]) + response = http.get("http://example.com/") + assert calls == ["http://example.com/"] + assert response.gambit_redirects == 0 + + +def test_private_literals_follow_a_same_host_scheme_change(monkeypatch): + calls = [] + + def fake_request(method, url, headers=None, timeout=None, **kwargs): + calls.append(url) + + class Response: + def __init__(self, url): + self.url = url + https = url.startswith("https://") + self.status_code = 200 if https else 301 + if https: + self.headers = {} + else: + self.headers = {"Location": "https://" + url[len("http://") :]} + + def close(self): + return None + + def iter_content(self, chunk_size=8192): + yield b"ok" + + return Response(url) + + monkeypatch.setattr(http.requests, "request", fake_request) + for origin in ( + "http://127.0.0.1/r", + "http://10.1.2.3/r", + "http://192.168.1.9/r", + "http://[::1]/r", + ): + calls.clear() + response = http.get(origin) + assert response.gambit_redirects == 1, origin + assert calls[1].startswith("https://"), calls + assert response.gambit_final_url == calls[1] + + +def test_relative_and_scheme_relative_redirects_on_a_private_target(): + server = _serve({}) + port = server.server_address[1] + host = f"127.0.0.1:{port}" + server.routes = { + "/r": (302, {"Location": "/next"}, b""), + "/next": (302, {"Location": f"//{host}/done"}, b""), + "/done": (200, {}, b"done"), + } + try: + response = http.get(f"http://127.0.0.1:{port}/r", timeout=2) + finally: + _close_server(server) + assert response.status_code == 200 + assert response.content == b"done" + assert response.gambit_redirects == 2 + assert [path for path, _host in server.seen] == ["/r", "/next", "/done"] + + +def test_private_target_does_not_follow_a_cross_host_redirect(): + server = _serve({"/r": (302, {"Location": "http://127.0.0.2/secret"}, b"stay")}) + port = server.server_address[1] + try: + response = http.get(f"http://127.0.0.1:{port}/r", timeout=2) + finally: + _close_server(server) + assert response.status_code == 302 + assert response.gambit_redirects == 0 + assert [path for path, _host in server.seen] == ["/r"] + + +def test_public_name_resolving_private_does_not_follow_redirect(monkeypatch): + server = _serve({}) + port = server.server_address[1] + server.routes = { + "/r": (302, {"Location": f"http://example.com:{port}/secret"}, b""), + "/secret": (200, {}, b"nope"), + } + monkeypatch.setattr(http, "resolve_ips", lambda host, timeout: ["127.0.0.1"]) + try: + response = http.get(f"http://example.com:{port}/r", timeout=2) + finally: + _close_server(server) + assert response.status_code == 302 + assert response.gambit_redirects == 0 + assert [path for path, _host in server.seen] == ["/r"] + + +def test_connection_uses_vetted_ips_when_getaddrinfo_changes(monkeypatch): + server = _serve({"/": (200, {}, b"pinned")}) + port = server.server_address[1] + + def blackhole(host, port, family=0, type=0, proto=0, flags=0): + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("203.0.113.1", port or 0))] + + monkeypatch.setattr(socket, "getaddrinfo", blackhole) + monkeypatch.setattr(http, "resolve_ips", lambda host, timeout: ["127.0.0.1"]) + try: + response = http.get(f"http://rebind.example:{port}/", timeout=2) + finally: + _close_server(server) + assert response.status_code == 200 + assert response.content == b"pinned" + assert server.seen == [("/", f"rebind.example:{port}")] + + +def test_private_redirect_stays_on_the_vetted_address(monkeypatch): + server = _serve( + { + "/r": (301, {"Location": "/x"}, b""), + "/x": (200, {}, b"ok"), + } + ) + port = server.server_address[1] + + def blackhole(host, port, family=0, type=0, proto=0, flags=0): + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("203.0.113.1", port or 0))] + + monkeypatch.setattr(socket, "getaddrinfo", blackhole) + try: + response = http.get(f"http://127.0.0.1:{port}/r", timeout=2) + finally: + _close_server(server) + assert response.status_code == 200 + assert response.content == b"ok" + assert response.gambit_redirects == 1 + assert response.gambit_final_url == f"http://127.0.0.1:{port}/x" + assert [path for path, _host in server.seen] == ["/r", "/x"] + + +def test_configured_proxy_does_not_follow_redirects(monkeypatch): + calls = [] + + def fake_request(method, url, headers=None, timeout=None, **kwargs): + calls.append(url) + + class Response: + status_code = 302 + headers = {"Location": "https://example.com/next"} + + def __init__(self): + self.url = url + + def close(self): + return None + + return Response() + + monkeypatch.setattr(http.requests, "request", fake_request) + monkeypatch.setattr(http, "resolve_ips", lambda host, timeout: ["93.184.216.34"]) + response = http.get("http://example.com/", proxies={"http": "http://127.0.0.1:9"}) + assert calls == ["http://example.com/"] + assert response.gambit_redirects == 0 + + calls.clear() + monkeypatch.setenv("https_proxy", "http://127.0.0.1:9") + response = http.get("http://example.com/") + assert calls == ["http://example.com/"] + assert response.gambit_redirects == 0 + + +def test_private_literal_follows_http_to_https_against_a_local_server(): + openssl = shutil.which("openssl") + if openssl is None: + pytest.skip("openssl is required to mint a certificate") + with tempfile.TemporaryDirectory() as tmp: + directory = Path(tmp) + key = directory / "key.pem" + cert = directory / "cert.pem" + subprocess.run( + [ + openssl, + "req", + "-x509", + "-newkey", + "rsa:2048", + "-keyout", + str(key), + "-out", + str(cert), + "-days", + "1", + "-nodes", + "-subj", + "/CN=127.0.0.1", + "-addext", + "subjectAltName=IP:127.0.0.1", + ], + check=True, + capture_output=True, + ) + https = ThreadingHTTPServer(("127.0.0.1", 0), _RecordingHandler) + https.routes = {"/x": (200, {}, b"secure")} + https.seen = [] + context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + context.load_cert_chain(cert, key) + https.socket = context.wrap_socket(https.socket, server_side=True) + https_port = https.server_address[1] + https_thread = threading.Thread(target=https.serve_forever, daemon=True) + https_thread.start() + http_server = _serve( + { + "/r": ( + 301, + {"Location": f"https://127.0.0.1:{https_port}/x"}, + b"", + ) + } + ) + try: + with warnings.catch_warnings(): + warnings.simplefilter("ignore", InsecureRequestWarning) + response = http.get( + f"http://127.0.0.1:{http_server.server_address[1]}/r", + timeout=2, + verify=False, + ) + finally: + _close_server(http_server) + https.shutdown() + https.server_close() + assert response.status_code == 200 + assert response.content == b"secure" + assert response.gambit_redirects == 1 + assert response.gambit_final_url == f"https://127.0.0.1:{https_port}/x" diff --git a/tests/test_net.py b/tests/test_net.py index cc49e22..db82bbf 100644 --- a/tests/test_net.py +++ b/tests/test_net.py @@ -1,9 +1,9 @@ import socket +import time -import dns.resolver import pytest -from gambit.net import connect, is_public_ip +from gambit.net import connect, is_public_ip, resolve_ips def test_public_address_check_rejects_metadata_and_private_ranges(): @@ -19,15 +19,11 @@ def test_public_address_check_rejects_metadata_and_private_ranges(): def test_connect_resolves_with_a_deadline_then_dials_the_ip(monkeypatch): seen = {} - def fake_resolve(name, rdtype, lifetime=None): - seen.setdefault("queries", []).append((name, rdtype, lifetime)) - if rdtype == "AAAA": - raise dns.resolver.NoAnswer() - - class Answer: - address = "93.184.216.34" - - return [Answer()] + def fake_getaddrinfo(host, port, family=0, type=0, proto=0, flags=0): + seen.setdefault("hosts", []).append(host) + return [ + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", port or 0)) + ] def fake_create_connection(address, timeout=None): seen["address"] = address @@ -39,14 +35,63 @@ def close(self): return Sock() - monkeypatch.setattr(dns.resolver, "resolve", fake_resolve) + monkeypatch.setattr(socket, "getaddrinfo", fake_getaddrinfo) monkeypatch.setattr(socket, "create_connection", fake_create_connection) sock = connect("example.com", 443, 1.25) assert sock is not None assert seen["address"] == ("93.184.216.34", 443) assert seen["timeout"] == 1.25 - assert ("example.com", "A", 1.25) in seen["queries"] - assert all(item[2] == 1.25 for item in seen["queries"]) + # One system lookup covers every family. The deadline is not paid per record type. + assert seen["hosts"] == ["example.com"] + + +def test_resolve_ips_uses_one_deadline_for_every_family(monkeypatch): + calls = [] + + def slow_getaddrinfo(host, port, family=0, type=0, proto=0, flags=0): + calls.append(host) + time.sleep(1.0) + return [ + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0)), + (socket.AF_INET6, socket.SOCK_STREAM, 6, "", ("2606:4700:4700::1111", 0, 0, 0)), + ] + + monkeypatch.setattr(socket, "getaddrinfo", slow_getaddrinfo) + started = time.monotonic() + assert resolve_ips("example.com", 0.3) == [] + elapsed = time.monotonic() - started + assert elapsed < 0.8 + assert calls == ["example.com"] + + +def test_resolve_ips_localhost_uses_the_system_resolver(): + ips = resolve_ips("localhost", 2.0) + assert any(ip in {"127.0.0.1", "::1"} for ip in ips) + + +def test_connect_reaches_localhost_from_the_system_resolver(): + ips = resolve_ips("localhost", 2.0) + assert ips + first = ips[0] + family = socket.AF_INET6 if ":" in first.split("%", 1)[0] else socket.AF_INET + listener = socket.socket(family, socket.SOCK_STREAM) + if family == socket.AF_INET6: + listener.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1) + listener.bind((first, 0)) + listener.listen(1) + listener.settimeout(2) + port = listener.getsockname()[1] + client = None + accepted = None + try: + client = connect("localhost", port, 2) + accepted, _addr = listener.accept() + finally: + if accepted is not None: + accepted.close() + if client is not None: + client.close() + listener.close() def test_connect_refuses_a_port_that_would_wrap(monkeypatch): From bcaab74ab54df747973f4b16d7a220b4e9febf53 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 21 Oct 2026 07:32:43 +0000 Subject: [PATCH 2/2] Honour CA bundles, NO_PROXY, and private-name redirects The pinned session kept trust_env off, so REQUESTS_CA_BUNDLE was ignored and a certificate failure was retried as plain HTTP. Apply the env CA bundle explicitly, and surface TLS verification errors instead of downgrading. Follow same-host redirects for names that resolve only to non-public addresses, still blocking a public-to-private pivot. Check proxies per URL, limit redirect ports to the same port or 80/443, return the 3xx when a followed hop fails, and resolve a port scan's host once. Co-authored-by: Kamal Choubey --- gambit/http.py | 385 ++++++++++++++++++++++++++--------- gambit/modules/ports.py | 17 +- gambit/net.py | 16 +- tests/test_http.py | 434 ++++++++++++++++++++++++++++++++++------ tests/test_net.py | 24 +++ tests/test_ports.py | 32 ++- 6 files changed, 733 insertions(+), 175 deletions(-) diff --git a/gambit/http.py b/gambit/http.py index a823c33..b9699b4 100644 --- a/gambit/http.py +++ b/gambit/http.py @@ -18,17 +18,28 @@ ``urllib.parse`` and urllib3 disagree on some Location values (a backslash in the authority is the one that turns into an SSRF). The TCP connection is opened to the addresses vetted for that host, not to -whatever a second lookup would return. A redirect whose addresses are -not all public is refused, except when the original target is itself a -non-public address literal — an internal scan of 10.x, 192.168.x, or -127.0.0.1 must still be able to follow HTTP to HTTPS. A name that -merely resolves to a private address is a pivot and is not followed, -including names that exist only in the hosts file. - -``HTTP(S)_PROXY`` / ``proxies=`` skips the pin: a proxy resolves the -name itself. Redirects are not followed on that path. A lookup that -outruns its deadline leaves a daemon thread blocked in ``getaddrinfo`` -until the libc call returns; the scan does not wait. +whatever a second lookup would return. The redirect port must be the +original port, or the other side of the 80/443 pair. A hop that fails +returns the redirect response already in hand. + +A same-host redirect whose addresses are all public is followed. So is +one whose addresses are all non-public, when the original target is a +non-public literal **or** a name whose own first resolution was entirely +non-public (a hosts-file lab name, or ``localhost``). The redirect is +resolved again; a later private answer is not allowed to ride a first +answer that was public or mixed. Mixed addresses are refused either +way, so a public name cannot pivot into a private network. + +``HTTP(S)_PROXY`` / ``proxies=`` skips the pin for URLs that would +actually use the proxy: a proxy resolves the name itself. ``NO_PROXY`` +is applied per URL, and a bypassed host keeps the pin and may follow +redirects. The pinned session does not trust the environment for +proxies (that would undo the pin) but still honours +``REQUESTS_CA_BUNDLE`` / ``CURL_CA_BUNDLE`` when the caller did not pass +``verify=False`` or an explicit bundle path. A TLS *verification* +failure is not retried over plain HTTP. +A lookup that outruns its deadline leaves a daemon thread blocked in +``getaddrinfo`` until the libc call returns; the scan does not wait. Response bodies are streamed and capped at :data:`MAX_BODY` bytes unless the caller is reading the stream itself. @@ -37,6 +48,8 @@ import ipaddress import os import socket +import ssl +from typing import NamedTuple from urllib.parse import urljoin import requests @@ -52,14 +65,7 @@ _ORIGINAL_REQUEST = requests.request _REDIRECT_STATUSES = (301, 302, 303, 307, 308) -_PROXY_ENV_VARS = ( - "http_proxy", - "https_proxy", - "all_proxy", - "HTTP_PROXY", - "HTTPS_PROXY", - "ALL_PROXY", -) +_HTTP_HTTPS_PORTS = frozenset((80, 443)) USER_AGENT = ( "GAMBIT/0.1 (+https://github.com/Phantom9869/gambit; authorized testing only)" @@ -88,13 +94,16 @@ def request( other keyword is forwarded to :func:`requests.request`, except: * ``allow_redirects`` (default True) follows only same-host redirects - whose vetted addresses are all public — or whose original target - is a non-public address literal — and at most :data:`MAX_REDIRECTS` - of them. The host compared is the one the next request connects - to. A cross-host redirect, a public-name-to-private pivot, or a - Location containing a backslash, whitespace, or a control - character is returned as the redirect response and is not fetched. - Each connection uses the addresses that decision vetted. + on the same port (or the 80/443 pair) whose vetted addresses are + all public — or all non-public when the original target is a + non-public literal or a name that resolved only to non-public + addresses — and at most :data:`MAX_REDIRECTS` of them. The host + compared is the one the next request connects to. A cross-host + redirect, a public-to-private pivot, a port change, or a Location + containing a backslash, whitespace, or a control character is + returned as the redirect response and is not fetched. Each + connection uses the addresses that decision vetted. A followed hop + that errors returns that redirect response instead of raising. * ``max_bytes`` (default :data:`MAX_BODY`) caps how much of the final body is buffered. Ignored when ``stream=True``; the caller then reads the body itself. @@ -112,71 +121,115 @@ def request( max_bytes = _as_limit(kwargs.pop("max_bytes", MAX_BODY)) kwargs["allow_redirects"] = False kwargs["stream"] = True - # A proxy resolves the name itself, so the vetted address would not - # be the address that is connected. One direct request is sent and - # redirects are not followed. - proxied = bool(kwargs.get("proxies")) or _env_proxy_configured() - use_pin = requests.request is _ORIGINAL_REQUEST and not proxied - follow = allow_redirects and not proxied + proxies = kwargs.get("proxies") current = url current_method = method original_host: str | None = None - vetted: list[str] | None = None + original_port: int | None = None + # None until the first resolution of the scan target is classified. + # A later hop must not overwrite it: public-then-private is a pivot. + original_private: bool | None = None + # Addresses the next pinned send must dial. Set only after a redirect + # was allowed, so the connect cannot look the name up again. + pending_ips: list[str] | None = None hops = 0 + response = None + previous = None while True: + target = _prepare_target(current) send_url = current - if use_pin: - if vetted is None: - target = _prepare_target(current) + proxied = _uses_proxy(target.url if target is not None else current, proxies) + # Tests replace requests.request. Honour that double, and do not + # pin through a proxy: the proxy would resolve the name itself. + use_pin = requests.request is _ORIGINAL_REQUEST and not proxied + if original_host is None: + if target is not None: + original_host = target.host + original_port = target.port + else: + original_host = "" + try: + if use_pin: if target is None: raise requests.InvalidURL(f"refusing ambiguous URL {current!r}") - send_url, host = target - if original_host is None: - original_host = host - vetted = resolve_ips(_dns_name(host), _timeout_seconds(timeout)) - if not vetted: - raise requests.ConnectionError(f"could not resolve {host}") - response = _request_pinned( - current_method, - send_url, - headers=merged, - timeout=timeout, - ips=vetted, - **kwargs, - ) - else: - if original_host is None: - target = _prepare_target(current) - original_host = target[1] if target else "" - response = requests.request( - current_method, current, headers=merged, timeout=timeout, **kwargs - ) - send_url = current - + send_url = target.url + if pending_ips is None: + ips = resolve_ips(_dns_name(target.host), _timeout_seconds(timeout)) + if not ips: + raise requests.ConnectionError( + f"could not resolve {target.host}" + ) + if original_private is None: + original_private = _is_private_target(original_host, ips) + else: + ips = pending_ips + pending_ips = None + response = _request_pinned( + current_method, + send_url, + headers=merged, + timeout=timeout, + ips=ips, + **kwargs, + ) + else: + response = requests.request( + current_method, + current, + headers=merged, + timeout=timeout, + **kwargs, + ) + send_url = current + except requests.RequestException: + if previous is not None: + response = previous + break + raise + if previous is not None: + _close_quietly(previous) + previous = None + hops += 1 _stamp_redirect(response, hops, send_url) - vetted = None - if not follow or hops >= MAX_REDIRECTS: + if not allow_redirects or proxied or hops >= MAX_REDIRECTS: break status = getattr(response, "status_code", 0) or 0 location = _location_header(response) if status not in _REDIRECT_STATUSES or not _location_is_acceptable(location): break - target = _prepare_target(urljoin(send_url, location.strip())) - if target is None or not original_host: + nxt = _prepare_target(urljoin(send_url, location.strip())) + if ( + nxt is None + or not original_host + or original_port is None + or not _ports_compatible(original_port, nxt.port) + or _host_key(original_host) != _host_key(nxt.host) + or _uses_proxy(nxt.url, proxies) + ): break - next_url, next_host = target - vetted = _vetted_redirect_ips(original_host, next_host, timeout) + if original_private is None: + try: + snapshot = resolve_ips( + _dns_name(original_host), _timeout_seconds(timeout) + ) + except Exception: + break + original_private = _is_private_target(original_host, snapshot) + vetted = _vetted_redirect_ips( + original_host, original_private, nxt.host, timeout + ) if vetted is None: break - _close_quietly(response) if status == 303: current_method = "GET" - current = next_url - if not use_pin: - vetted = None - hops += 1 - if not caller_streams: + # Keep the redirect until the next hop succeeds. A failure + # returns this response instead of raising. + previous = response + current = nxt.url + if requests.request is _ORIGINAL_REQUEST: + pending_ips = vetted + if response is not None and not caller_streams: _buffer_limited(response, max_bytes) return response @@ -260,9 +313,12 @@ def url_authority(target: str, scheme: str) -> str: def _request_for_host(method: str, target: str, path: str = "", **kwargs): """``method`` request to ``target`` + ``path``, HTTPS first, HTTP fallback. - Only connection-level failures (:class:`requests.RequestException`) - trigger the fallback; an HTTP error status is a real answer and is - returned as-is. Raises the last exception if every scheme fails. + Connection failures (:class:`requests.RequestException`) trigger the + fallback; an HTTP error status is a real answer and is returned + as-is. A TLS certificate verification failure is not a dead port: + it is re-raised, and plain HTTP is not tried. That keeps an internal + CA problem (or a hostname mismatch) from being silently downgraded. + Raises the last exception if every scheme fails. """ if path and not path.startswith("/"): path = "/" + path @@ -272,6 +328,11 @@ def _request_for_host(method: str, target: str, path: str = "", **kwargs): try: response = request(method, url, **kwargs) except requests.RequestException as exc: + if _is_tls_verify_failure(exc): + raise requests.exceptions.SSLError( + "TLS certificate verification failed for " + f"{target}; not retrying over HTTP: {exc}" + ) from exc last_exc = exc continue set_preferred_scheme(target, scheme) @@ -378,8 +439,95 @@ def _location_header(response) -> str: return "" -def _env_proxy_configured() -> bool: - return any(os.environ.get(name) for name in _PROXY_ENV_VARS) +def _uses_proxy(url: str, proxies) -> bool: + """True when ``url`` would be sent through a proxy. + + ``NO_PROXY`` / ``no_proxy`` is checked for this URL. A bypassed host + is direct even when a proxy variable is set for other hosts. Any + other configured proxy counts: the proxy would resolve the name, so + the pin would not be the address that is contacted. + """ + if isinstance(proxies, dict): + mapping = proxies + elif proxies: + return True + else: + mapping = {} + no_proxy = mapping.get("no_proxy") + try: + if requests.utils.should_bypass_proxies(url, no_proxy=no_proxy): + return False + except Exception: + return True + for key, value in mapping.items(): + if key == "no_proxy" or not value: + continue + return True + try: + return bool(requests.utils.get_environ_proxies(url, no_proxy=no_proxy)) + except Exception: + return True + + +def _is_tls_verify_failure(exc: BaseException) -> bool: + """True when ``exc`` is a certificate verification failure. + + Hostname mismatch, an untrusted internal CA, and an expired + certificate all surface as ``CERTIFICATE_VERIFY_FAILED``. A protocol + mismatch (plain HTTP on the HTTPS port) does not, and still falls + back. + """ + seen: set[int] = set() + stack: list[BaseException] = [exc] + while stack: + current = stack.pop() + marker = id(current) + if marker in seen: + continue + seen.add(marker) + if isinstance(current, ssl.SSLCertVerificationError): + return True + message = str(current).lower() + if ( + "certificate_verify_failed" in message + or "certificate verify failed" in message + ): + return True + for attr in ("__cause__", "__context__", "reason"): + nested = getattr(current, attr, None) + if isinstance(nested, BaseException): + stack.append(nested) + for arg in getattr(current, "args", ()): + if isinstance(arg, BaseException): + stack.append(arg) + return False + + +def _with_env_ca_bundle(kwargs: dict) -> dict: + """Copy ``kwargs`` and apply ``REQUESTS_CA_BUNDLE`` / ``CURL_CA_BUNDLE``. + + The pinned session keeps ``trust_env`` false so an environment proxy + cannot replace the dialed address. requests reads the CA bundle only + when ``trust_env`` is true, so copy it into ``verify`` here. + ``verify=False`` and an explicit bundle path are left alone. + """ + copied = dict(kwargs) + copied.pop("proxies", None) + if "verify" not in copied: + verify = True + else: + verify = copied.get("verify") + if verify is False or (isinstance(verify, str) and verify): + return copied + if verify is True or verify is None: + bundle = os.environ.get("REQUESTS_CA_BUNDLE") or os.environ.get( + "CURL_CA_BUNDLE" + ) + if bundle: + copied["verify"] = bundle + elif verify is None: + copied["verify"] = True + return copied def _location_is_acceptable(location: str) -> bool: @@ -403,13 +551,21 @@ def _location_is_acceptable(location: str) -> bool: return True -def _prepare_target(url: str) -> tuple[str, str] | None: - """Return ``(prepared_url, host)`` for the URL requests would connect to. +class _Target(NamedTuple): + url: str + host: str + port: int + scheme: str + + +def _prepare_target(url: str) -> _Target | None: + """Return the URL requests would connect to, plus host and port. ``PreparedRequest.url`` is what the adapter sends, and urllib3 parses that string again with :func:`urllib3.util.parse_url`. The host of that parse is the connection host. A URL that does not survive a - second prepare unchanged is ambiguous and is refused. + second prepare unchanged is ambiguous and is refused. A missing port + becomes 80 or 443 from the scheme, which is the port that is dialed. """ try: prepared_url = requests.Request("GET", url).prepare().url @@ -426,7 +582,15 @@ def _prepare_target(url: str) -> tuple[str, str] | None: return None if parsed.scheme not in ("http", "https") or not parsed.host: return None - return prepared_url, parsed.host + port = parsed.port + if port is None: + port = 443 if parsed.scheme == "https" else 80 + return _Target(prepared_url, parsed.host, int(port), parsed.scheme) + + +def _ports_compatible(origin: int, nxt: int) -> bool: + """Same port, or the HTTP/HTTPS pair (80 and 443).""" + return origin == nxt or {origin, nxt} == _HTTP_HTTPS_PORTS def _host_key(host: str) -> str: @@ -466,13 +630,43 @@ def _is_private_literal(host: str) -> bool: return is_ip_address(key) and not is_public_ip(key) -def _vetted_redirect_ips(original_host: str, next_host: str, timeout) -> list[str] | None: +def _is_private_target(host: str, ips: list[str]) -> bool: + """True when the scan target itself is non-public. + + A non-public literal qualifies on its spelling. A name qualifies + only when every address from its first resolution is non-public, so + ``localhost`` and a hosts-file lab name follow same-host redirects. + An empty answer does not: there is nothing to call private. + """ + if _is_private_literal(host): + return True + if not ips: + return False + return all(not is_public_ip(_policy_ip(ip)) for ip in ips) + + +def _redirect_ips_allowed(original_private: bool, ips: list[str]) -> bool: + """Whether ``ips`` may be dialed for a same-host redirect. + + All-public is the normal case. All-non-public is allowed only when + the original target was itself private, so a public name cannot + pivot. Mixed answers fail closed. + """ + if not ips: + return False + public = [is_public_ip(_policy_ip(ip)) for ip in ips] + if all(public): + return True + return original_private and not any(public) + + +def _vetted_redirect_ips( + original_host: str, original_private: bool, next_host: str, timeout +) -> list[str] | None: """Addresses the redirect may be connected to, or None to stop. - The next host must be the original host (the one the scan was aimed - at). Every address must be public, unless that original host is - itself a non-public literal — then a same-host hop such as HTTP to - HTTPS is allowed, and a mix of public and private addresses is not. + The next host must be the original host. It is resolved again; the + first snapshot is only the privacy classification of the target. """ if not original_host or not next_host: return None @@ -482,14 +676,9 @@ def _vetted_redirect_ips(original_host: str, next_host: str, timeout) -> list[st ips = resolve_ips(_dns_name(next_host), _timeout_seconds(timeout)) except Exception: return None - if not ips: + if not _redirect_ips_allowed(original_private, ips): return None - public = [is_public_ip(_policy_ip(ip)) for ip in ips] - if all(public): - return list(ips) - if _is_private_literal(original_host) and not any(public): - return list(ips) - return None + return list(ips) def _open_tcp_socket(ip: str, port: int, timeout, source_address, socket_options): @@ -616,14 +805,22 @@ def init_poolmanager(self, connections, maxsize, block=False, **pool_kwargs): def _request_pinned(method, url, *, headers, timeout, ips, **kwargs): - """Send one request, dialing ``ips`` instead of resolving ``url`` again.""" + """Send one request, dialing ``ips`` instead of resolving ``url`` again. + + ``trust_env`` stays false so environment proxies are not applied on + top of the pin. The CA bundle is copied in explicitly; see + :func:`_with_env_ca_bundle`. + """ session = requests.Session() session.trust_env = False adapter = _PinnedAdapter(ips) session.mount("http://", adapter) session.mount("https://", adapter) + send_kwargs = _with_env_ca_bundle(kwargs) try: - return session.request(method, url, headers=headers, timeout=timeout, **kwargs) + return session.request( + method, url, headers=headers, timeout=timeout, **send_kwargs + ) finally: session.close() diff --git a/gambit/modules/ports.py b/gambit/modules/ports.py index c588d80..964622f 100644 --- a/gambit/modules/ports.py +++ b/gambit/modules/ports.py @@ -108,11 +108,13 @@ def run(target: str) -> list[Finding]: """Try one TCP connect per port in PORTS and report the open ones. An explicit port on ``target`` (``example.com:9090``) is probed too, - in addition to :data:`PORTS`. DNS for each connect is bounded by - :data:`TIMEOUT`. A port outside 1–65535 is not probed: the platform - resolver would otherwise truncate it into a different port. + in addition to :data:`PORTS`. The host is resolved once; every port + dials that address list, so a scan does not start one resolver + thread per port. The lookup is bounded by :data:`TIMEOUT`. A port + outside 1–65535 is not probed: the platform resolver would otherwise + truncate it into a different port. """ - from gambit.net import connect + from gambit.net import connect, resolve_ips from gambit.pipeline import split_host_port hostname, explicit_port = split_host_port(target) @@ -120,12 +122,15 @@ def run(target: str) -> list[Finding]: ports = list(PORTS) if explicit_port and explicit_port not in ports: ports.append(explicit_port) + ips = resolve_ips(host, TIMEOUT) + if not ips: + return [] findings = [] for port in ports: try: - with connect(host, port, TIMEOUT): + with connect(host, port, TIMEOUT, resolved=ips): pass # connected == open; send nothing, read nothing except OSError: - continue # refused, filtered, timed out, or unresolvable + continue # refused, filtered, or timed out findings.append(evaluate(host, port)) return findings diff --git a/gambit/net.py b/gambit/net.py index c0053ab..c97b3e5 100644 --- a/gambit/net.py +++ b/gambit/net.py @@ -96,19 +96,27 @@ def _resolve_system(hostname: str) -> list[str]: return found -def connect(hostname: str, port: int, timeout: float): +def connect( + hostname: str, + port: int, + timeout: float, + *, + resolved: list[str] | None = None, +): """TCP-connect to ``hostname``:``port`` with DNS bounded by ``timeout``. The socket is connected to a resolved address, never to a name, so the second ``getaddrinfo`` only parses an IP literal. Ports outside 1–65535 raise ``OSError`` instead of being truncated by the resolver. - Raises ``OSError`` when the name does not resolve or every address - refuses the connection. + Pass ``resolved`` to dial addresses already looked up (one lookup + for many ports). ``None`` resolves ``hostname``; an empty list does + not, and raises. Raises ``OSError`` when the name does not resolve + or every address refuses the connection. """ number = int(port) if number < 1 or number > 65535: raise OSError(f"port out of range: {port}") - ips = resolve_ips(hostname, timeout) + ips = list(resolved) if resolved is not None else resolve_ips(hostname, timeout) if not ips: raise OSError(f"could not resolve {hostname}") last: OSError | None = None diff --git a/tests/test_http.py b/tests/test_http.py index fa7102a..1cdb1d9 100644 --- a/tests/test_http.py +++ b/tests/test_http.py @@ -5,13 +5,11 @@ import tempfile import threading import time -import warnings from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path import pytest import requests -from urllib3.exceptions import InsecureRequestWarning from urllib3.util.url import parse_url import gambit.http as http @@ -36,6 +34,10 @@ def clean_cache(monkeypatch): "HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", + "no_proxy", + "NO_PROXY", + "REQUESTS_CA_BUNDLE", + "CURL_CA_BUNDLE", ): monkeypatch.delenv(name, raising=False) yield @@ -182,7 +184,18 @@ def iter_content(self, chunk_size=8192): def test_same_host_redirect_to_a_private_address_is_not_followed(monkeypatch): - monkeypatch.setattr(http, "resolve_ips", lambda host, timeout: ["127.0.0.1"]) + """Public first, private on the redirect: that pivot is not followed. + + A name whose every address is non-public is a private target and + would be followed. This mock is public for the classification lookup + and private for the hop, which is the rebind the check still blocks. + """ + answers = iter([["93.184.216.34"], ["127.0.0.1"]]) + + def resolve(host, timeout): + return next(answers) + + monkeypatch.setattr(http, "resolve_ips", resolve) calls = [] def fake_request(method, url, headers=None, timeout=None, **kwargs): @@ -294,6 +307,15 @@ def _close_server(server): server.server_close() +def _wait_until(predicate, timeout=1.0): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(0.01) + return False + + class _Listener(threading.Thread): def __init__(self): super().__init__(daemon=True) @@ -540,21 +562,27 @@ def test_private_target_does_not_follow_a_cross_host_redirect(): assert [path for path, _host in server.seen] == ["/r"] -def test_public_name_resolving_private_does_not_follow_redirect(monkeypatch): +def test_private_only_name_follows_same_host_redirect(monkeypatch): + """A name that resolves only to non-public addresses is the scan target. + + Hosts-file lab names (target.htb -> 10.x) and localhost are in this + bucket. The redirect is resolved again and must stay non-public. + """ server = _serve({}) port = server.server_address[1] server.routes = { - "/r": (302, {"Location": f"http://example.com:{port}/secret"}, b""), - "/secret": (200, {}, b"nope"), + "/r": (302, {"Location": f"http://target.htb:{port}/secret"}, b""), + "/secret": (200, {}, b"lab"), } monkeypatch.setattr(http, "resolve_ips", lambda host, timeout: ["127.0.0.1"]) try: - response = http.get(f"http://example.com:{port}/r", timeout=2) + response = http.get(f"http://target.htb:{port}/r", timeout=2) finally: _close_server(server) - assert response.status_code == 302 - assert response.gambit_redirects == 0 - assert [path for path, _host in server.seen] == ["/r"] + assert response.status_code == 200 + assert response.content == b"lab" + assert response.gambit_redirects == 1 + assert [path for path, _host in server.seen] == ["/r", "/secret"] def test_connection_uses_vetted_ips_when_getaddrinfo_changes(monkeypatch): @@ -630,67 +658,341 @@ def close(self): assert response.gambit_redirects == 0 -def test_private_literal_follows_http_to_https_against_a_local_server(): +def test_redirect_to_a_different_high_port_is_not_followed(): + dest = _serve({"/x": (200, {}, b"secret")}) + dest_port = dest.server_address[1] + origin = _serve( + { + "/r": ( + 301, + {"Location": f"http://127.0.0.1:{dest_port}/x"}, + b"stay", + ) + } + ) + try: + response = http.get( + f"http://127.0.0.1:{origin.server_address[1]}/r", timeout=2 + ) + finally: + _close_server(origin) + _close_server(dest) + assert response.status_code == 301 + assert response.gambit_redirects == 0 + assert response.content == b"stay" + assert dest.seen == [] + + +def test_redirect_port_is_the_same_port_or_the_http_https_pair(monkeypatch): + monkeypatch.setattr(http, "resolve_ips", lambda host, timeout: ["10.1.2.3"]) + calls = [] + location = {"value": "/"} + + def fake_request(method, url, headers=None, timeout=None, **kwargs): + calls.append(url) + + class Response: + def __init__(self, url): + self.url = url + done = url.endswith("/x") + self.status_code = 200 if done else 302 + self.headers = {} if done else {"Location": location["value"]} + + def close(self): + return None + + def iter_content(self, chunk_size=8192): + yield b"ok" + + return Response(url) + + monkeypatch.setattr(http.requests, "request", fake_request) + + location["value"] = "https://127.0.0.1:8080/x" + response = http.get("http://127.0.0.1:8080/r") + assert response.gambit_redirects == 1 + assert calls[-1].startswith("https://127.0.0.1:8080/") + + calls.clear() + location["value"] = "http://127.0.0.1:9090/x" + response = http.get("http://127.0.0.1:8080/r") + assert calls == ["http://127.0.0.1:8080/r"] + assert response.status_code == 302 + assert response.gambit_redirects == 0 + + calls.clear() + location["value"] = "https://10.1.2.3/x" + response = http.get("http://10.1.2.3/r") + assert response.gambit_redirects == 1 + assert calls[-1].startswith("https://10.1.2.3/") + + +def test_failed_redirect_hop_returns_the_redirect(monkeypatch): + monkeypatch.setattr(http, "resolve_ips", lambda host, timeout: ["93.184.216.34"]) + calls = [] + + def fake_request(method, url, headers=None, timeout=None, **kwargs): + calls.append(url) + if len(calls) > 1: + raise requests.ConnectionError("hop failed") + + class Response: + status_code = 302 + headers = {"Location": "http://example.com/next"} + + def __init__(self): + self.url = url + + def close(self): + return None + + def iter_content(self, chunk_size=8192): + yield b"stay" + + return Response() + + monkeypatch.setattr(http.requests, "request", fake_request) + response = http.get("http://example.com/") + assert calls == ["http://example.com/", "http://example.com/next"] + assert response.status_code == 302 + assert response.gambit_redirects == 0 + assert response._content == b"stay" + + +def test_localhost_same_host_redirect_is_followed(): + server = _serve( + { + "/r": (302, {"Location": "/x"}, b""), + "/x": (200, {}, b"local"), + } + ) + port = server.server_address[1] + try: + response = http.get(f"http://localhost:{port}/r", timeout=2) + finally: + _close_server(server) + assert response.status_code == 200 + assert response.content == b"local" + assert response.gambit_redirects == 1 + assert [path for path, _host in server.seen] == ["/r", "/x"] + + +def test_no_proxy_host_keeps_pinning_and_follows_redirects(monkeypatch): + server = _serve( + { + "/r": (302, {"Location": "/x"}, b""), + "/x": (200, {}, b"direct"), + } + ) + port = server.server_address[1] + + def blackhole(host, port, family=0, type=0, proto=0, flags=0): + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("203.0.113.1", port or 0))] + + monkeypatch.setattr(socket, "getaddrinfo", blackhole) + monkeypatch.setenv("http_proxy", "http://127.0.0.1:9") + monkeypatch.setenv("https_proxy", "http://127.0.0.1:9") + monkeypatch.setenv("no_proxy", "127.0.0.1,localhost") + try: + response = http.get(f"http://127.0.0.1:{port}/r", timeout=2) + finally: + _close_server(server) + assert response.status_code == 200 + assert response.content == b"direct" + assert response.gambit_redirects == 1 + assert [path for path, _host in server.seen] == ["/r", "/x"] + + +def test_pinned_path_honours_env_ca_bundle_without_trusting_proxies(monkeypatch): + captured = [] + + class FakeSession: + def __init__(self): + self.trust_env = True + + def mount(self, prefix, adapter): + return None + + def request(self, method, url, **kwargs): + captured.append( + ( + self.trust_env, + kwargs.get("verify", "missing"), + kwargs.get("proxies"), + ) + ) + + class Response: + status_code = 200 + headers = {} + + def __init__(self, url): + self.url = url + + def close(self): + return None + + def iter_content(self, chunk_size=8192): + yield b"ok" + + return Response(url) + + def close(self): + return None + + monkeypatch.setattr(http.requests, "Session", FakeSession) + monkeypatch.setattr(http, "resolve_ips", lambda host, timeout: ["93.184.216.34"]) + monkeypatch.setenv("REQUESTS_CA_BUNDLE", "/tmp/gambit-test-ca.pem") + monkeypatch.setenv("https_proxy", "http://127.0.0.1:9") + monkeypatch.setenv("no_proxy", "example.com") + http.get("https://example.com/") + assert captured[-1] == (False, "/tmp/gambit-test-ca.pem", None) + + http.get("https://example.com/", verify=False) + assert captured[-1][1] is False + + http.get("https://example.com/", verify="/explicit.pem") + assert captured[-1][1] == "/explicit.pem" + + monkeypatch.delenv("REQUESTS_CA_BUNDLE", raising=False) + monkeypatch.setenv("CURL_CA_BUNDLE", "/tmp/curl-ca.pem") + http.get("https://example.com/") + assert captured[-1][1] == "/tmp/curl-ca.pem" + + +def test_tls_verify_failure_is_not_retried_over_http(monkeypatch): + attempts = [] + + def fake_request(method, url, **kwargs): + attempts.append(url) + if url.startswith("https://"): + raise requests.exceptions.SSLError( + "[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: Hostname mismatch" + ) + return FakeResponse(url) + + monkeypatch.setattr(http, "request", fake_request) + with pytest.raises(requests.exceptions.SSLError, match="not retrying over HTTP"): + http.get_for_host("internal.example") + assert attempts == ["https://internal.example"] + assert http.preferred_scheme("internal.example") is None + + +def test_tls_protocol_mismatch_still_falls_back_to_http(monkeypatch): + attempts = [] + + def fake_request(method, url, **kwargs): + attempts.append(url) + if url.startswith("https://"): + raise requests.exceptions.SSLError( + "[SSL: WRONG_VERSION_NUMBER] wrong version number" + ) + return FakeResponse(url) + + monkeypatch.setattr(http, "request", fake_request) + response = http.get_for_host("plain.example") + assert response.url == "http://plain.example" + assert attempts == ["https://plain.example", "http://plain.example"] + + +def _mint_hostname_cert(directory: Path, hostname: str) -> tuple[Path, Path, Path]: + """CA plus a server cert whose SAN is ``hostname``. Returns (ca, cert, key).""" openssl = shutil.which("openssl") if openssl is None: pytest.skip("openssl is required to mint a certificate") + ca_key = directory / "ca.key" + ca_cert = directory / "ca.pem" + srv_key = directory / "srv.key" + csr = directory / "srv.csr" + cert = directory / "srv.pem" + subprocess.run( + [ + openssl, "req", "-x509", "-newkey", "rsa:2048", + "-keyout", str(ca_key), "-out", str(ca_cert), + "-days", "1", "-nodes", "-subj", "/CN=Gambit Test CA", + ], + check=True, capture_output=True, cwd=directory, + ) + subprocess.run( + [ + openssl, "req", "-newkey", "rsa:2048", + "-keyout", str(srv_key), "-out", str(csr), + "-nodes", "-subj", f"/CN={hostname}", + "-addext", f"subjectAltName=DNS:{hostname}", + ], + check=True, capture_output=True, cwd=directory, + ) + subprocess.run( + [ + openssl, "x509", "-req", "-in", str(csr), + "-CA", str(ca_cert), "-CAkey", str(ca_key), + "-CAcreateserial", "-out", str(cert), + "-days", "1", "-copy_extensions", "copy", + ], + check=True, capture_output=True, cwd=directory, + ) + return ca_cert, cert, srv_key + + +class _CountingServer(ThreadingHTTPServer): + def __init__(self, server_address, handler): + super().__init__(server_address, handler) + self.accepts = 0 + self.routes = {} + self.seen = [] + + def get_request(self): + # Hostname rejection can abort the handshake inside accept(). + # Count that too: it is still one TCP connection, and a second + # connection would mean the client retried (for example over HTTP). + try: + conn, addr = super().get_request() + except ssl.SSLError: + self.accepts += 1 + raise + self.accepts += 1 + return conn, addr + + +def test_pinned_https_honours_requests_ca_bundle_and_rejects_hostname_mismatch(monkeypatch): + if shutil.which("openssl") is None: + pytest.skip("openssl is required to mint a certificate") with tempfile.TemporaryDirectory() as tmp: - directory = Path(tmp) - key = directory / "key.pem" - cert = directory / "cert.pem" - subprocess.run( - [ - openssl, - "req", - "-x509", - "-newkey", - "rsa:2048", - "-keyout", - str(key), - "-out", - str(cert), - "-days", - "1", - "-nodes", - "-subj", - "/CN=127.0.0.1", - "-addext", - "subjectAltName=IP:127.0.0.1", - ], - check=True, - capture_output=True, - ) - https = ThreadingHTTPServer(("127.0.0.1", 0), _RecordingHandler) - https.routes = {"/x": (200, {}, b"secure")} - https.seen = [] + ca_cert, cert, key = _mint_hostname_cert(Path(tmp), "pinned.gambit.test") + server = _CountingServer(("127.0.0.1", 0), _RecordingHandler) + server.routes = {"/": (200, {}, b"trusted")} context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) context.load_cert_chain(cert, key) - https.socket = context.wrap_socket(https.socket, server_side=True) - https_port = https.server_address[1] - https_thread = threading.Thread(target=https.serve_forever, daemon=True) - https_thread.start() - http_server = _serve( - { - "/r": ( - 301, - {"Location": f"https://127.0.0.1:{https_port}/x"}, - b"", - ) - } - ) + server.socket = context.wrap_socket(server.socket, server_side=True) + port = server.server_address[1] + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + monkeypatch.setattr(http, "resolve_ips", lambda host, timeout: ["127.0.0.1"]) + monkeypatch.setenv("REQUESTS_CA_BUNDLE", str(ca_cert)) try: - with warnings.catch_warnings(): - warnings.simplefilter("ignore", InsecureRequestWarning) - response = http.get( - f"http://127.0.0.1:{http_server.server_address[1]}/r", - timeout=2, - verify=False, - ) + response = http.get(f"https://pinned.gambit.test:{port}/", timeout=2) + assert response.status_code == 200 + assert response.content == b"trusted" + assert _wait_until(lambda: server.accepts >= 1) + good = server.accepts + + with pytest.raises(requests.exceptions.SSLError): + http.get(f"https://wrong.gambit.test:{port}/", timeout=2) + assert _wait_until(lambda: server.accepts >= good + 1) + after_mismatch = server.accepts + assert after_mismatch == good + 1 + + with pytest.raises( + requests.exceptions.SSLError, match="not retrying over HTTP" + ): + http.get_for_host(f"wrong.gambit.test:{port}", timeout=2) + # One TLS handshake. Plain HTTP on the same port is not tried. + assert _wait_until(lambda: server.accepts >= after_mismatch + 1) + assert server.accepts == after_mismatch + 1 + + monkeypatch.delenv("REQUESTS_CA_BUNDLE", raising=False) + with pytest.raises(requests.exceptions.SSLError): + http.get(f"https://pinned.gambit.test:{port}/", timeout=2) finally: - _close_server(http_server) - https.shutdown() - https.server_close() - assert response.status_code == 200 - assert response.content == b"secure" - assert response.gambit_redirects == 1 - assert response.gambit_final_url == f"https://127.0.0.1:{https_port}/x" + server.shutdown() + server.server_close() diff --git a/tests/test_net.py b/tests/test_net.py index db82bbf..b9e3b7c 100644 --- a/tests/test_net.py +++ b/tests/test_net.py @@ -94,6 +94,30 @@ def test_connect_reaches_localhost_from_the_system_resolver(): listener.close() +def test_connect_uses_supplied_addresses_without_resolving(monkeypatch): + seen = [] + + def fail_getaddrinfo(*args, **kwargs): + raise AssertionError("getaddrinfo should not be called") + + def fake_create_connection(address, timeout=None): + seen.append((address, timeout)) + + class Sock: + def close(self): + pass + + return Sock() + + monkeypatch.setattr(socket, "getaddrinfo", fail_getaddrinfo) + monkeypatch.setattr(socket, "create_connection", fake_create_connection) + sock = connect("example.com", 80, 1.5, resolved=["203.0.113.5"]) + assert sock is not None + assert seen == [(("203.0.113.5", 80), 1.5)] + with pytest.raises(OSError): + connect("example.com", 80, 1.5, resolved=[]) + + def test_connect_refuses_a_port_that_would_wrap(monkeypatch): called = [] monkeypatch.setattr( diff --git a/tests/test_ports.py b/tests/test_ports.py index af38a0f..fbd3f86 100644 --- a/tests/test_ports.py +++ b/tests/test_ports.py @@ -3,6 +3,7 @@ ALT_WEB_PORTS, DATABASE_PORTS, PORTS, + TIMEOUT, WEB_PORTS, evaluate, run, @@ -48,17 +49,38 @@ def test_service_for_port(): def test_explicit_port_is_probed_and_out_of_range_port_is_not(monkeypatch): tried = [] - def fake_connect(hostname, port, timeout): - tried.append((hostname, port)) + def fake_connect(hostname, port, timeout, resolved=None): + tried.append((hostname, port, resolved)) raise OSError("closed") monkeypatch.setattr("gambit.net.connect", fake_connect) + monkeypatch.setattr("gambit.net.resolve_ips", lambda host, timeout: ["203.0.113.1"]) assert run("example.com:9090") == [] - assert ("example.com", 9090) in tried + assert ("example.com", 9090, ["203.0.113.1"]) in tried tried.clear() assert run("example.com:99999") == [] - assert all(port != 99999 and port != 34463 for _host, port in tried) - assert ("example.com", 443) in tried + assert all(port != 99999 and port != 34463 for _host, port, _ips in tried) + assert ("example.com", 443, ["203.0.113.1"]) in tried + + +def test_run_resolves_the_host_once(monkeypatch): + resolved_calls = [] + connect_calls = [] + + def fake_resolve(host, timeout): + resolved_calls.append((host, timeout)) + return ["203.0.113.1"] + + def fake_connect(hostname, port, timeout, resolved=None): + connect_calls.append(resolved) + raise OSError("closed") + + monkeypatch.setattr("gambit.net.resolve_ips", fake_resolve) + monkeypatch.setattr("gambit.net.connect", fake_connect) + assert run("example.com") == [] + assert resolved_calls == [("example.com", TIMEOUT)] + assert len(connect_calls) == len(PORTS) + assert all(ips == ["203.0.113.1"] for ips in connect_calls) def test_evaluate_builds_finding_for_open_port():