From 954e8119c53f36617f6173852368c8e1a9cb495b Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 21 Oct 2026 06:32:54 +0000 Subject: [PATCH 1/3] Fix high and medium scan safety issues Stop cross-host and private redirects, cap response bodies, recover unverified TLS certificates, and parse IPv6 and explicit ports before DNS or socket connect. Pin runtime dependencies and audit them in CI. Co-authored-by: Kamal Choubey --- .github/workflows/ci.yml | 5 + gambit/data/takeover_fingerprints.json | 62 ++++-- gambit/http.py | 259 ++++++++++++++++++++++++- gambit/modules/baseline_headers.py | 4 +- gambit/modules/cors.py | 4 +- gambit/modules/fingerprint.py | 4 +- gambit/modules/gather.py | 25 ++- gambit/modules/http_methods.py | 4 +- gambit/modules/ports.py | 20 +- gambit/modules/takeover.py | 71 +++++-- gambit/modules/tls.py | 76 +++++++- gambit/net.py | 88 +++++++++ gambit/pipeline.py | 162 ++++++++++++++-- gambit/triage.py | 41 +++- pyproject.toml | 5 +- requirements.txt | 1 + tests/test_fingerprint.py | 18 ++ tests/test_gather.py | 33 ++++ tests/test_http.py | 131 +++++++++++++ tests/test_net.py | 59 ++++++ tests/test_packaging.py | 13 ++ tests/test_pipeline.py | 83 ++++++++ tests/test_ports.py | 17 ++ tests/test_takeover.py | 53 +++++ tests/test_tls.py | 82 +++++++- tests/test_triage.py | 22 +++ 26 files changed, 1244 insertions(+), 98 deletions(-) create mode 100644 gambit/net.py create mode 100644 tests/test_net.py create mode 100644 tests/test_packaging.py 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..ca8e2c2 100644 --- a/gambit/http.py +++ b/gambit/http.py @@ -11,30 +11,92 @@ 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 when the next URL stays on the same host and +every resolved address is globally routable, and only up to +:data:`MAX_REDIRECTS` hops. Response bodies are streamed and capped at +:data:`MAX_BODY` bytes unless the caller is reading the stream itself. """ +from urllib.parse import urljoin, urlsplit + import requests +from gambit.net import is_public_ip, resolve_ips + 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 + whose resolved addresses are all public, and at most + :data:`MAX_REDIRECTS` of them. A cross-host or private redirect is + returned as the redirect response and is not fetched. + * ``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 + + current = url + current_method = method + hops = 0 + while True: + response = requests.request( + current_method, current, headers=merged, timeout=timeout, **kwargs + ) + _stamp_redirect(response, hops, current) + if not allow_redirects or hops >= MAX_REDIRECTS: + break + status = getattr(response, "status_code", 0) or 0 + location = _location_header(response) + if status not in (301, 302, 303, 307, 308) or not location: + break + nxt = urljoin(current, location.strip()) + if not _redirect_is_safe(current, nxt, timeout): + break + _close_quietly(response) + if status == 303: + current_method = "GET" + current = nxt + hops += 1 + if not caller_streams: + _buffer_limited(response, max_bytes) + return response def get(url: str, *, timeout: float = TIMEOUT, headers: dict | None = None, **kwargs): @@ -52,15 +114,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,6 +150,31 @@ 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. @@ -77,16 +182,20 @@ def _request_for_host(method: str, target: str, path: str = "", **kwargs): trigger the fallback; an HTTP error status is a real answer and is returned as-is. 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: 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 +209,151 @@ 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 _redirect_is_safe(current: str, nxt: str, timeout) -> bool: + """Same host, http(s) only, and every resolved address globally routable.""" + new = urlsplit(nxt) + old = urlsplit(current) + if new.scheme not in ("http", "https") or not new.hostname: + return False + current_host = (old.hostname or "").lower().rstrip(".") + next_host = new.hostname.lower().rstrip(".") + if not current_host or current_host != next_host: + return False + try: + ips = resolve_ips(next_host, _timeout_seconds(timeout)) + except Exception: + return False + return bool(ips) and all(is_public_ip(ip) for ip in ips) + + +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..c588d80 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,15 +105,25 @@ 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`. DNS for each connect is bounded by + :data:`TIMEOUT`. A port outside 1–65535 is not probed: the platform + resolver would otherwise truncate it into a different port. + """ + from gambit.net import connect 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) findings = [] - for port in PORTS: + for port in ports: try: - with socket.create_connection((host, port), timeout=TIMEOUT): + with connect(host, port, TIMEOUT): pass # connected == open; send nothing, read nothing except OSError: continue # refused, filtered, timed out, or unresolvable 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..81a3ac3 --- /dev/null +++ b/gambit/net.py @@ -0,0 +1,88 @@ +"""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. +Callers here resolve with dnspython's ``lifetime`` first, then connect to +the resulting address. +""" + +import ipaddress +import socket + +import dns.exception +import dns.resolver + + +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 DNS query. DNS + failures and an empty answer both yield ``[]``; they never raise. + """ + if is_ip_address(hostname): + return [hostname] + found: list[str] = [] + seen: set[str] = set() + for rdtype in ("A", "AAAA"): + try: + answers = dns.resolver.resolve(hostname, rdtype, lifetime=timeout) + except dns.exception.DNSException: + continue + for rdata in answers: + ip = getattr(rdata, "address", None) or str(rdata).strip() + if ip and ip not in seen: + seen.add(ip) + found.append(ip) + return found + + +def connect(hostname: str, port: int, timeout: float): + """TCP-connect to ``hostname``:``port`` with DNS bounded by ``timeout``. + + The socket is connected to a resolved address, never to a name, so + ``getaddrinfo`` is only asked to parse an IP literal. Ports outside + 1–65535 raise ``OSError`` instead of being truncated by the resolver. + Raises ``OSError`` when the name does not resolve or every address + refuses the connection. + """ + number = int(port) + if number < 1 or number > 65535: + raise OSError(f"port out of range: {port}") + ips = 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..c7f8ef7 100644 --- a/tests/test_http.py +++ b/tests/test_http.py @@ -90,3 +90,134 @@ 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): + monkeypatch.setattr(http, "resolve_ips", lambda host, timeout: ["127.0.0.1"]) + 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) diff --git a/tests/test_net.py b/tests/test_net.py new file mode 100644 index 0000000..cc49e22 --- /dev/null +++ b/tests/test_net.py @@ -0,0 +1,59 @@ +import socket + +import dns.resolver +import pytest + +from gambit.net import connect, is_public_ip + + +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_resolve(name, rdtype, lifetime=None): + seen.setdefault("queries", []).append((name, rdtype, lifetime)) + if rdtype == "AAAA": + raise dns.resolver.NoAnswer() + + class Answer: + address = "93.184.216.34" + + return [Answer()] + + def fake_create_connection(address, timeout=None): + seen["address"] = address + seen["timeout"] = timeout + + class Sock: + def close(self): + pass + + return Sock() + + monkeypatch.setattr(dns.resolver, "resolve", fake_resolve) + monkeypatch.setattr(socket, "create_connection", fake_create_connection) + sock = connect("example.com", 443, 1.25) + assert sock is not None + assert seen["address"] == ("93.184.216.34", 443) + assert seen["timeout"] == 1.25 + assert ("example.com", "A", 1.25) in seen["queries"] + assert all(item[2] == 1.25 for item in seen["queries"]) + + +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..af38a0f 100644 --- a/tests/test_ports.py +++ b/tests/test_ports.py @@ -5,6 +5,7 @@ PORTS, WEB_PORTS, evaluate, + run, service_for_port, severity_for_port, ) @@ -44,6 +45,22 @@ 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): + tried.append((hostname, port)) + raise OSError("closed") + + monkeypatch.setattr("gambit.net.connect", fake_connect) + assert run("example.com:9090") == [] + assert ("example.com", 9090) in tried + tried.clear() + assert run("example.com:99999") == [] + assert all(port != 99999 and port != 34463 for _host, port in tried) + assert ("example.com", 443) in tried + + 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 "