diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 31dc1fb..8050cb1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -30,3 +30,8 @@ jobs: - name: Run tests run: python -m pytest -q + + - name: Audit pinned dependencies + run: | + python -m pip install pip-audit + python -m pip_audit --requirement requirements.txt --progress-spinner=off diff --git a/gambit/data/takeover_fingerprints.json b/gambit/data/takeover_fingerprints.json index 20162e3..de66036 100644 --- a/gambit/data/takeover_fingerprints.json +++ b/gambit/data/takeover_fingerprints.json @@ -1,21 +1,24 @@ { "version": 1, - "source": "Seeded from can-i-take-over-xyz (https://github.com/EdOverflow/can-i-take-over-xyz); unclaimed-response strings are the services' well-known placeholder/error messages.", + "source": "Seeded from can-i-take-over-xyz (https://github.com/EdOverflow/can-i-take-over-xyz); unclaimed-response strings are the services' well-known placeholder/error messages. claimable=false means a dangling CNAME alone is not reported as a possible takeover. Generic page text (stock IIS 404s, GitLab's ordinary 404, WordPress marketing copy) is omitted so a live site cannot match it by accident.", "services": [ { "service": "GitHub Pages", "cname_suffixes": ["github.io"], - "fingerprints": ["There isn't a GitHub Pages site here."] + "fingerprints": ["There isn't a GitHub Pages site here."], + "claimable": true }, { "service": "GitLab Pages", "cname_suffixes": ["gitlab.io"], - "fingerprints": ["The page you're looking for could not be found."] + "fingerprints": [], + "claimable": false }, { "service": "Bitbucket Pages", "cname_suffixes": ["bitbucket.io"], - "fingerprints": ["Repository not found"] + "fingerprints": ["Repository not found"], + "claimable": true }, { "service": "Amazon S3", @@ -31,12 +34,14 @@ "s3-website.ap-northeast-1.amazonaws.com", "s3-website-sa-east-1.amazonaws.com" ], - "fingerprints": ["NoSuchBucket", "The specified bucket does not exist"] + "fingerprints": ["NoSuchBucket", "The specified bucket does not exist"], + "claimable": true }, { "service": "Heroku", "cname_suffixes": ["herokuapp.com", "herokudns.com"], - "fingerprints": ["No such app", "herokucdn.com/error-pages/no-such-app.html"] + "fingerprints": ["No such app", "herokucdn.com/error-pages/no-such-app.html"], + "claimable": true }, { "service": "Microsoft Azure", @@ -48,77 +53,92 @@ "blob.core.windows.net", "azureedge.net" ], - "fingerprints": ["404 Web Site not found", "The resource you are looking for has been removed, had its name changed, or is temporarily unavailable."] + "fingerprints": ["404 Web Site not found"], + "claimable": false }, { "service": "Shopify", "cname_suffixes": ["myshopify.com"], - "fingerprints": ["Sorry, this shop is currently unavailable."] + "fingerprints": ["Sorry, this shop is currently unavailable."], + "claimable": false }, { "service": "Fastly", "cname_suffixes": ["fastly.net"], - "fingerprints": ["Fastly error: unknown domain"] + "fingerprints": ["Fastly error: unknown domain"], + "claimable": false }, { "service": "Firebase Hosting", "cname_suffixes": ["web.app", "firebaseapp.com"], - "fingerprints": ["Site Not Found"] + "fingerprints": ["Site Not Found"], + "claimable": true }, { "service": "Netlify", "cname_suffixes": ["netlify.app", "netlify.com"], - "fingerprints": ["Not Found - Request ID"] + "fingerprints": ["Not Found - Request ID"], + "claimable": true }, { "service": "Vercel", "cname_suffixes": ["vercel.app", "vercel-dns.com"], - "fingerprints": ["The deployment could not be found", "DEPLOYMENT_NOT_FOUND"] + "fingerprints": ["The deployment could not be found", "DEPLOYMENT_NOT_FOUND"], + "claimable": true }, { "service": "Webflow", "cname_suffixes": ["webflow.io", "proxy-ssl.webflow.com"], - "fingerprints": ["The page you are looking for doesn't exist or has been moved"] + "fingerprints": ["The page you are looking for doesn't exist or has been moved"], + "claimable": false }, { "service": "Ghost", "cname_suffixes": ["ghost.io"], - "fingerprints": ["The thing you were looking for is no longer here, or never was"] + "fingerprints": ["The thing you were looking for is no longer here, or never was"], + "claimable": true }, { "service": "Pantheon", "cname_suffixes": ["pantheonsite.io"], - "fingerprints": ["The gods are wise, but do not know of the site which you seek."] + "fingerprints": ["The gods are wise, but do not know of the site which you seek."], + "claimable": true }, { "service": "Tumblr", "cname_suffixes": ["domains.tumblr.com"], - "fingerprints": ["Whatever you were looking for doesn't currently exist at this address."] + "fingerprints": ["Whatever you were looking for doesn't currently exist at this address."], + "claimable": false }, { "service": "WordPress.com", "cname_suffixes": ["wordpress.com"], - "fingerprints": ["Do you want to register"] + "fingerprints": [], + "claimable": false }, { "service": "Surge", "cname_suffixes": ["surge.sh"], - "fingerprints": ["project not found"] + "fingerprints": ["project not found"], + "claimable": true }, { "service": "Render", "cname_suffixes": ["onrender.com"], - "fingerprints": ["Service Suspended", "This service has been suspended."] + "fingerprints": ["Service Suspended", "This service has been suspended."], + "claimable": false }, { "service": "Readthedocs", "cname_suffixes": ["readthedocs.io"], - "fingerprints": ["Documentation not built yet", "unknown project"] + "fingerprints": ["Documentation not built yet", "unknown project"], + "claimable": true }, { "service": "Zendesk", "cname_suffixes": ["zendesk.com"], - "fingerprints": ["Help Center closed", "this help center no longer exists"] + "fingerprints": ["Help Center closed", "this help center no longer exists"], + "claimable": false } ] } diff --git a/gambit/http.py b/gambit/http.py index ccbf095..b9699b4 100644 --- a/gambit/http.py +++ b/gambit/http.py @@ -11,30 +11,227 @@ files. The scheme that works for a host is remembered for the rest of the process, so only the first module to touch an HTTP-only host pays for the failed HTTPS attempt. + +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. 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. """ +import ipaddress +import os +import socket +import ssl +from typing import NamedTuple +from urllib.parse import urljoin + import requests +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) +_HTTP_HTTPS_PORTS = frozenset((80, 443)) USER_AGENT = ( "GAMBIT/0.1 (+https://github.com/Phantom9869/gambit; authorized testing only)" ) TIMEOUT = 10 # seconds +MAX_REDIRECTS = 5 +MAX_BODY = 1_048_576 # bytes buffered from a response that the caller does not stream # host (as passed in, e.g. "example.com" or "example.com:8080") -> scheme _SCHEME_BY_HOST: dict[str, str] = {} +# host -> non-default TCP port discovered by the liveness pre-flight +_PORT_BY_HOST: dict[str, int] = {} -def request(method: str, url: str, *, timeout: float = TIMEOUT, headers: dict | None = None, **kwargs): +def request( + method: str, + url: str, + *, + timeout: float = TIMEOUT, + headers: dict | None = None, + **kwargs, +): """Send ``method`` to ``url`` with GAMBIT's User-Agent and timeout. ``headers`` (if given) are merged over the default User-Agent. Every - other keyword is forwarded verbatim to :func:`requests.request`. - Network errors propagate as ``requests.RequestException``. + other keyword is forwarded to :func:`requests.request`, except: + + * ``allow_redirects`` (default True) follows only same-host redirects + 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. + * ``stream=True`` leaves the body unread. + + The final response carries ``gambit_redirects`` (hops actually + followed) and ``gambit_final_url``. Network errors propagate as + ``requests.RequestException``. """ merged = {"User-Agent": USER_AGENT} if headers: merged.update(headers) - return requests.request(method, url, headers=merged, timeout=timeout, **kwargs) + allow_redirects = bool(kwargs.pop("allow_redirects", True)) + caller_streams = bool(kwargs.pop("stream", False)) + max_bytes = _as_limit(kwargs.pop("max_bytes", MAX_BODY)) + kwargs["allow_redirects"] = False + kwargs["stream"] = True + proxies = kwargs.get("proxies") + + current = url + current_method = method + original_host: 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 + 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 = 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) + 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 + 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 + 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 + if status == 303: + current_method = "GET" + # 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 def get(url: str, *, timeout: float = TIMEOUT, headers: dict | None = None, **kwargs): @@ -52,15 +249,33 @@ def preferred_scheme(host: str) -> str | None: return _SCHEME_BY_HOST.get(host) +def preferred_port(host: str) -> int | None: + """Non-default port discovered for ``host``, if the pre-flight found one.""" + return _PORT_BY_HOST.get(host) + + def set_preferred_scheme(host: str, scheme: str) -> None: """Remember that ``scheme`` ("https"/"http") works for ``host``.""" if scheme in ("https", "http"): _SCHEME_BY_HOST[host] = scheme +def set_preferred_port(host: str, port: int | None) -> None: + """Remember a non-default port for ``host``, or forget it when ``None``.""" + if port is None: + _PORT_BY_HOST.pop(host, None) + return + number = int(port) + if 1 <= number <= 65535 and number not in (80, 443): + _PORT_BY_HOST[host] = number + else: + _PORT_BY_HOST.pop(host, None) + + def clear_scheme_cache() -> None: - """Forget all remembered schemes (mainly for tests).""" + """Forget remembered schemes and discovered ports (mainly for tests).""" _SCHEME_BY_HOST.clear() + _PORT_BY_HOST.clear() def _schemes_to_try(host: str) -> list[str]: @@ -70,23 +285,60 @@ def _schemes_to_try(host: str) -> list[str]: return ["https", "http"] +def url_authority(target: str, scheme: str) -> str: + """``host`` or ``host:port`` (IPv6 bracketed) for ``target``. + + An explicit valid port on ``target`` is always kept. Otherwise a port + remembered by the pre-flight is included when it is not the default + for ``scheme``. An out-of-range explicit port is omitted. + """ + from gambit.pipeline import format_url_host, port_out_of_range, split_host_port + + if port_out_of_range(target): + name, _port = split_host_port(target) + return format_url_host(name or target, None) + name, explicit = split_host_port(target) + if not name: + return target + if explicit is not None: + return format_url_host(name, explicit) + discovered = preferred_port(target) + if discovered is None: + return format_url_host(name, None) + if (scheme == "https" and discovered == 443) or (scheme == "http" and discovered == 80): + return format_url_host(name, None) + return format_url_host(name, discovered) + + 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 last_exc: Exception | None = None for scheme in _schemes_to_try(target): + url = f"{scheme}://{url_authority(target, scheme)}{path}" try: - response = request(method, f"{scheme}://{target}{path}", **kwargs) + 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) return response - assert last_exc is not None # _schemes_to_try is never empty + if last_exc is None: + raise requests.RequestException(f"no request attempted for {target}") raise last_exc @@ -100,15 +352,520 @@ def options_for_host(target: str, path: str = "", **kwargs): return _request_for_host("OPTIONS", target, path, **kwargs) +def read_limited(response, limit: int) -> bytes: + """Read at most ``limit`` bytes from a response body. + + Uses ``iter_content`` when the response has it, so a multi-gigabyte + body is not buffered. A response with no streaming helper contributes + its ``content`` attribute, still sliced to ``limit``. + """ + if limit < 0: + limit = 0 + reader = getattr(response, "iter_content", None) + if not callable(reader): + raw = getattr(response, "content", b"") or b"" + if isinstance(raw, str): + raw = raw.encode("utf-8", errors="replace") + return bytes(raw[:limit]) + chunks: list[bytes] = [] + total = 0 + for chunk in reader(chunk_size=8192): + if not chunk: + continue + if isinstance(chunk, str): + chunk = chunk.encode("utf-8", errors="replace") + remaining = limit - total + if remaining <= 0: + break + if len(chunk) > remaining: + chunks.append(chunk[:remaining]) + break + chunks.append(chunk) + total += len(chunk) + return b"".join(chunks) + + +def annotate_findings(findings, response): + """Append the final URL when this response followed a redirect. + + Findings are unchanged when no redirect was followed, so graders that + never see a live response keep their evidence text. + """ + if not findings: + return findings or [] + hops = int(getattr(response, "gambit_redirects", 0) or 0) + final = getattr(response, "gambit_final_url", "") or "" + if hops <= 0 or not final: + return findings + for finding in findings: + if final not in (finding.evidence or ""): + finding.evidence = f"{finding.evidence} (final URL: {final})" + return findings + + +def _as_limit(value) -> int: + try: + number = int(value) + except (TypeError, ValueError): + return MAX_BODY + if number < 0: + return 0 + return number + + +def _timeout_seconds(timeout) -> float: + if isinstance(timeout, (tuple, list)) and timeout: + return float(timeout[0]) + return float(timeout) + + +def _stamp_redirect(response, hops: int, url: str) -> None: + try: + response.gambit_redirects = hops + response.gambit_final_url = url + except Exception: + return + + +def _location_header(response) -> str: + headers = getattr(response, "headers", None) or {} + getter = getattr(headers, "get", None) + if not callable(getter): + return "" + for name in ("Location", "location"): + value = getter(name) + if value: + return str(value) + return "" + + +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: + """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 + 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 + + +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. 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 + 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 + 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: + """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: + 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 _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. 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 + 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 None + if not _redirect_ips_allowed(original_private, ips): + return None + return list(ips) + + +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. + + ``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, **send_kwargs + ) + finally: + session.close() + + +def _close_quietly(response) -> None: + closer = getattr(response, "close", None) + if not callable(closer): + return + try: + closer() + except Exception: + return + + +def _buffer_limited(response, limit: int) -> None: + if not callable(getattr(response, "iter_content", None)): + return + try: + data = read_limited(response, limit) + except Exception: + return + try: + response._content = data + response._content_consumed = True + except Exception: + return + _close_quietly(response) + + __all__ = [ + "MAX_BODY", + "MAX_REDIRECTS", "TIMEOUT", "USER_AGENT", + "annotate_findings", "clear_scheme_cache", "get", "get_for_host", "options", "options_for_host", + "preferred_port", "preferred_scheme", + "read_limited", "request", + "set_preferred_port", "set_preferred_scheme", + "url_authority", ] diff --git a/gambit/modules/baseline_headers.py b/gambit/modules/baseline_headers.py index 1cbe114..cc9d703 100644 --- a/gambit/modules/baseline_headers.py +++ b/gambit/modules/baseline_headers.py @@ -2,7 +2,7 @@ import requests -from gambit.http import TIMEOUT, USER_AGENT, get_for_host +from gambit.http import TIMEOUT, USER_AGENT, annotate_findings, get_for_host from gambit.models import Finding, Severity # Re-exported for backwards compatibility; canonical values live in gambit.http. @@ -85,4 +85,4 @@ def run(target: str) -> list[Finding]: next_step="Check the hostname and whether the site is reachable.", ) ] - return check_headers(target, response.headers) + return annotate_findings(check_headers(target, response.headers), response) diff --git a/gambit/modules/cors.py b/gambit/modules/cors.py index 768e7a4..42f8200 100644 --- a/gambit/modules/cors.py +++ b/gambit/modules/cors.py @@ -10,7 +10,7 @@ import requests -from gambit.http import TIMEOUT, USER_AGENT, get_for_host +from gambit.http import TIMEOUT, USER_AGENT, annotate_findings, get_for_host from gambit.models import Finding, Severity # Re-exported for backwards compatibility; canonical values live in gambit.http. @@ -140,4 +140,4 @@ def run(target: str) -> list[Finding]: next_step="Check the hostname and whether the site is reachable.", ) ] - return check_cors(target, response.headers) + return annotate_findings(check_cors(target, response.headers), response) diff --git a/gambit/modules/fingerprint.py b/gambit/modules/fingerprint.py index eb10e68..a67f323 100644 --- a/gambit/modules/fingerprint.py +++ b/gambit/modules/fingerprint.py @@ -11,7 +11,7 @@ import requests -from gambit.http import TIMEOUT, USER_AGENT, get_for_host +from gambit.http import TIMEOUT, USER_AGENT, annotate_findings, get_for_host from gambit.models import Finding, Severity # Re-exported for backwards compatibility; canonical values live in gambit.http. @@ -265,4 +265,4 @@ def run(target: str) -> list[Finding]: except Exception: body = "" cookie_names = _cookie_names(getattr(response, "cookies", None)) - return identify(target, response.headers, cookie_names, body) + return annotate_findings(identify(target, response.headers, cookie_names, body), response) diff --git a/gambit/modules/gather.py b/gambit/modules/gather.py index 1c7b49b..1807f4a 100644 --- a/gambit/modules/gather.py +++ b/gambit/modules/gather.py @@ -21,7 +21,7 @@ from gambit.http import TIMEOUT, USER_AGENT, get from gambit.models import Finding, Severity -from gambit.pipeline import normalize_host +from gambit.pipeline import normalize_host, split_host_port __all__ = [ "RECORD_TYPES", @@ -185,6 +185,12 @@ def evaluate_dns( return findings +def _dns_name(host: str) -> str: + """Hostname only: ``example.com:8080`` is not a DNS name.""" + name, _port = split_host_port(host) + return name + + def _resolve(name: str, rdtype: str) -> list[str]: """Resolve one record type; every DNS failure means "no values".""" try: @@ -201,7 +207,7 @@ def lookup_dns(host: str) -> dict[str, list[str]]: records of ``_dmarc.`` (stored under ``"DMARC"``). Types with no answer are omitted. DNS errors never propagate. """ - name = normalize_host(host) + name = _dns_name(host) records: dict[str, list[str]] = {} if not name: return records @@ -229,7 +235,11 @@ def _fetch_crtsh_subdomains(host: str) -> list[str]: raise cached return cached try: - response = get(_CRT_SH_URL.format(host=host), timeout=TIMEOUT) + response = get( + _CRT_SH_URL.format(host=host), + timeout=TIMEOUT, + allow_redirects=False, + ) response.raise_for_status() names = parse_crtsh(response.json()) except Exception as exc: @@ -259,7 +269,7 @@ def discover_subdomains(target: str) -> list[str]: called with the target, returns host strings (sorted, deduped, apex excluded), and never raises. """ - host = normalize_host(target) + host = _dns_name(target) if not host: return [] try: @@ -272,10 +282,11 @@ def discover_subdomains(target: str) -> list[str]: def run(target: str) -> list[Finding]: """Report DNS records for the target and its crt.sh subdomains.""" host = normalize_host(target) or target - findings = evaluate_dns(host, lookup_dns(host)) + dns_host = _dns_name(host) or host + findings = evaluate_dns(host, lookup_dns(dns_host)) try: - names = _fetch_crtsh_subdomains(host) + names = _fetch_crtsh_subdomains(dns_host) except (requests.RequestException, ValueError) as exc: findings.append( Finding( @@ -289,7 +300,7 @@ def run(target: str) -> list[Finding]: ) ) else: - subdomains = _own_subdomains(host, names) + subdomains = _own_subdomains(dns_host, names) if subdomains: findings.append( Finding( diff --git a/gambit/modules/http_methods.py b/gambit/modules/http_methods.py index 0f53370..fb26535 100644 --- a/gambit/modules/http_methods.py +++ b/gambit/modules/http_methods.py @@ -12,7 +12,7 @@ import requests -from gambit.http import TIMEOUT, USER_AGENT, options_for_host +from gambit.http import TIMEOUT, USER_AGENT, annotate_findings, options_for_host from gambit.models import Finding, Severity # Re-exported for consistency; canonical values live in gambit.http. @@ -151,4 +151,4 @@ def run(target: str) -> list[Finding]: next_step="Check the hostname and whether the site is reachable.", ) ] - return check_methods(target, response.headers) + return annotate_findings(check_methods(target, response.headers), response) diff --git a/gambit/modules/ports.py b/gambit/modules/ports.py index 3e1830a..964622f 100644 --- a/gambit/modules/ports.py +++ b/gambit/modules/ports.py @@ -6,8 +6,6 @@ the interesting logic testable without any network access. """ -import socket - from gambit.models import Finding, Severity __all__ = [ @@ -107,17 +105,32 @@ def evaluate(host: str, port: int) -> Finding: def run(target: str) -> list[Finding]: - """Try one TCP connect per port in PORTS and report the open ones.""" + """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`. 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, resolve_ips from gambit.pipeline import split_host_port - hostname, _explicit_port = split_host_port(target) + hostname, explicit_port = split_host_port(target) host = hostname or target + 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: + for port in ports: try: - with socket.create_connection((host, port), timeout=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/modules/takeover.py b/gambit/modules/takeover.py index b92d6a1..f59e457 100644 --- a/gambit/modules/takeover.py +++ b/gambit/modules/takeover.py @@ -30,7 +30,7 @@ from gambit.http import TIMEOUT, USER_AGENT, get_for_host from gambit.models import Finding, Severity -from gambit.pipeline import normalize_host +from gambit.pipeline import normalize_host, split_host_port __all__ = [ "FINGERPRINT_PATH", @@ -56,11 +56,18 @@ @dataclass(frozen=True) class Fingerprint: - """One third-party service a dangling CNAME can point at.""" + """One third-party service a dangling CNAME can point at. + + ``claimable`` means a name that simply fails to resolve is enough to + call the case a possible takeover (GitHub Pages, S3, Heroku, …). + Services that are not claimable that way stay INFO unless a specific + unclaimed-page fingerprint is served with an HTTP error status. + """ service: str cname_suffixes: tuple[str, ...] fingerprints: tuple[str, ...] + claimable: bool = True def _as_str_tuple(raw) -> tuple[str, ...]: @@ -115,6 +122,11 @@ def load_fingerprints(path=None) -> list[Fingerprint]: ) if not service or not str(service).strip() or not suffixes: continue + raw_claimable = entry.get("claimable", True) + if isinstance(raw_claimable, str): + claimable = raw_claimable.strip().lower() in ("1", "true", "yes") + else: + claimable = bool(raw_claimable) fingerprints.append( Fingerprint( service=str(service).strip(), @@ -124,6 +136,7 @@ def load_fingerprints(path=None) -> list[Fingerprint]: or entry.get("unclaimed_fingerprints") or entry.get("fingerprint") ), + claimable=claimable, ) ) return fingerprints @@ -178,12 +191,13 @@ def evaluate( ``resolves`` says whether the CNAME destination resolves (``True``/``False``; ``None`` = unknown), ``body`` is the host's - HTTP response body if one was fetched (``status_code`` is accepted - for evidence context and does not change the verdict). Returns - ``None`` when there is no CNAME or it points at an unknown service. - HIGH requires a known service *plus* an unresolvable destination - or a served unclaimed fingerprint; anything else for a known - service is INFO. + HTTP response body if one was fetched. ``status_code``, when + provided, gates the fingerprint: a 2xx or 3xx response does not + count as an unclaimed page. ``None`` leaves that check unchanged. + Returns ``None`` when there is no CNAME or it points at an unknown + service. HIGH requires a known service plus either a claimable + destination that does not resolve, or a served unclaimed + fingerprint. Anything else for a known service is INFO. """ destination = normalize_host(cname or "") if not destination: @@ -192,11 +206,19 @@ def evaluate( if fingerprint is None: return None - matched = find_unclaimed_fingerprint(body, fingerprint) + # A 2xx/3xx page that happens to quote a fingerprint string is not an + # unclaimed service. Unknown status (callers that only have a body) + # still counts, so a fingerprint match is not silently dropped. + status_supports_fingerprint = status_code is None or status_code >= 400 + matched = ( + find_unclaimed_fingerprint(body, fingerprint) + if status_supports_fingerprint + else None + ) reasons = [] if matched is not None: reasons.append(f"response serves the unclaimed fingerprint {matched!r}") - if resolves is False: + if resolves is False and fingerprint.claimable: reasons.append("the CNAME destination does not resolve") if reasons: @@ -216,14 +238,20 @@ def evaluate( "yours and abandoned, remove the CNAME or reclaim the resource. Report " "it to the owner; do not claim a resource that is not yours.", ) + if resolves is True: + observed = "destination resolves and no unclaimed fingerprint was observed" + elif resolves is False: + observed = ( + "destination does not resolve, but this service is not treated " + "as claimable from DNS alone" + ) + else: + observed = "no unclaimed fingerprint was observed" return Finding( severity=Severity.INFO, host=host, title=f"CNAME points to {fingerprint.service}", - evidence=f"CNAME {host} -> {destination}; destination resolves and no " - "unclaimed fingerprint was observed" - if resolves is True - else f"CNAME {host} -> {destination}; no unclaimed fingerprint was observed", + evidence=f"CNAME {host} -> {destination}; {observed}", why="Third-party CNAMEs become a takeover risk when the resource behind " "them is abandoned, so this dependency is worth tracking.", next_step="Confirm the service resource behind this host is still owned " @@ -233,7 +261,7 @@ def evaluate( def lookup_cname(host: str) -> Optional[str]: """Return the host's first CNAME destination (no trailing dot), or None.""" - name = normalize_host(host) + name, _port = split_host_port(host) if not name: return None try: @@ -280,11 +308,18 @@ def run(target: str) -> list[Finding]: resolves = _resolves(cname) body = None status_code = None + response = None if resolves is not False: try: - response = get_for_host(host, timeout=TIMEOUT, allow_redirects=True) + response = get_for_host( + host, + timeout=TIMEOUT, + allow_redirects=True, + max_bytes=_MAX_BODY, + ) except requests.RequestException: body = None + response = None else: status_code = response.status_code body = (response.text or "")[:_MAX_BODY] @@ -297,4 +332,8 @@ def run(target: str) -> list[Finding]: fingerprints=fingerprints, status_code=status_code, ) + if finding and response is not None: + from gambit.http import annotate_findings + + annotate_findings([finding], response) return [finding] if finding else [] diff --git a/gambit/modules/tls.py b/gambit/modules/tls.py index a93dd1e..f065b1c 100644 --- a/gambit/modules/tls.py +++ b/gambit/modules/tls.py @@ -9,8 +9,9 @@ collected over an unverified handshake for the expiry/version checks. """ -import socket +import os import ssl +import tempfile from collections.abc import Mapping from datetime import date, datetime, time, timezone @@ -27,6 +28,7 @@ "evaluate_tls", "parse_cert_datetime", "days_until_expiry", + "certificate_from_socket", "run", ] @@ -296,10 +298,76 @@ def _split_host_port(target: str) -> tuple[str, int]: return name, port if port is not None else 443 +def certificate_from_socket(tls_sock) -> dict: + """Peer certificate as a ``getpeercert()`` dict. + + ``getpeercert()`` with no arguments returns ``{}`` when the handshake + did not validate the certificate, which is exactly the unverified + second connection this module uses to read an expired or self-signed + cert. The DER form is still available; decode that instead of + reporting that no certificate was presented. + """ + try: + cert = tls_sock.getpeercert() or {} + except Exception: + cert = {} + if cert: + return cert + try: + der = tls_sock.getpeercert(binary_form=True) + except Exception: + return {} + if not der: + return {} + return _cert_dict_from_der(der) + + +def _cert_dict_from_der(der: bytes) -> dict: + """Decode DER bytes into the dict shape ``getpeercert()`` uses. + + The stdlib exposes that decoder only as ``ssl._ssl._test_decode_cert``, + which reads a PEM file. The file is created privately and removed + before this function returns; nothing is sent on the socket. + """ + try: + pem = ssl.DER_cert_to_PEM_cert(der) + except Exception: + return {} + fd, path = tempfile.mkstemp(prefix="gambit-cert-", suffix=".pem") + try: + os.write(fd, pem.encode("ascii")) + os.close(fd) + fd = -1 + decoded = ssl._ssl._test_decode_cert(path) + except Exception: + return {} + finally: + if fd >= 0: + os.close(fd) + try: + os.unlink(path) + except OSError: + pass + return decoded or {} + + def _handshake(hostname: str, port: int, context: ssl.SSLContext): - with socket.create_connection((hostname, port), timeout=TIMEOUT) as sock: - with context.wrap_socket(sock, server_hostname=hostname) as tls_sock: - return tls_sock.getpeercert(), tls_sock.version() + from gambit.net import connect + + sock = connect(hostname, port, TIMEOUT) + try: + tls_sock = context.wrap_socket(sock, server_hostname=hostname) + except Exception: + closer = getattr(sock, "close", None) + if callable(closer): + closer() + raise + try: + return certificate_from_socket(tls_sock), tls_sock.version() + finally: + closer = getattr(tls_sock, "close", None) + if callable(closer): + closer() def _unverified_context() -> ssl.SSLContext: diff --git a/gambit/net.py b/gambit/net.py new file mode 100644 index 0000000..c97b3e5 --- /dev/null +++ b/gambit/net.py @@ -0,0 +1,133 @@ +"""Network helpers with a deadline on DNS, not only on the TCP handshake. + +``socket.create_connection`` resolves names with ``getaddrinfo`` before it +starts the connect timer, so a stuck resolver can hang a scan forever. +: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 threading + + +def is_ip_address(hostname: str) -> bool: + """True when ``hostname`` is an IPv4 or IPv6 literal.""" + try: + ipaddress.ip_address(hostname) + except ValueError: + return False + return True + + +def is_public_ip(ip: str) -> bool: + """True when ``ip`` is a globally routable unicast address. + + Loopback, private, link-local (including cloud metadata), reserved, + and documentation ranges are rejected. ``ipaddress``'s ``is_global`` + is the single check so CGNAT and TEST-NET stay on the private side. + """ + try: + addr = ipaddress.ip_address(ip) + except ValueError: + return False + return bool(getattr(addr, "is_global", False)) + + +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 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 _family, socktype, _proto, _canon, sockaddr in infos: + if socktype != socket.SOCK_STREAM: + continue + ip = sockaddr[0] + if ip and ip not in seen: + seen.add(ip) + found.append(ip) + return found + + +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. + 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 = 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 + for ip in ips: + try: + return socket.create_connection((ip, number), timeout=timeout) + except OSError as exc: + last = exc + if last is None: + raise OSError(f"could not connect to {hostname}:{number}") + raise last + + +__all__ = ["connect", "is_ip_address", "is_public_ip", "resolve_ips"] diff --git a/gambit/pipeline.py b/gambit/pipeline.py index 2c4baa5..31b2911 100644 --- a/gambit/pipeline.py +++ b/gambit/pipeline.py @@ -13,6 +13,7 @@ finding for that host and everything else keeps going. """ +import ipaddress import time from typing import Callable, Iterable, Optional, Protocol @@ -35,8 +36,9 @@ def normalize_host(host: str) -> str: Strips whitespace, any scheme (``https://``), userinfo, path/query/ fragment, a leading wildcard (``*.``), and a trailing dot; lowercases - the result. A port, if present, is preserved (``example.com:8080``), - because existing modules interpolate the host straight into a URL. + the result. A valid port is preserved (``example.com:8080``). IPv6 + literals are rewritten with brackets (``::1`` and ``[::1]`` both + become ``[::1]``) so they survive being placed in a URL. """ if not host: return "" @@ -49,6 +51,79 @@ def normalize_host(host: str) -> str: if "@" in value: # userinfo, e.g. user@example.com value = value.rsplit("@", 1)[1] value = value.strip().lower() + if value.startswith("*."): + value = value[2:] + if value.endswith("."): + value = value[:-1] + return _canonicalize_ip_literal(value.strip()) + + +def _canonicalize_ip_literal(value: str) -> str: + """Bracket a bare IPv6 literal. Leave hostnames and ``host:port`` alone.""" + if not value: + return value + if value.startswith("["): + end = value.find("]") + if end == -1: + return value + inner = value[1:end] + rest = value[end + 1 :] + try: + ipaddress.IPv6Address(inner) + except ValueError: + return value + if rest.startswith(":") and rest[1:].isdigit(): + number = int(rest[1:]) + if 1 <= number <= 65535: + return f"[{inner}]:{number}" + return f"[{inner}]" + if rest == "": + return f"[{inner}]" + return value + if value.count(":") >= 2: + try: + ipaddress.IPv6Address(value) + except ValueError: + return value + return f"[{value}]" + return value + + +def port_out_of_range(host: str) -> bool: + """True when ``host`` carries a numeric port outside 1–65535. + + ``getaddrinfo`` truncates such ports (99999 becomes 34463). Callers + must refuse them instead of connecting. Checked on the undecorated + text, because :func:`normalize_host` drops an invalid bracketed port + while rewriting an IPv6 literal. + """ + raw = _host_without_decoration(host) + if raw.startswith("[") and "]:" in raw: + rest = raw.split("]", 1)[1] + if rest.startswith(":") and rest[1:].isdigit(): + number = int(rest[1:]) + return not (1 <= number <= 65535) + return False + if raw.count(":") == 1 and not raw.startswith("["): + name, port = raw.split(":", 1) + if name and port.isdigit(): + number = int(port) + return not (1 <= number <= 65535) + return False + + +def _host_without_decoration(host: str) -> str: + """``normalize_host`` steps that do not rewrite IPv6 literals.""" + if not host: + return "" + value = str(host).strip() + if "://" in value: + value = value.split("://", 1)[1] + for sep in ("/", "?", "#"): + value = value.split(sep, 1)[0] + if "@" in value: + value = value.rsplit("@", 1)[1] + value = value.strip().lower() if value.startswith("*."): value = value[2:] if value.endswith("."): @@ -60,44 +135,95 @@ def split_host_port(host: str) -> tuple[str, int | None]: """Split a target into (hostname, port-or-None). No network access. Accepts anything :func:`normalize_host` accepts (scheme, path, - userinfo, trailing dot are stripped first). Socket-level code should - use the hostname from here, never the raw target: a raw - ``"example.com:8080"`` is not a resolvable name. + userinfo, trailing dot are stripped first). The hostname never + includes URL brackets or a port: socket calls want ``::1``, not + ``[::1]``. A bare IPv6 address is not mistaken for ``host:port``. + A numeric port outside 1–65535 is dropped (returned as ``None``) + so it cannot be truncated into a different port. """ + if port_out_of_range(host): + normalized = normalize_host(host) + if normalized.startswith("[") and normalized.endswith("]"): + return normalized[1:-1], None + if ":" in normalized and not normalized.startswith("["): + name, _sep, port = normalized.rpartition(":") + if port.isdigit() and name: + return name, None + return normalized, None normalized = normalize_host(host) + if normalized.startswith("["): + end = normalized.find("]") + if end != -1: + inner = normalized[1:end] + rest = normalized[end + 1 :] + if rest.startswith(":") and rest[1:].isdigit(): + number = int(rest[1:]) + if 1 <= number <= 65535: + return inner, number + if rest == "": + return inner, None name, sep, port = normalized.rpartition(":") - if sep and name and port.isdigit(): - return name, int(port) + if sep and name and port.isdigit() and ":" not in name: + number = int(port) + if 1 <= number <= 65535: + return name, number + return name, None + if normalized.startswith("[") and normalized.endswith("]"): + return normalized[1:-1], None return normalized, None +def format_url_host(name: str, port: int | None = None) -> str: + """Host text safe to place in an ``http(s)://`` URL. + + IPv6 literals are bracketed. ``port`` is appended only when it is + inside 1–65535. + """ + try: + ipaddress.IPv6Address(name) + literal = f"[{name}]" + except ValueError: + literal = name + if port is not None and 1 <= int(port) <= 65535: + return f"{literal}:{int(port)}" + return literal + + def host_reachable(host: str, *, timeout: float = 1.0) -> bool: """Cheap liveness pre-flight: does the host answer on any web port? Tries the target's explicit port if it has one, else the common web - ports (443, 8443, 80, 8080, 8000, 3000) with a short timeout. This - exists so a filtered/dead host costs a few seconds instead of every - web module's full timeout stacked together. Also primes the HTTP - scheme cache: if only a plain-HTTP port answers, later web modules - go straight to HTTP. Only gates web-fetch modules -- takeover and - port checks must run even for hosts that answer on no web port. + ports (443, 8443, 80, 8080, 8000, 3000) with a short timeout. DNS is + bounded by that same timeout. An explicit port outside 1–65535 is + refused rather than truncated. + + The open port is remembered for later HTTP fetches: success on 8443 + (or 8080, 8000, 3000) makes those requests use that port, not 443/80. + Only gates web-fetch modules -- takeover and port checks must run + even for hosts that answer on no web port. """ - import socket - - from gambit.http import set_preferred_scheme + from gambit.http import set_preferred_port, set_preferred_scheme + from gambit.net import connect + if port_out_of_range(host): + return False name, explicit_port = split_host_port(host) if not name: return False ports = [explicit_port] if explicit_port else [443, 8443, 80, 8080, 8000, 3000] for port in ports: + if port is None: + continue try: - with socket.create_connection((name, port), timeout=timeout): + with connect(name, port, timeout): pass except OSError: continue if explicit_port is None: set_preferred_scheme(host, "https" if port in (443, 8443) else "http") + # 443 and 80 are the scheme defaults. Any other open port has + # to be written into the URL or the fetch hits a closed port. + set_preferred_port(host, None if port in (443, 80) else port) return True return False @@ -275,8 +401,10 @@ def run_pipeline( "ModuleLike", "dedupe_hosts", "discover_hosts", + "format_url_host", "host_reachable", "normalize_host", + "port_out_of_range", "run_for_host", "run_modules_for_host", "run_pipeline", diff --git a/gambit/triage.py b/gambit/triage.py index 73551b4..8814a1b 100644 --- a/gambit/triage.py +++ b/gambit/triage.py @@ -1,15 +1,37 @@ """Triage: sort findings by severity and format them for humans or JSON.""" import json +import re from gambit.models import Finding +# Target-controlled text (headers, generator tags, DNS TXT) must not be +# able to open a new line or a new Markdown construct in the report. +_CONTROLS = re.compile(r"[\x00-\x1f\x7f]") +_MARKDOWN_SPECIAL = set("\\`*_{}[]()#+!<>") + def rank(findings: list[Finding]) -> list[Finding]: """Highest severity first; ties broken by host, then title.""" return sorted(findings, key=lambda f: (-int(f.severity), f.host, f.title)) +def _single_line(value: str) -> str: + """Collapse control characters so one field cannot become two lines.""" + return _CONTROLS.sub(" ", value or "").strip() + + +def _markdown(value: str) -> str: + """Escape Markdown and HTML punctuation after collapsing control characters.""" + text = ( + _single_line(value) + .replace("&", "&") + .replace("<", "<") + .replace(">", ">") + ) + return "".join("\\" + ch if ch in _MARKDOWN_SPECIAL else ch for ch in text) + + def to_text(findings: list[Finding]) -> str: """Human-readable report, most severe findings first.""" if not findings: @@ -18,10 +40,10 @@ def to_text(findings: list[Finding]) -> str: blocks = [] for f in rank(findings): blocks.append( - f"[{f.severity.name}] {f.host}: {f.title}\n" - f" Evidence: {f.evidence}\n" - f" Why: {f.why}\n" - f" Next: {f.next_step}" + f"[{f.severity.name}] {_single_line(f.host)}: {_single_line(f.title)}\n" + f" Evidence: {_single_line(f.evidence)}\n" + f" Why: {_single_line(f.why)}\n" + f" Next: {_single_line(f.next_step)}" ) return "\n\n".join(blocks) @@ -31,7 +53,8 @@ def to_markdown(findings: list[Finding]) -> str: Table-free: a heading per finding in the README example style (``[SEVERITY] host: title``) followed by Evidence/Why/Next bullets, - so the report reads well pasted into notes or an issue. + so the report reads well pasted into notes or an issue. Field values + are escaped so a target cannot inject headings, links, or HTML. """ if not findings: return "# GAMBIT report\n\nNo findings." @@ -39,10 +62,10 @@ def to_markdown(findings: list[Finding]) -> str: blocks = ["# GAMBIT report"] for f in rank(findings): blocks.append( - f"## [{f.severity.name}] {f.host}: {f.title}\n\n" - f"- **Evidence:** {f.evidence}\n" - f"- **Why:** {f.why}\n" - f"- **Next:** {f.next_step}" + f"## [{f.severity.name}] {_markdown(f.host)}: {_markdown(f.title)}\n\n" + f"- **Evidence:** {_markdown(f.evidence)}\n" + f"- **Why:** {_markdown(f.why)}\n" + f"- **Next:** {_markdown(f.next_step)}" ) return "\n\n".join(blocks) diff --git a/pyproject.toml b/pyproject.toml index 8c73124..ba88463 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,8 +14,9 @@ authors = [ ] keywords = ["recon", "pentest", "security", "triage"] dependencies = [ - "requests", - "dnspython", + "dnspython==2.8.0", + "requests==2.34.2", + "urllib3==2.8.0", ] [project.optional-dependencies] diff --git a/requirements.txt b/requirements.txt index 813c161..dcc0354 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,3 @@ dnspython==2.8.0 requests==2.34.2 +urllib3==2.8.0 diff --git a/tests/test_fingerprint.py b/tests/test_fingerprint.py index 9b7f69b..a458fbc 100644 --- a/tests/test_fingerprint.py +++ b/tests/test_fingerprint.py @@ -103,6 +103,24 @@ def test_generator_meta_tag_both_attribute_orders(): assert any("Joomla! 4" in t for t in titles(identify("example.com", {}, [], body2))) +def test_run_records_final_url_after_a_same_host_redirect(monkeypatch): + from gambit.modules import fingerprint + + class Response: + status_code = 200 + headers = {"Server": "nginx"} + text = "" + cookies = [] + url = "https://example.com/app" + gambit_redirects = 1 + gambit_final_url = "https://example.com/app" + + monkeypatch.setattr(fingerprint, "get_for_host", lambda *args, **kwargs: Response()) + findings = fingerprint.run("example.com") + assert findings + assert any("final URL: https://example.com/app" in finding.evidence for finding in findings) + + def test_findings_carry_host(): findings = identify("sub.example.com", {"Server": "nginx"}, [], "") assert all(f.host == "sub.example.com" for f in findings) diff --git a/tests/test_gather.py b/tests/test_gather.py index 8113e81..f6ae319 100644 --- a/tests/test_gather.py +++ b/tests/test_gather.py @@ -140,3 +140,36 @@ def fake_get(url, **kwargs): assert len(calls) == 1 # second call served from cache finally: gather.clear_crtsh_cache() + + +def test_dns_and_crtsh_drop_an_explicit_port(monkeypatch): + seen = [] + + def fake_resolve(name, rdtype): + seen.append(name) + return [] + + monkeypatch.setattr(gather, "_resolve", fake_resolve) + gather.lookup_dns("example.com:8080") + assert seen + assert all(":" not in name for name in seen) + assert "_dmarc.example.com" in seen + + gather.clear_crtsh_cache() + calls = [] + + class FakeResponse: + def raise_for_status(self): + pass + + def json(self): + return [{"name_value": "www.example.com"}] + + monkeypatch.setattr(gather, "get", lambda url, **kwargs: calls.append(url) or FakeResponse()) + try: + assert gather.discover_subdomains("example.com:8080") == ["www.example.com"] + finally: + gather.clear_crtsh_cache() + assert calls + assert ":8080" not in calls[0] + assert "q=%25.example.com&" in calls[0] diff --git a/tests/test_http.py b/tests/test_http.py index 60218b7..1cdb1d9 100644 --- a/tests/test_http.py +++ b/tests/test_http.py @@ -1,5 +1,16 @@ +import shutil +import socket +import ssl +import subprocess +import tempfile +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + import pytest import requests +from urllib3.util.url import parse_url import gambit.http as http @@ -12,8 +23,23 @@ 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", + "no_proxy", + "NO_PROXY", + "REQUESTS_CA_BUNDLE", + "CURL_CA_BUNDLE", + ): + monkeypatch.delenv(name, raising=False) yield http.clear_scheme_cache() @@ -90,3 +116,883 @@ def test_options_uses_shared_request(monkeypatch): http, "request", lambda method, url, **kw: (method, url) ) assert http.options_for_host("example.com") == ("OPTIONS", "https://example.com") + + +def test_ipv6_literal_is_bracketed_in_the_url(monkeypatch): + monkeypatch.setattr(http, "request", lambda method, url, **kw: FakeResponse(url)) + response = http.get_for_host("2001:db8::1", "/robots.txt") + assert response.url == "https://[2001:db8::1]/robots.txt" + response = http.get_for_host("[::1]") + assert response.url == "https://[::1]" + + +def test_cross_host_redirect_is_not_followed(monkeypatch): + calls = [] + + def fake_request(method, url, headers=None, timeout=None, **kwargs): + calls.append(url) + + class Response: + def __init__(self, url): + self.url = url + self.status_code = 302 + self.headers = {"Location": "http://169.254.169.254/latest/meta-data/"} + + def close(self): + pass + + return Response(url) + + monkeypatch.setattr(http.requests, "request", fake_request) + response = http.get("http://example.com/") + assert calls == ["http://example.com/"] + assert response.status_code == 302 + assert response.gambit_redirects == 0 + + +def test_same_host_redirect_followed_when_address_is_public(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) + + class Response: + def __init__(self, url): + self.url = url + self.status_code = 200 if url.endswith("/next") else 302 + self.headers = {} if self.status_code == 200 else {"Location": "http://example.com/next"} + + def close(self): + pass + + def iter_content(self, chunk_size=8192): + yield b"ok" + + return Response(url) + + monkeypatch.setattr(http.requests, "request", fake_request) + response = http.get("http://example.com/") + assert calls == ["http://example.com/", "http://example.com/next"] + assert response.gambit_redirects == 1 + assert response.gambit_final_url == "http://example.com/next" + note = http.annotate_findings( + [type("F", (), {"evidence": "Server: nginx"})()], + response, + ) + assert "final URL: http://example.com/next" in note[0].evidence + + +def test_same_host_redirect_to_a_private_address_is_not_followed(monkeypatch): + """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): + calls.append(url) + + class Response: + def __init__(self, url): + self.url = url + self.status_code = 302 + self.headers = {"Location": "http://example.com/admin"} + + def close(self): + pass + + return Response(url) + + monkeypatch.setattr(http.requests, "request", fake_request) + response = http.get("http://example.com/") + assert calls == ["http://example.com/"] + assert response.gambit_redirects == 0 + + +def test_redirect_chain_is_capped(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) + + class Response: + def __init__(self, url): + self.url = url + self.status_code = 302 + hop = len(calls) + self.headers = {"Location": f"http://example.com/r{hop}"} + + def close(self): + pass + + return Response(url) + + monkeypatch.setattr(http.requests, "request", fake_request) + http.get("http://example.com/start") + assert len(calls) == http.MAX_REDIRECTS + 1 + + +def test_response_body_is_capped(monkeypatch): + class Response: + status_code = 200 + headers = {} + url = "https://example.com/" + + def iter_content(self, chunk_size=8192): + yield b"A" * 100 + yield b"B" * 100 + + def close(self): + pass + + 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() + + +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) + 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_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://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://target.htb:{port}/r", timeout=2) + finally: + _close_server(server) + 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): + 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_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: + 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) + 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: + 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: + server.shutdown() + server.server_close() diff --git a/tests/test_net.py b/tests/test_net.py new file mode 100644 index 0000000..b9e3b7c --- /dev/null +++ b/tests/test_net.py @@ -0,0 +1,128 @@ +import socket +import time + +import pytest + +from gambit.net import connect, is_public_ip, resolve_ips + + +def test_public_address_check_rejects_metadata_and_private_ranges(): + assert is_public_ip("93.184.216.34") + assert is_public_ip("2606:4700:4700::1111") + assert not is_public_ip("169.254.169.254") + assert not is_public_ip("127.0.0.1") + assert not is_public_ip("10.1.1.1") + assert not is_public_ip("100.64.0.1") + assert not is_public_ip("not-an-ip") + + +def test_connect_resolves_with_a_deadline_then_dials_the_ip(monkeypatch): + seen = {} + + 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 + seen["timeout"] = timeout + + class Sock: + def close(self): + pass + + return Sock() + + 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 + # 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_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( + socket, "create_connection", lambda *args, **kwargs: called.append(args) + ) + with pytest.raises(OSError): + connect("example.com", 99999, 0.2) + assert called == [] diff --git a/tests/test_packaging.py b/tests/test_packaging.py new file mode 100644 index 0000000..404f155 --- /dev/null +++ b/tests/test_packaging.py @@ -0,0 +1,13 @@ +from pathlib import Path + + +def test_runtime_dependencies_are_pinned_and_ci_audits_them(): + root = Path(__file__).resolve().parents[1] + project = (root / "pyproject.toml").read_text(encoding="utf-8") + requirements = (root / "requirements.txt").read_text(encoding="utf-8") + workflow = (root / ".github" / "workflows" / "ci.yml").read_text(encoding="utf-8") + for pin in ("dnspython==2.8.0", "requests==2.34.2", "urllib3==2.8.0"): + assert pin in project + assert pin in requirements + assert "pip_audit" in workflow or "pip-audit" in workflow + assert "requirements.txt" in workflow diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index a764a26..da35c10 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -166,6 +166,89 @@ def test_split_host_port(): assert split_host_port("example.com:notaport") == ("example.com:notaport", None) +def test_split_host_port_ipv6_and_out_of_range_ports(): + from gambit.pipeline import normalize_host, port_out_of_range, split_host_port + + assert normalize_host("2001:db8::1") == "[2001:db8::1]" + assert normalize_host("[::1]") == "[::1]" + assert normalize_host("HTTPS://[2001:db8::1]:8443/path") == "[2001:db8::1]:8443" + assert dedupe_hosts(["::1", "[::1]"]) == ["[::1]"] + + assert split_host_port("[::1]") == ("::1", None) + assert split_host_port("2001:db8::1") == ("2001:db8::1", None) + assert split_host_port("[2001:db8::1]:8443") == ("2001:db8::1", 8443) + assert split_host_port("https://[2001:db8::1]:8443/x") == ("2001:db8::1", 8443) + + # 99999 must not survive as a port: getaddrinfo would wrap it to 34463. + assert port_out_of_range("example.com:99999") + assert split_host_port("example.com:99999") == ("example.com", None) + assert split_host_port("example.com:65536") == ("example.com", None) + assert split_host_port("example.com:0") == ("example.com", None) + assert split_host_port("example.com:65535") == ("example.com", 65535) + assert split_host_port("[2001:db8::1]:99999") == ("2001:db8::1", None) + + +def test_out_of_range_port_does_not_connect(monkeypatch): + from gambit.pipeline import host_reachable + + called = [] + + def fake_connect(*args, **kwargs): + called.append(args) + raise OSError("should not be called") + + monkeypatch.setattr("gambit.net.connect", fake_connect) + assert host_reachable("example.com:99999", timeout=0.1) is False + assert called == [] + + +def test_preflight_open_alt_port_is_used_in_the_url(monkeypatch): + import gambit.http as http + from gambit.pipeline import host_reachable + + http.clear_scheme_cache() + + def fake_connect(hostname, port, timeout): + if port not in (8443, 8080): + raise OSError("closed") + + class Sock: + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + # 8443 is tried before 8080, so this host looks HTTPS-on-8443. + if port == 8443 and hostname == "alt.example": + return Sock() + if port == 8080 and hostname == "dev.example": + return Sock() + raise OSError("closed") + + monkeypatch.setattr("gambit.net.connect", fake_connect) + assert host_reachable("alt.example", timeout=0.2) is True + assert host_reachable("dev.example", timeout=0.2) is True + + seen = [] + + def fake_request(method, url, **kwargs): + seen.append(url) + + class Response: + def __init__(self, url): + self.url = url + + return Response(url) + + monkeypatch.setattr(http, "request", fake_request) + http.get_for_host("alt.example") + http.get_for_host("dev.example") + assert seen[0] == "https://alt.example:8443" + assert seen[1] == "http://dev.example:8080" + http.clear_scheme_cache() + + def test_run_pipeline_preflight_skips_dead_hosts(): findings = run_pipeline( "example.com", diff --git a/tests/test_ports.py b/tests/test_ports.py index 3169692..fbd3f86 100644 --- a/tests/test_ports.py +++ b/tests/test_ports.py @@ -3,8 +3,10 @@ ALT_WEB_PORTS, DATABASE_PORTS, PORTS, + TIMEOUT, WEB_PORTS, evaluate, + run, service_for_port, severity_for_port, ) @@ -44,6 +46,43 @@ def test_service_for_port(): assert service_for_port(12345) == "unknown" +def test_explicit_port_is_probed_and_out_of_range_port_is_not(monkeypatch): + tried = [] + + 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, ["203.0.113.1"]) in tried + tried.clear() + assert run("example.com:99999") == [] + 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(): finding = evaluate("example.com", 6379) assert finding.severity == Severity.MED diff --git a/tests/test_takeover.py b/tests/test_takeover.py index ec02323..61190fd 100644 --- a/tests/test_takeover.py +++ b/tests/test_takeover.py @@ -99,6 +99,59 @@ def test_evaluate_unknown_resolution_without_fingerprint_is_info(): assert finding.severity == Severity.INFO +def test_fingerprint_on_http_200_is_not_high_and_404_still_is(): + page = "There isn't a GitHub Pages site here." + live = evaluate( + "sub.example.com", + "foo.github.io", + resolves=True, + body=page, + status_code=200, + fingerprints=SERVICES, + ) + assert live.severity == Severity.INFO + missing = evaluate( + "sub.example.com", + "foo.github.io", + resolves=True, + body=page, + status_code=404, + fingerprints=SERVICES, + ) + assert missing.severity == Severity.HIGH + + +def test_unclaimable_nxdomain_and_generic_iis_text_are_not_high(): + from gambit.modules.takeover import load_fingerprints + + bundled = load_fingerprints() + fastly = evaluate( + "a.example.com", "x.fastly.net", resolves=False, fingerprints=bundled + ) + assert fastly.severity == Severity.INFO + github = evaluate( + "a.example.com", "x.github.io", resolves=False, fingerprints=bundled + ) + assert github.severity == Severity.HIGH + + azure = next(item for item in bundled if item.service == "Microsoft Azure") + generic = ( + "The resource you are looking for has been removed, had its name " + "changed, or is temporarily unavailable." + ) + assert find_unclaimed_fingerprint(generic, azure) is None + gitlab = next(item for item in bundled if item.service == "GitLab Pages") + ordinary = evaluate( + "a.example.com", + "x.gitlab.io", + resolves=True, + body="The page you're looking for could not be found.", + status_code=404, + fingerprints=[gitlab], + ) + assert ordinary.severity == Severity.INFO + + def test_evaluate_evidence_never_claims_resource_is_unclaimed(): finding = evaluate("sub.example.com", "foo.github.io", resolves=False, fingerprints=SERVICES) # Title says "Possible"; evidence/why must stay hedged, not definitive. diff --git a/tests/test_tls.py b/tests/test_tls.py index 38f292e..b8708a1 100644 --- a/tests/test_tls.py +++ b/tests/test_tls.py @@ -1,7 +1,35 @@ +import ssl from datetime import datetime, timezone from gambit.models import Severity -from gambit.modules.tls import days_until_expiry, evaluate, parse_cert_datetime +from gambit.modules.tls import ( + certificate_from_socket, + days_until_expiry, + evaluate, + parse_cert_datetime, +) + +# Self-signed fixture, not valid after 2020-06-01. Generated offline so the +# suite does not shell out to openssl (CI includes Windows). +EXPIRED_PEM = """\ +-----BEGIN CERTIFICATE----- +MIICrTCCAZWgAwIBAgIBATANBgkqhkiG9w0BAQsFADAaMRgwFgYDVQQDDA9leHBp +cmVkLmV4YW1wbGUwHhcNMjAwMTAxMDAwMDAwWhcNMjAwNjAxMDAwMDAwWjAaMRgw +FgYDVQQDDA9leHBpcmVkLmV4YW1wbGUwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAw +ggEKAoIBAQDCOSGTct1N3/vVYqvU1fi/JC2nNgxArQUgbFnMyzv6U5WmWBCYDYq/ +YKGBUoJegqaRTMHH4dNizxhtIquhvcqW9mu21gnNE4GIn9ND/n0ixuDx5bUdZKLc +Lw10tOzuT6+t/LxlLTZJP7CV6pC+4eeV0nIOYHpeITUcfA2RhG284jqVPGNWXIbP +tKYJXFQBVrZhIYZGq4vYaiIIepFighzaUKBBJ6XCFm3eBUTDkMA+25LmhnpoHgr8 +duPlBUWnhaMN0z3drYlT7AFag7H18y6w/VcHHGzYp1/KTtl9kB8k4NAaOsqkccYo +B8L2f2ADnwJf7JB8X/t8CuI0IP4nwYVtAgMBAAEwDQYJKoZIhvcNAQELBQADggEB +AFE6U80jH5RXlyK+yUeR+gc58RJ+N0rhweU3AxB3OChEoPYkqnLWaV4Lu8wEX5eE +MusQ2yK4GusXTJ9tdn83eTVfrieOAz2jhpcnnlHfK8theu9q0XMSKCVNX/5XXTDd +6z4cXK/w9fhMX0S1mZqKCaub9+Qs11tOrCnZUub2w1o6ttJPhjtI00pQ+vbPCHsE +vDAo78pnJGhg64snxvyfxYoiOgboBjRDOptr7JGvp9W27XtI1nWNBFyoqlAqhB4u +qfyo+FW4cCbvIjr3hxkJGQEoDRmxYC9QgA7HtkofWNE/Tj7Okj2sJ3Lt9LzpC0oN +7lLtSJvVIZ5OQq4Pvy3AR/A= +-----END CERTIFICATE----- +""" NOW = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc) @@ -76,6 +104,58 @@ def test_no_certificate_info_is_info_not_high(): assert evaluate("example.com", None, "TLSv1.3", now=NOW)[0].severity == Severity.INFO +def test_getpeercert_empty_dict_is_recovered_from_der(): + der = ssl.PEM_cert_to_DER_cert(EXPIRED_PEM) + + class Sock: + def getpeercert(self, binary_form=False): + if binary_form: + return der + return {} + + cert = certificate_from_socket(Sock()) + assert cert + assert cert.get("notAfter") + findings = evaluate( + "expired.example", + cert, + "TLSv1.2", + now=datetime(2026, 1, 1, tzinfo=timezone.utc), + ) + assert any(f.severity == Severity.HIGH and "expired" in f.title.lower() for f in findings) + + +def test_handshake_reads_binary_cert_when_getpeercert_returns_empty(monkeypatch): + der = ssl.PEM_cert_to_DER_cert(EXPIRED_PEM) + + class Raw: + def close(self): + pass + + class TLS: + def getpeercert(self, binary_form=False): + if binary_form: + return der + return {} + + def version(self): + return "TLSv1.2" + + def close(self): + pass + + class Context: + def wrap_socket(self, sock, server_hostname=None): + return TLS() + + monkeypatch.setattr("gambit.net.connect", lambda *args, **kwargs: Raw()) + from gambit.modules.tls import _handshake + + cert, protocol = _handshake("expired.example", 443, Context()) + assert protocol == "TLSv1.2" + assert "2020" in cert["notAfter"] + + def test_hostname_mismatch_and_self_signed_are_info(): findings = evaluate( "example.com", diff --git a/tests/test_triage.py b/tests/test_triage.py index 999ade2..e05936b 100644 --- a/tests/test_triage.py +++ b/tests/test_triage.py @@ -50,3 +50,25 @@ def test_to_markdown_lists_most_severe_first_with_headings_and_bullets(): def test_to_markdown_is_table_free(): text = to_markdown([make(Severity.MED, "thing")]) assert "|" not in text + + +def test_text_and_markdown_strip_injected_markup_json_keeps_it(): + forged = ( + "ok\n\n## [HIGH] example.com: forged\n\n" + "- **Evidence:** pwned " + ) + finding = Finding(Severity.INFO, "example.com", "title", forged, "why", "next") + text = to_text([finding]) + # The payload stays inside the evidence line; it cannot open a new finding. + assert not any(line.startswith("## ") or line.startswith("[HIGH]") for line in text.splitlines()) + evidence_lines = [line for line in text.splitlines() if line.strip().startswith("Evidence:")] + assert len(evidence_lines) == 1 + assert "\n## [HIGH]" not in text + + markdown = to_markdown([finding]) + assert "\n## [HIGH] example.com: forged" not in markdown + assert "