diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8050cb1..2a6c61d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -12,7 +12,7 @@ jobs: fail-fast: false matrix: os: [ubuntu-latest, windows-latest] - python-version: ["3.11", "3.12"] + python-version: ["3.11", "3.12", "3.13"] steps: - name: Check out repository diff --git a/README.md b/README.md index fa0c54f..bb66ea5 100644 --- a/README.md +++ b/README.md @@ -52,6 +52,8 @@ python -m gambit example.com GAMBIT first runs Gather for the target, then runs the per-host checks (takeover, Analyze, Baseline, Inspect, and Map) on the apex host and the subdomains it discovers, and finally ranks everything with Triage. Discovered hosts are de-duplicated and capped (25 by default; see `--max-hosts`) so a scan stays small. Each host first gets a cheap connection pre-flight, so a dead subdomain costs a second instead of stacking every check's timeout; hosts are then checked one at a time with a short pause between them. Web checks try HTTPS first and fall back to plain HTTP, so HTTP-only sites are still checked. +Certificate verification stays on unless you pass `--insecure`. A certificate problem (self-signed, untrusted CA, hostname mismatch) is reported and is **not** retried over plain HTTP, so a scan cannot silently downgrade. `--insecure` accepts those certificates and keeps speaking HTTPS. GAMBIT prints a warning and records the choice in the report. Use it for lab targets and appliances that ship a private certificate, and leave it off when you want the client's trust store to apply. + | Flag | Description | |---|---| | `--format text\|md\|json` | Report format: human-readable text (default), Markdown (`md`), or JSON (`json`). | @@ -59,6 +61,7 @@ GAMBIT first runs Gather for the target, then runs the per-host checks (takeover | `-o FILE`, `--output FILE` | Write the report to `FILE` instead of standard output. | | `--max-hosts N` | Maximum number of hosts to check, including the target itself (default: 25). | | `--yes` | Skip the interactive authorization confirmation (see [Responsible use](#responsible-use)). Use only where you already have permission. | +| `--insecure` | Disable TLS certificate verification for HTTPS requests and the TLS handshake. Self-signed certificates and hostname mismatches are accepted. Off by default. Does not fall back to plain HTTP. The report notes that verification was disabled. | For example, to save a Markdown report for at most 10 hosts without the confirmation prompt: @@ -146,7 +149,7 @@ Adding a check means adding one file. The Triage stage sorts every finding from - [x] **Map:** common-port check - [x] **Triage:** severity ranking and Markdown/JSON export - [x] Tests for every module -- [x] CI (pytest on Ubuntu and Windows, Python 3.11 and 3.12) +- [x] CI (pytest on Ubuntu and Windows, Python 3.11, 3.12, and 3.13) - [ ] First tagged release ## Prior art diff --git a/gambit/__main__.py b/gambit/__main__.py index b58a6f4..f831ea5 100644 --- a/gambit/__main__.py +++ b/gambit/__main__.py @@ -11,6 +11,8 @@ import sys from pathlib import Path +from gambit.http import set_tls_verification +from gambit.models import Finding, Severity from gambit.modules import ( baseline_headers, cors, @@ -22,6 +24,7 @@ takeover, tls, ) +from gambit.net import clear_resolution_cache from gambit.pipeline import ( DEFAULT_MAX_HOSTS, discover_hosts, @@ -31,6 +34,12 @@ ) from gambit.triage import to_json, to_markdown, to_text +INSECURE_NOTE = ( + "NOTE: TLS certificate verification was disabled (--insecure). " + "HTTPS requests accepted untrusted certificates and were not retried " + "over plain HTTP." +) + # Run once on the apex target, in this order. APEX_MODULES = [gather] @@ -77,16 +86,47 @@ def _positive_int(value: str) -> int: return number -def format_report(findings, report_format: str) -> str: - """Render ``findings`` as ``text``, ``md``, or ``json`` (ranked).""" +def verification_disabled_finding(host: str) -> Finding: + """INFO finding recorded when the operator passed ``--insecure``.""" + return Finding( + severity=Severity.INFO, + host=host, + title="TLS certificate verification disabled", + evidence=( + "--insecure: HTTPS fetches and the TLS handshake did not verify " + "certificates. A certificate failure was not retried over plain HTTP." + ), + why=( + "Self-signed and untrusted certificates are common on pentest " + "targets. With verification off, those sites can still be checked, " + "but a network attacker could present a different certificate." + ), + next_step="Re-run without --insecure when the target uses a certificate you trust.", + ) + + +def format_report(findings, report_format: str, *, insecure: bool = False) -> str: + """Render ``findings`` as ``text``, ``md``, or ``json`` (ranked). + + ``insecure=True`` puts a one-line note at the top of text and Markdown + reports. JSON keeps a pure list; the matching INFO finding carries + the same fact there. + """ try: formatter = _FORMATTERS[report_format] except KeyError: raise ValueError(f"unknown report format: {report_format!r}") from None - return formatter(findings) + body = formatter(findings) + if not insecure or report_format == "json": + return body + if report_format == "md": + return f"> {INSECURE_NOTE}\n\n{body}" + return f"{INSECURE_NOTE}\n\n{body}" -def collect_findings(target: str, max_hosts: int = DEFAULT_MAX_HOSTS) -> list: +def collect_findings( + target: str, max_hosts: int = DEFAULT_MAX_HOSTS, *, insecure: bool = False +) -> list: """Run the full pipeline for ``target`` and return all findings. Gather runs once on the apex; host discovery uses Gather's crt.sh @@ -95,8 +135,16 @@ def collect_findings(target: str, max_hosts: int = DEFAULT_MAX_HOSTS) -> list: then run against the discovered (deduped, capped) hosts that pass a cheap liveness pre-flight. Per-module failures are isolated by the pipeline helpers, so one failing check never aborts the scan. + + ``insecure=True`` disables TLS certificate verification for the scan + (HTTPS is still used) and records that choice as an INFO finding. + DNS answers cached by an earlier scan are dropped first. """ + set_tls_verification(not insecure) + clear_resolution_cache() findings = [] + if insecure: + findings.append(verification_disabled_finding(target)) findings.extend(run_for_host(target, APEX_MODULES)) hosts = discover_hosts( @@ -159,6 +207,16 @@ def build_parser() -> argparse.ArgumentParser: action="store_true", help="confirm you own the target or have written permission to test it", ) + parser.add_argument( + "--insecure", + action="store_true", + help=( + "disable TLS certificate verification for HTTPS requests and the " + "TLS handshake (self-signed certificates and hostname mismatches " + "are accepted). Off by default. Does not fall back to plain HTTP. " + "The report notes that verification was disabled." + ), + ) return parser @@ -174,10 +232,15 @@ def main(argv=None) -> int: print("Aborting: authorization not confirmed.", file=sys.stderr) return 1 - findings = collect_findings(args.target, max_hosts=args.max_hosts) + if args.insecure: + print(INSECURE_NOTE, file=sys.stderr) + + findings = collect_findings( + args.target, max_hosts=args.max_hosts, insecure=args.insecure + ) report_format = "json" if args.json else args.format - report = format_report(findings, report_format) + report = format_report(findings, report_format, insecure=args.insecure) if args.output: try: diff --git a/gambit/http.py b/gambit/http.py index b9699b4..b6b7660 100644 --- a/gambit/http.py +++ b/gambit/http.py @@ -37,9 +37,14 @@ 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. +failure is not retried over plain HTTP. :func:`set_tls_verification` +can turn verification off for a scan (the ``--insecure`` flag). That +still speaks HTTPS; it does not downgrade a certificate failure to +plain HTTP. + +DNS lookups are bounded, cached for the scan, and capped at a few +in-flight resolver threads. One that outruns its deadline does not +block the scan; see :mod:`gambit.net`. Response bodies are streamed and capped at :data:`MAX_BODY` bytes unless the caller is reading the stream itself. @@ -49,15 +54,17 @@ import os import socket import ssl +import warnings from typing import NamedTuple from urllib.parse import urljoin import requests import urllib3.connection import urllib3.exceptions +from urllib3.exceptions import InsecureRequestWarning from urllib3.util.url import parse_url -from gambit.net import is_ip_address, is_public_ip, resolve_ips +from gambit.net import is_ip_address, is_public_ip, resolution_saturated, resolve_ips # Identity of the real requests entry point. Tests replace # ``gambit.http.requests.request``; when they have, pinning is skipped @@ -78,6 +85,43 @@ _SCHEME_BY_HOST: dict[str, str] = {} # host -> non-default TCP port discovered by the liveness pre-flight _PORT_BY_HOST: dict[str, int] = {} +# Safer default: verify certificates unless the operator opts out. +_tls_verification_enabled = True +# The warnings filter installed while verification is off, if any. +_active_warning_filter: tuple | None = None + + +def tls_verification_enabled() -> bool: + """True when HTTPS fetches still check certificates.""" + return _tls_verification_enabled + + +def set_tls_verification(enabled: bool) -> None: + """Turn TLS certificate verification on or off for later fetches. + + Off is the ``--insecure`` opt-in. Hostname and chain checks are + skipped (``verify=False``) when a caller did not pass ``verify`` + itself. A certificate failure is still not retried over plain HTTP. + """ + global _tls_verification_enabled, _active_warning_filter + enabled = bool(enabled) + _tls_verification_enabled = enabled + # requests warns once per unverified call. The CLI surfaces one note + # instead. Installing the filter must be idempotent: simplefilter + # inserts a new entry on every call. + if enabled: + if _active_warning_filter is not None: + try: + warnings.filters.remove(_active_warning_filter) + except ValueError: + pass + _active_warning_filter = None + return + if _active_warning_filter is not None: + return + item = ("ignore", None, InsecureRequestWarning, None, 0) + warnings.filters.insert(0, item) + _active_warning_filter = item def request( @@ -157,6 +201,11 @@ def request( if pending_ips is None: ips = resolve_ips(_dns_name(target.host), _timeout_seconds(timeout)) if not ips: + if resolution_saturated(target.host): + raise requests.ConnectionError( + "DNS resolver saturated; could not resolve " + f"{target.host}" + ) raise requests.ConnectionError( f"could not resolve {target.host}" ) @@ -179,7 +228,7 @@ def request( current, headers=merged, timeout=timeout, - **kwargs, + **_apply_tls_verification(kwargs), ) send_url = current except requests.RequestException: @@ -318,7 +367,10 @@ def _request_for_host(method: str, target: str, path: str = "", **kwargs): 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. + With verification disabled (:func:`set_tls_verification`), those + certificates are accepted on HTTPS instead, and plain HTTP is still + not used as a substitute. Raises the last exception if every scheme + fails. """ if path and not path.startswith("/"): path = "/" + path @@ -503,15 +555,30 @@ def _is_tls_verify_failure(exc: BaseException) -> bool: return False +def _apply_tls_verification(kwargs: dict) -> dict: + """Copy ``kwargs`` and apply the process verification policy. + + An explicit ``verify`` argument always wins, including ``verify=False`` + and a bundle path. When the caller left it unset and verification is + disabled, certificate checks are turned off. + """ + copied = dict(kwargs) + if "verify" not in copied and not _tls_verification_enabled: + copied["verify"] = False + return copied + + 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. + ``verify=False`` and an explicit bundle path are left alone. With + verification disabled and no explicit ``verify``, the result is + ``verify=False`` and the environment bundle is not consulted. """ - copied = dict(kwargs) + copied = _apply_tls_verification(kwargs) copied.pop("proxies", None) if "verify" not in copied: verify = True @@ -867,5 +934,7 @@ def _buffer_limited(response, limit: int) -> None: "request", "set_preferred_port", "set_preferred_scheme", + "set_tls_verification", + "tls_verification_enabled", "url_authority", ] diff --git a/gambit/modules/gather.py b/gambit/modules/gather.py index 1807f4a..51b7088 100644 --- a/gambit/modules/gather.py +++ b/gambit/modules/gather.py @@ -239,6 +239,9 @@ def _fetch_crtsh_subdomains(host: str) -> list[str]: _CRT_SH_URL.format(host=host), timeout=TIMEOUT, allow_redirects=False, + # crt.sh is a third party. --insecure must not turn off + # verification for this call. + verify=True, ) response.raise_for_status() names = parse_crtsh(response.json()) diff --git a/gambit/modules/ports.py b/gambit/modules/ports.py index 964622f..8382c8a 100644 --- a/gambit/modules/ports.py +++ b/gambit/modules/ports.py @@ -114,8 +114,8 @@ def run(target: str) -> list[Finding]: 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 + from gambit.net import connect, resolution_saturated, resolve_ips + from gambit.pipeline import resolver_saturated_finding, split_host_port hostname, explicit_port = split_host_port(target) host = hostname or target @@ -124,6 +124,8 @@ def run(target: str) -> list[Finding]: ports.append(explicit_port) ips = resolve_ips(host, TIMEOUT) if not ips: + if resolution_saturated(host): + return [resolver_saturated_finding(host)] return [] findings = [] for port in ports: diff --git a/gambit/modules/tls.py b/gambit/modules/tls.py index f065b1c..b46da01 100644 --- a/gambit/modules/tls.py +++ b/gambit/modules/tls.py @@ -3,10 +3,12 @@ Standard library only (``ssl`` + ``socket``). The pure :func:`evaluate` function does the date/version maths on a certificate dict like the one ``ssl.SSLSocket.getpeercert()`` returns, so tests never need a network. -``run`` performs a single verified handshake; a certificate that fails -verification (hostname mismatch, self-signed, ...) is reported as INFO -evidence rather than a vulnerability claim, and its details are still -collected over an unverified handshake for the expiry/version checks. +``run`` performs a verified handshake first so a failure can be +classified (hostname mismatch, self-signed, or an untrusted chain) and +reported as INFO rather than a vulnerability claim. Certificate details +are then read over an unverified handshake for the expiry and version +checks. ``--insecure`` skips verification on content fetches only; this +module still classifies the certificate the same way. """ import os @@ -378,7 +380,15 @@ def _unverified_context() -> ssl.SSLContext: def run(target: str) -> list[Finding]: - """Check the target's TLS certificate and protocol over one handshake.""" + """Check the target's TLS certificate and protocol. + + The first handshake verifies, including when ``--insecure`` disabled + verification for content fetches. That failure is what classifies a + hostname mismatch, a self-signed certificate, or an untrusted chain. + A second, unverified handshake then supplies expiry and protocol + details. The scan-level note still records that content fetches did + not verify. + """ hostname, port = _split_host_port(target) try: cert, protocol = _handshake(hostname, port, ssl.create_default_context()) diff --git a/gambit/net.py b/gambit/net.py index c97b3e5..6116de1 100644 --- a/gambit/net.py +++ b/gambit/net.py @@ -6,15 +6,47 @@ 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. +A lookup that outruns the deadline is abandoned. Python cannot cancel +``getaddrinfo``, so the thread keeps running until libc returns, but the +scan does not wait for it. At most :data:`MAX_RESOLVER_THREADS` of those +threads exist at once. A further lookup waits for a free slot only until +its own deadline, then returns ``[]`` without caching that skip. +:func:`resolution_saturated` is true for that host until a later call +actually runs, and a warning is logged, so a full pool is not reported +as a name that does not resolve. Only a non-empty answer is remembered +until :func:`clear_resolution_cache`. The threads are daemons and do +not hold the process open. """ import ipaddress +import logging import socket import threading +import time + +# Hung getaddrinfo calls cannot be cancelled. The cap has to be large +# enough that a burst of slow-but-healthy lookups still finishes inside +# one deadline; a handful of hung names must not crowd everyone else out. +MAX_RESOLVER_THREADS = 16 + +_log = logging.getLogger("gambit.net") +_slots = threading.BoundedSemaphore(MAX_RESOLVER_THREADS) +_state_lock = threading.Lock() +_cache: dict[str, tuple[str, ...]] = {} +_pending: dict[str, threading.Event] = {} +# Hosts whose latest attempt did not run because every slot was busy. +# This is a signal, not a negative cache: the next call tries again. +_saturated: set[str] = set() +_generation = 0 + + +class ResolverSaturated(OSError): + """The name was not looked up; every resolver thread is still in libc. + + Distinct from an empty answer. Nothing was cached, so the next call + tries again. :func:`connect` raises this; :func:`resolve_ips` still + returns ``[]`` and never raises. + """ def is_ip_address(hostname: str) -> bool: @@ -40,6 +72,20 @@ def is_public_ip(ip: str) -> bool: return bool(getattr(addr, "is_global", False)) +def clear_resolution_cache() -> None: + """Forget cached DNS answers so the next scan resolves names again. + + A lookup already blocked in ``getaddrinfo`` keeps its slot until libc + returns. If this is called first, that late answer is not cached. + """ + global _generation + with _state_lock: + _generation += 1 + _cache.clear() + _pending.clear() + _saturated.clear() + + def resolve_ips(hostname: str, timeout: float) -> list[str]: """Resolve ``hostname`` to IP strings within ``timeout`` seconds. @@ -48,6 +94,14 @@ def resolve_ips(hostname: str, timeout: float) -> list[str]: 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. + + A finished lookup that returned at least one address is reused until + :func:`clear_resolution_cache`. An empty answer or a resolver error + (including a transient ``EAI_AGAIN``) is not cached, so the next call + tries again. The same host never occupies more than one resolver + thread. When every slot is already taken, this waits only until + ``timeout``, logs a warning, marks the host via + :func:`resolution_saturated`, and returns ``[]`` without caching. """ if is_ip_address(hostname): return [hostname] @@ -60,20 +114,108 @@ def resolve_ips(hostname: str, timeout: float) -> list[str]: if budget <= 0: return [] - box: dict = {} + name = str(hostname).strip() + key = _resolution_key(name) + deadline = time.monotonic() + budget + + with _state_lock: + cached = _cache.get(key) + if cached is not None: + return list(cached) + event = _pending.get(key) + generation = _generation + if event is None: + event = threading.Event() + _pending[key] = event + owner = True + else: + owner = False + + if not owner: + return _wait_for_resolution(key, event, deadline) + + remaining = deadline - time.monotonic() + if remaining <= 0 or not _slots.acquire(timeout=remaining): + with _state_lock: + if _generation == generation: + _saturated.add(key) + _abandon_resolution(key, event, generation) + _log.warning( + "DNS resolver saturated (%d lookups in flight); skipped %s. " + "The skip is not cached.", + MAX_RESOLVER_THREADS, + name, + ) + return [] + + with _state_lock: + _saturated.discard(key) def worker() -> None: + ips: tuple[str, ...] = () try: - box["ips"] = _resolve_system(hostname) - except Exception as exc: - box["error"] = exc + ips = tuple(_resolve_system(name)) + except Exception: + ips = () + finally: + try: + with _state_lock: + if _generation == generation and _pending.get(key) is event: + # Empty and failed answers stay uncached so a + # transient EAI_AGAIN can succeed on the next call. + if ips: + _cache[key] = ips + _pending.pop(key, None) + _saturated.discard(key) + event.set() + finally: + _slots.release() thread = threading.Thread(target=worker, name="gambit-resolve", daemon=True) - thread.start() - thread.join(budget) - if thread.is_alive() or "ips" not in box: + try: + thread.start() + except Exception: + _slots.release() + _abandon_resolution(key, event, generation) + raise + return _wait_for_resolution(key, event, deadline) + + +def _abandon_resolution(key: str, event: threading.Event, generation: int) -> None: + """Drop a lookup that never started so later calls can try again.""" + with _state_lock: + if _generation == generation and _pending.get(key) is event: + _pending.pop(key, None) + event.set() + + +def _wait_for_resolution( + key: str, event: threading.Event, deadline: float +) -> list[str]: + """Wait until ``deadline`` for ``key`` to land in the cache.""" + remaining = deadline - time.monotonic() + if remaining > 0: + event.wait(remaining) + with _state_lock: + cached = _cache.get(key) + if cached is None: return [] - return list(box["ips"]) + return list(cached) + + +def _resolution_key(hostname: str) -> str: + return str(hostname).strip().lower() + + +def resolution_saturated(hostname: str) -> bool: + """True when the latest lookup for ``hostname`` was skipped. + + The pool was full, so no ``getaddrinfo`` ran and nothing was cached. + A later :func:`resolve_ips` tries again and clears this once a lookup + actually starts. An empty answer from the resolver does not set it. + """ + with _state_lock: + return _resolution_key(hostname) in _saturated def _resolve_system(hostname: str) -> list[str]: @@ -110,14 +252,19 @@ def connect( 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. + not, and raises. Raises :class:`ResolverSaturated` when the lookup + was skipped because every resolver thread was busy, and ``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: + if resolved is None and resolution_saturated(hostname): + raise ResolverSaturated( + f"DNS resolver saturated; could not resolve {hostname}" + ) raise OSError(f"could not resolve {hostname}") last: OSError | None = None for ip in ips: @@ -130,4 +277,13 @@ def connect( raise last -__all__ = ["connect", "is_ip_address", "is_public_ip", "resolve_ips"] +__all__ = [ + "MAX_RESOLVER_THREADS", + "ResolverSaturated", + "clear_resolution_cache", + "connect", + "is_ip_address", + "is_public_ip", + "resolution_saturated", + "resolve_ips", +] diff --git a/gambit/pipeline.py b/gambit/pipeline.py index 31b2911..c1ec076 100644 --- a/gambit/pipeline.py +++ b/gambit/pipeline.py @@ -203,7 +203,7 @@ def host_reachable(host: str, *, timeout: float = 1.0) -> bool: even for hosts that answer on no web port. """ from gambit.http import set_preferred_port, set_preferred_scheme - from gambit.net import connect + from gambit.net import ResolverSaturated, connect if port_out_of_range(host): return False @@ -217,6 +217,11 @@ def host_reachable(host: str, *, timeout: float = 1.0) -> bool: try: with connect(name, port, timeout): pass + except ResolverSaturated: + # Not "offline". The caller reports the saturated pool. + # Do not try the remaining ports: each one would wait out + # the same deadline for a resolver slot. + raise except OSError: continue if explicit_port is None: @@ -228,6 +233,30 @@ def host_reachable(host: str, *, timeout: float = 1.0) -> bool: return False +def resolver_saturated_finding(host: str, detail: str = "") -> Finding: + """INFO finding for a lookup skipped because the resolver pool was full. + + This is not a negative cache and not "the name does not resolve". + """ + evidence = ( + "Every resolver thread was busy in getaddrinfo, so this host was " + "not looked up. The skip was not cached." + ) + if detail: + evidence = f"{evidence} ({detail})" + return Finding( + severity=Severity.INFO, + host=host, + title="DNS resolver saturated", + evidence=evidence, + why=( + "A hung lookup is still occupying the resolver. This is not " + "the same as a name that does not resolve." + ), + next_step="Re-run the scan. If it persists, a hostname lookup is hanging.", + ) + + def dedupe_hosts( hosts: Iterable[str] | None, max_hosts: Optional[int] = DEFAULT_MAX_HOSTS ) -> list[str]: @@ -360,6 +389,9 @@ def run_pipeline( it rejects gets a single INFO "Host did not respond" finding and its modules are skipped, so one dead host cannot stack every module's full timeout. (:func:`host_reachable` is the production preflight.) + :class:`gambit.net.ResolverSaturated` from the preflight is reported + as "DNS resolver saturated" instead, so a full resolver pool is not + described as an offline host. Findings are returned in host-then-module order; ranking for display stays Triage's job (``gambit.triage.rank``). @@ -371,25 +403,33 @@ def run_pipeline( target, extra_hosts, discover=discover, max_hosts=max_hosts ) + from gambit.net import ResolverSaturated + findings: list[Finding] = [] for index, host in enumerate(resolved): if index and delay: sleep(delay) - if preflight is not None and not preflight(host): - findings.append( - Finding( - severity=Severity.INFO, - host=host, - title="Host did not respond to a connection attempt", - evidence="TCP connection to the host failed or timed out " - "on the checked ports (pre-flight)", - why="The host may be offline, filtered, or no longer " - "in service, so the per-host checks were skipped.", - next_step="Confirm the host is still in use; if it is " - "a stale subdomain, consider removing its DNS records.", + if preflight is not None: + try: + reachable = preflight(host) + except ResolverSaturated as exc: + findings.append(resolver_saturated_finding(host, str(exc))) + continue + if not reachable: + findings.append( + Finding( + severity=Severity.INFO, + host=host, + title="Host did not respond to a connection attempt", + evidence="TCP connection to the host failed or timed out " + "on the checked ports (pre-flight)", + why="The host may be offline, filtered, or no longer " + "in service, so the per-host checks were skipped.", + next_step="Confirm the host is still in use; if it is " + "a stale subdomain, consider removing its DNS records.", + ) ) - ) - continue + continue findings.extend(run_for_host(host, modules)) return findings @@ -405,6 +445,7 @@ def run_pipeline( "host_reachable", "normalize_host", "port_out_of_range", + "resolver_saturated_finding", "run_for_host", "run_modules_for_host", "run_pipeline", diff --git a/tests/test_cli.py b/tests/test_cli.py index ba459ff..204bfd5 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,4 +1,7 @@ +import json + import gambit.__main__ as cli +from gambit.models import Severity from gambit.modules import gather, takeover @@ -28,6 +31,40 @@ def test_parser_defaults(): assert args.max_hosts == 25 assert args.yes is False assert args.output is None + assert args.insecure is False + + +def test_insecure_flag_is_opt_in(): + args = cli.build_parser().parse_args(["example.com", "--insecure"]) + assert args.insecure is True + help_text = cli.build_parser().format_help() + assert "--insecure" in help_text + assert "plain HTTP" in help_text + + +def test_insecure_report_notes_that_verification_was_disabled(): + finding = cli.verification_disabled_finding("example.com") + assert finding.severity == Severity.INFO + assert "verification disabled" in finding.title.lower() + text = cli.format_report([finding], "text", insecure=True) + assert text.startswith("NOTE:") + assert "--insecure" in text + assert "not retried" in text + markdown = cli.format_report([finding], "md", insecure=True) + assert markdown.startswith("> NOTE:") + payload = json.loads(cli.format_report([finding], "json", insecure=True)) + assert payload[0]["title"] == finding.title + assert "plain HTTP" in payload[0]["evidence"] + + +def test_main_prints_insecure_note(monkeypatch, capsys): + monkeypatch.setattr(cli, "collect_findings", lambda *args, **kwargs: []) + code = cli.main(["example.com", "--yes", "--insecure"]) + captured = capsys.readouterr() + assert code == 0 + assert "--insecure" in captured.err + assert "verification was disabled" in captured.err + assert "NOTE:" in captured.out def test_json_flag_is_kept_as_alias(): diff --git a/tests/test_gather.py b/tests/test_gather.py index f6ae319..1ea5f8c 100644 --- a/tests/test_gather.py +++ b/tests/test_gather.py @@ -115,6 +115,33 @@ def test_evaluate_dns_ignores_empty_values(): # --- crt.sh is queried once per host per process ------------------------- +def test_crtsh_request_keeps_verification_when_insecure(monkeypatch): + import gambit.http as http + + gather.clear_crtsh_cache() + seen = [] + + class FakeResponse: + def raise_for_status(self): + pass + + def json(self): + return [] + + def fake_get(url, **kwargs): + seen.append(kwargs.get("verify", "missing")) + return FakeResponse() + + monkeypatch.setattr(gather, "get", fake_get) + http.set_tls_verification(False) + try: + assert gather.discover_subdomains("example.com") == [] + finally: + http.set_tls_verification(True) + gather.clear_crtsh_cache() + assert seen == [True] + + def test_crtsh_fetch_is_cached_between_run_and_discovery(monkeypatch): gather.clear_crtsh_cache() calls = [] diff --git a/tests/test_http.py b/tests/test_http.py index 1cdb1d9..a6b78c6 100644 --- a/tests/test_http.py +++ b/tests/test_http.py @@ -25,6 +25,7 @@ def __init__(self, url): @pytest.fixture(autouse=True) def clean_cache(monkeypatch): http.clear_scheme_cache() + http.set_tls_verification(True) # 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 ( @@ -42,6 +43,7 @@ def clean_cache(monkeypatch): monkeypatch.delenv(name, raising=False) yield http.clear_scheme_cache() + http.set_tls_verification(True) def test_get_for_host_prefers_https(monkeypatch): @@ -896,20 +898,61 @@ def fake_request(method, url, **kwargs): 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).""" + """CA plus a server cert whose SAN is ``hostname``. Returns (ca, cert, key). + + The CA carries ``basicConstraints=critical,CA:TRUE`` and + ``keyUsage=critical,keyCertSign,cRLSign``. The leaf carries a SAN and + ``serverAuth``. Python 3.13 verifies with ``VERIFY_X509_STRICT`` by + default and rejects a bare self-signed certificate used as its own CA + ("CA cert does not include key usage extension"). + """ 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" + ca_conf = directory / "ca.cnf" srv_key = directory / "srv.key" csr = directory / "srv.csr" cert = directory / "srv.pem" + leaf_conf = directory / "leaf.cnf" + ca_conf.write_text( + "\n".join( + [ + "[req]", + "distinguished_name = dn", + "x509_extensions = v3_ca", + "prompt = no", + "[dn]", + "CN = Gambit Test CA", + "[v3_ca]", + "basicConstraints = critical,CA:TRUE", + "keyUsage = critical,keyCertSign,cRLSign", + "subjectKeyIdentifier = hash", + "", + ] + ), + encoding="utf-8", + ) + leaf_conf.write_text( + "\n".join( + [ + "[v3_leaf]", + "basicConstraints = CA:FALSE", + "keyUsage = critical,digitalSignature,keyEncipherment", + "extendedKeyUsage = serverAuth", + f"subjectAltName = DNS:{hostname}", + "authorityKeyIdentifier = keyid,issuer", + "", + ] + ), + encoding="utf-8", + ) subprocess.run( [ openssl, "req", "-x509", "-newkey", "rsa:2048", "-keyout", str(ca_key), "-out", str(ca_cert), - "-days", "1", "-nodes", "-subj", "/CN=Gambit Test CA", + "-days", "1", "-nodes", "-config", str(ca_conf), ], check=True, capture_output=True, cwd=directory, ) @@ -918,7 +961,6 @@ def _mint_hostname_cert(directory: Path, hostname: str) -> tuple[Path, Path, Pat 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, ) @@ -927,7 +969,7 @@ def _mint_hostname_cert(directory: Path, hostname: str) -> tuple[Path, Path, Pat openssl, "x509", "-req", "-in", str(csr), "-CA", str(ca_cert), "-CAkey", str(ca_key), "-CAcreateserial", "-out", str(cert), - "-days", "1", "-copy_extensions", "copy", + "-days", "1", "-extfile", str(leaf_conf), "-extensions", "v3_leaf", ], check=True, capture_output=True, cwd=directory, ) @@ -996,3 +1038,165 @@ def test_pinned_https_honours_requests_ca_bundle_and_rejects_hostname_mismatch(m finally: server.shutdown() server.server_close() + + +class _VerifyCapturingSession: + """Stand-in for ``requests.Session`` that records ``verify`` and URLs.""" + + def __init__(self, attempts, https_error=None): + self.trust_env = True + self._attempts = attempts + self._https_error = https_error + + def mount(self, prefix, adapter): + return None + + def request(self, method, url, **kwargs): + self._attempts.append((url, kwargs.get("verify", "missing"))) + if url.startswith("https://") and self._https_error is not None: + raise self._https_error + + 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 + + +def test_set_tls_verification_warning_filter_is_idempotent(): + import warnings + + from urllib3.exceptions import InsecureRequestWarning + + def installed(): + return [item for item in warnings.filters if item[2] is InsecureRequestWarning] + + http.set_tls_verification(True) + before = len(installed()) + try: + for _ in range(5): + http.set_tls_verification(False) + assert len(installed()) == before + 1 + for _ in range(5): + http.set_tls_verification(False) + assert len(installed()) == before + 1 + for _ in range(5): + http.set_tls_verification(True) + assert len(installed()) == before + finally: + http.set_tls_verification(True) + + +def test_insecure_sends_verify_false_and_ignores_the_env_bundle(monkeypatch): + attempts = [] + + def session_factory(): + return _VerifyCapturingSession(attempts) + + monkeypatch.setattr(http.requests, "Session", session_factory) + monkeypatch.setattr(http, "resolve_ips", lambda host, timeout: ["93.184.216.34"]) + monkeypatch.setenv("REQUESTS_CA_BUNDLE", "/tmp/gambit-test-ca.pem") + http.set_tls_verification(False) + try: + http.get("https://example.com/") + assert attempts[-1] == ("https://example.com/", False) + + http.get("https://example.com/", verify="/explicit.pem") + assert attempts[-1][1] == "/explicit.pem" + + http.get("https://example.com/", verify=True) + assert attempts[-1][1] == "/tmp/gambit-test-ca.pem" + finally: + http.set_tls_verification(True) + + +def test_insecure_does_not_downgrade_a_verification_failure(monkeypatch): + attempts = [] + error = requests.exceptions.SSLError( + "[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: self-signed certificate" + ) + + def session_factory(): + return _VerifyCapturingSession(attempts, https_error=error) + + monkeypatch.setattr(http.requests, "Session", session_factory) + monkeypatch.setattr(http, "resolve_ips", lambda host, timeout: ["93.184.216.34"]) + http.set_tls_verification(False) + try: + with pytest.raises(requests.exceptions.SSLError, match="not retrying over HTTP"): + http.get_for_host("internal.example") + finally: + http.set_tls_verification(True) + assert [url for url, _verify in attempts] == ["https://internal.example/"] + assert attempts[0][1] is False + assert http.preferred_scheme("internal.example") is None + + +def test_insecure_protocol_mismatch_still_falls_back_to_http(monkeypatch): + attempts = [] + error = requests.exceptions.SSLError("[SSL: WRONG_VERSION_NUMBER] wrong version number") + + def session_factory(): + return _VerifyCapturingSession(attempts, https_error=error) + + monkeypatch.setattr(http.requests, "Session", session_factory) + monkeypatch.setattr(http, "resolve_ips", lambda host, timeout: ["93.184.216.34"]) + http.set_tls_verification(False) + try: + response = http.get_for_host("plain.example") + finally: + http.set_tls_verification(True) + assert response.url == "http://plain.example/" + assert [url for url, _verify in attempts] == [ + "https://plain.example/", + "http://plain.example/", + ] + assert attempts[0][1] is False + assert http.preferred_scheme("plain.example") == "http" + + +def test_insecure_accepts_untrusted_and_mismatched_certs_over_https(monkeypatch): + if shutil.which("openssl") is None: + pytest.skip("openssl is required to mint a certificate") + with tempfile.TemporaryDirectory() as tmp: + _ca, cert, key = _mint_hostname_cert(Path(tmp), "pinned.gambit.test") + server = _CountingServer(("127.0.0.1", 0), _RecordingHandler) + server.routes = {"/": (200, {}, b"untrusted")} + 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"]) + http.set_tls_verification(False) + try: + response = http.get_for_host(f"pinned.gambit.test:{port}", timeout=2) + assert response.status_code == 200 + assert response.content == b"untrusted" + assert response.url.startswith("https://") + assert _wait_until(lambda: server.accepts >= 1) + assert server.accepts == 1 + + mismatched = http.get_for_host(f"wrong.gambit.test:{port}", timeout=2) + assert mismatched.status_code == 200 + assert mismatched.url.startswith("https://") + assert _wait_until(lambda: server.accepts >= 2) + # Still one handshake per host. Plain HTTP is not tried. + assert server.accepts == 2 + finally: + http.set_tls_verification(True) + server.shutdown() + server.server_close() diff --git a/tests/test_net.py b/tests/test_net.py index b9e3b7c..5254f1f 100644 --- a/tests/test_net.py +++ b/tests/test_net.py @@ -1,9 +1,53 @@ import socket +import threading import time import pytest -from gambit.net import connect, is_public_ip, resolve_ips +import logging + +from gambit.net import ( + MAX_RESOLVER_THREADS, + ResolverSaturated, + clear_resolution_cache, + connect, + is_public_ip, + resolution_saturated, + resolve_ips, +) + + +def _resolver_threads() -> list[threading.Thread]: + return [ + thread + for thread in threading.enumerate() + if thread.name.startswith("gambit-resolve") and thread.is_alive() + ] + + +def _join_resolver_threads(timeout: float = 2.0) -> None: + """Wait out lookups left in ``getaddrinfo`` so they release their slot. + + The resolver cap is process-wide. A sleeper from an earlier test still + occupies one slot, and on a slow scheduler that slot can free in the + middle of the next test. + """ + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + alive = _resolver_threads() + if not alive: + return + remaining = deadline - time.monotonic() + for thread in alive: + thread.join(min(0.1, max(0.0, remaining))) + + +@pytest.fixture(autouse=True) +def _clear_dns_cache(): + clear_resolution_cache() + yield + clear_resolution_cache() + _join_resolver_threads(2.0) def test_public_address_check_rejects_metadata_and_private_ranges(): @@ -47,21 +91,28 @@ def close(self): def test_resolve_ips_uses_one_deadline_for_every_family(monkeypatch): calls = [] + release = threading.Event() def slow_getaddrinfo(host, port, family=0, type=0, proto=0, flags=0): calls.append(host) - time.sleep(1.0) + # One second unless the test is done. Leaving the thread in sleep + # would hold a resolver slot into the next test. + release.wait(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"] + try: + started = time.monotonic() + assert resolve_ips("example.com", 0.3) == [] + elapsed = time.monotonic() - started + assert elapsed < 0.8 + assert calls == ["example.com"] + finally: + release.set() + _join_resolver_threads() def test_resolve_ips_localhost_uses_the_system_resolver(): @@ -118,6 +169,200 @@ def close(self): connect("example.com", 80, 1.5, resolved=[]) +def test_resolve_ips_reuses_a_finished_lookup(monkeypatch): + calls = [] + + def fake_getaddrinfo(host, port, family=0, type=0, proto=0, flags=0): + calls.append(host) + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("203.0.113.9", 0))] + + monkeypatch.setattr(socket, "getaddrinfo", fake_getaddrinfo) + assert resolve_ips("Cache.Example", 1) == ["203.0.113.9"] + assert resolve_ips("cache.example", 1) == ["203.0.113.9"] + assert calls == ["Cache.Example"] + clear_resolution_cache() + assert resolve_ips("cache.example", 1) == ["203.0.113.9"] + assert calls == ["Cache.Example", "cache.example"] + + +def test_hanging_resolver_across_many_hosts_does_not_grow_threads(monkeypatch): + release = threading.Event() + entered = [] + entered_lock = threading.Lock() + + def hang(host, port, family=0, type=0, proto=0, flags=0): + with entered_lock: + entered.append(host) + # Loop so a spurious wakeup cannot drop the slot mid-test. + while not release.is_set(): + release.wait(0.2) + return [] + + monkeypatch.setattr(socket, "getaddrinfo", hang) + before = { + thread.ident + for thread in threading.enumerate() + if thread.name.startswith("gambit-resolve") + } + hosts = [f"hung-{index}.example" for index in range(MAX_RESOLVER_THREADS + 20)] + errors = [] + + def lookup(host): + try: + if resolve_ips(host, 0.4) != []: + errors.append(host) + except Exception as exc: # the resolver must not raise + errors.append(repr(exc)) + + callers = [threading.Thread(target=lookup, args=(host,)) for host in hosts] + try: + for caller in callers: + caller.start() + for caller in callers: + caller.join(3) + assert errors == [] + assert all(not caller.is_alive() for caller in callers) + alive = [ + thread + for thread in threading.enumerate() + if thread.name.startswith("gambit-resolve") + and thread.is_alive() + and thread.ident not in before + ] + assert len(hosts) > MAX_RESOLVER_THREADS + assert 1 <= len(entered) <= MAX_RESOLVER_THREADS + assert len(alive) <= MAX_RESOLVER_THREADS + # A host that is already hung must not start a second lookup. + hung_host = entered[0] + with entered_lock: + before_count = entered.count(hung_host) + assert resolve_ips(hung_host, 0.1) == [] + with entered_lock: + assert entered.count(hung_host) == before_count + finally: + release.set() + for caller in callers: + caller.join(2) + _join_resolver_threads(3) + + +def test_transient_resolver_failure_is_not_cached(monkeypatch): + calls = [] + + def flaky(host, port, family=0, type=0, proto=0, flags=0): + calls.append(host) + if len(calls) == 1: + raise socket.gaierror(socket.EAI_AGAIN, "temporary failure in name resolution") + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("203.0.113.7", 0))] + + monkeypatch.setattr(socket, "getaddrinfo", flaky) + assert resolve_ips("flaky.example", 1) == [] + assert resolution_saturated("flaky.example") is False + assert resolve_ips("flaky.example", 1) == ["203.0.113.7"] + assert calls == ["flaky.example", "flaky.example"] + + +def test_saturated_resolver_is_signalled_and_not_cached(monkeypatch, caplog): + # A lookup still running from an earlier test holds one shared slot. + # On Windows that straggler can finish between two probes and make the + # second probe look like a real lookup instead of a saturated skip. + _join_resolver_threads(2.0) + release = threading.Event() + entered = [] + entered_lock = threading.Lock() + full = threading.Event() + + def hang(host, port, family=0, type=0, proto=0, flags=0): + with entered_lock: + entered.append(host) + if len(entered) >= MAX_RESOLVER_THREADS: + full.set() + while not release.is_set(): + release.wait(0.2) + return [] + + def fill(prefix: str) -> list[threading.Thread]: + full.clear() + threads = [ + threading.Thread(target=resolve_ips, args=(f"{prefix}-{index}.example", 5)) + for index in range(MAX_RESOLVER_THREADS) + ] + for thread in threads: + thread.start() + assert full.wait(5), "resolver pool did not fill" + return threads + + monkeypatch.setattr(socket, "getaddrinfo", hang) + holders: list[threading.Thread] = [] + try: + holders = fill("hold") + with entered_lock: + assert len(entered) >= MAX_RESOLVER_THREADS + with caplog.at_level(logging.WARNING, logger="gambit.net"): + assert resolve_ips("healthy.example", 0.3) == [] + assert resolution_saturated("healthy.example") is True + assert any( + "saturated" in record.message.lower() and "healthy.example" in record.message + for record in caplog.records + ) + assert any("not cached" in record.message.lower() for record in caplog.records) + # The pool is still full, so this is another skip, not a cached miss. + assert resolve_ips("healthy.example", 0.3) == [] + assert resolution_saturated("healthy.example") is True + with entered_lock: + assert entered.count("healthy.example") == 0 + finally: + release.set() + for holder in holders: + holder.join(2) + _join_resolver_threads(2.0) + + def answer(host, port, family=0, type=0, proto=0, flags=0): + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("198.51.100.8", 0))] + + monkeypatch.setattr(socket, "getaddrinfo", answer) + assert resolve_ips("healthy.example", 1) == ["198.51.100.8"] + assert resolution_saturated("healthy.example") is False + + # connect distinguishes a full pool from an unresolvable name. + release.clear() + with entered_lock: + entered.clear() + monkeypatch.setattr(socket, "getaddrinfo", hang) + blockers: list[threading.Thread] = [] + try: + blockers = fill("block") + with pytest.raises(ResolverSaturated): + connect("again.example", 80, 0.3) + assert resolution_saturated("again.example") is True + finally: + release.set() + for blocker in blockers: + blocker.join(2) + _join_resolver_threads(2.0) + + +def test_slow_burst_finishes_inside_one_deadline(monkeypatch): + def slow(host, port, family=0, type=0, proto=0, flags=0): + time.sleep(0.3) + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("203.0.113.10", 0))] + + monkeypatch.setattr(socket, "getaddrinfo", slow) + hosts = [f"slow-{index}.example" for index in range(20)] + results: dict[str, list[str]] = {} + + def lookup(host): + results[host] = resolve_ips(host, 1.0) + + callers = [threading.Thread(target=lookup, args=(host,)) for host in hosts] + for caller in callers: + caller.start() + for caller in callers: + caller.join(3) + assert all(not caller.is_alive() for caller in callers) + assert [results.get(host) for host in hosts] == [["203.0.113.10"]] * len(hosts) + + def test_connect_refuses_a_port_that_would_wrap(monkeypatch): called = [] monkeypatch.setattr( diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index da35c10..aef4312 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -1,3 +1,5 @@ +import pytest + from gambit.models import Finding, Severity from gambit.pipeline import dedupe_hosts, discover_hosts, normalize_host, run_for_host, run_pipeline @@ -249,6 +251,38 @@ def __init__(self, url): http.clear_scheme_cache() +def test_preflight_saturation_is_not_reported_as_offline(): + from gambit.net import ResolverSaturated + + def saturated(host): + raise ResolverSaturated("DNS resolver saturated; could not resolve " + host) + + findings = run_pipeline( + "example.com", + [GoodModule()], + hosts=["example.com"], + delay=0, + sleep=lambda s: None, + preflight=saturated, + ) + assert len(findings) == 1 + assert findings[0].title == "DNS resolver saturated" + assert "did not respond" not in findings[0].title + assert "good:" not in findings[0].title + + +def test_host_reachable_propagates_resolver_saturation(monkeypatch): + from gambit.net import ResolverSaturated + from gambit.pipeline import host_reachable + + def fake_connect(*args, **kwargs): + raise ResolverSaturated("busy") + + monkeypatch.setattr("gambit.net.connect", fake_connect) + with pytest.raises(ResolverSaturated): + host_reachable("busy.example", timeout=0.1) + + 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 fbd3f86..0cb3f11 100644 --- a/tests/test_ports.py +++ b/tests/test_ports.py @@ -63,6 +63,15 @@ def fake_connect(hostname, port, timeout, resolved=None): assert ("example.com", 443, ["203.0.113.1"]) in tried +def test_saturated_resolver_is_reported_instead_of_skipped(monkeypatch): + monkeypatch.setattr("gambit.net.resolve_ips", lambda host, timeout: []) + monkeypatch.setattr("gambit.net.resolution_saturated", lambda host: host == "example.com") + findings = run("example.com") + assert len(findings) == 1 + assert findings[0].title == "DNS resolver saturated" + assert "not cached" in findings[0].evidence + + def test_run_resolves_the_host_once(monkeypatch): resolved_calls = [] connect_calls = [] diff --git a/tests/test_tls.py b/tests/test_tls.py index b8708a1..23e15cd 100644 --- a/tests/test_tls.py +++ b/tests/test_tls.py @@ -1,7 +1,9 @@ import ssl from datetime import datetime, timezone +import gambit.http as http from gambit.models import Severity +from gambit.modules import tls as tls_module from gambit.modules.tls import ( certificate_from_socket, days_until_expiry, @@ -156,6 +158,26 @@ def wrap_socket(self, sock, server_hostname=None): assert "2020" in cert["notAfter"] +def test_insecure_tls_still_classifies_a_verification_failure(monkeypatch): + seen = [] + + def fake_handshake(hostname, port, context): + seen.append(context.verify_mode) + if context.verify_mode != ssl.CERT_NONE: + raise ssl.SSLCertVerificationError(1, "self-signed certificate") + return ({"notAfter": "Dec 1 00:00:00 2099 GMT"}, "TLSv1.3") + + monkeypatch.setattr(tls_module, "_handshake", fake_handshake) + http.set_tls_verification(False) + try: + findings = tls_module.run("example.com") + finally: + http.set_tls_verification(True) + assert seen[0] != ssl.CERT_NONE + assert ssl.CERT_NONE in seen + assert any("self-signed" in finding.title.lower() for finding in findings) + + def test_hostname_mismatch_and_self_signed_are_info(): findings = evaluate( "example.com",