From 56ea38a4cc860890e20e940cc30caed4521338eb Mon Sep 17 00:00:00 2001 From: DevCop95 Date: Mon, 17 Aug 2026 08:26:47 -0500 Subject: [PATCH] Release v1.0.5 --- CHANGELOG.md | 11 + README.md | 63 +++++- pyproject.toml | 2 +- src/recons101x.py | 429 ++++++++++++++++++++++++++++++++++++--- tests/test_recons101x.py | 90 +++++++- 5 files changed, 566 insertions(+), 29 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1a849d0..07fbe9e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,16 @@ All notable changes to this project are documented here. +## [1.0.5] - 2026-08-17 + +### Added + +- Optional certificate enrichment through the Shodan CTL domain and certificate endpoints. +- Active-node status based on DNS resolution, with green terminal output. +- Opt-in TCP reachability probes and an `--active-only` filter for stale CT names. +- High-level `--live` mode for the common active-host workflow. +- Reachability states and per-domain DNS/TCP/HTTP/stale summaries. + ## [1.0.0] - 2026-07-28 ### Added @@ -15,3 +25,4 @@ All notable changes to this project are documented here. - Cross-platform tests and automated GitHub releases. [1.0.0]: https://github.com/DevCop95/shodan_reconsx/releases/tag/v1.0.0 +[1.0.5]: https://github.com/DevCop95/shodan_reconsx/releases/tag/v1.0.5 diff --git a/README.md b/README.md index f397d67..3cab505 100644 --- a/README.md +++ b/README.md @@ -91,8 +91,64 @@ removed automatically. ``` With `--resolve`, TXT output becomes -`domainhostnameip1,ip2`. This option performs DNS lookups only. It -does not scan ports or send requests to discovered hostnames. +`domainhostnameip1,ip2`. Add `--status` to append the more precise +state. `DNS_ONLY` means that the hostname resolves but no web service was +verified; `TCP_REACHABLE` means that 443/80 accepted a connection; +`HTTP_REACHABLE` means that HTTP/HTTPS returned a response; and `STALE` means +that DNS did not resolve. Reachable states are shown in green in an +interactive terminal. Without `--probe`, resolution only distinguishes +`DNS_ONLY` and `STALE`. + +Because certificate transparency can contain old names, DNS activity is only a +first filter. To verify that a web service is reachable, use the explicit +probe and keep only active results: + +```sh +./scan.sh example.com --probe --active-only --status --probe-timeout 3 +``` + +`--probe` makes TCP and HTTP/HTTPS checks on ports 443 and 80 and records the +reachable ports and HTTP status codes in JSON as `ports` and `http`. It is +intentionally opt-in and should only be used on domains you own or are +authorized to test. With `--probe`, `--active-only` removes `STALE` and +`DNS_ONLY` names from the output; with `--resolve` alone it keeps names that +resolve in DNS. Without the filter, the tool reports every state. + +When resolving or probing, stderr also prints a compact summary such as: + +```text +[+] Summary: 312 DNS | 86 TCP | 54 HTTP | 936 stale +``` + +The JSON output contains the same counters in each domain's `summary` object, +while retaining the complete hostname details unless `--active-only` is used. + +In an interactive terminal, status output uses a fixed-width table and +truncates long hostnames or IPv6 lists so rows do not wrap. Redirected output +and files keep the complete tab-separated values. + +For normal use, the concise equivalent is: + +```sh +./scan.sh example.com --live +``` + +`--live` is the high-level mode: it resolves names, checks web reachability, +keeps only active names, and displays their status. The individual switches +remain available for automation and advanced tuning. + +## Certificate enrichment + +Use `--certificates` with JSON output to query all three Shodan CTL resources: + +```sh +./scan.sh example.com --resolve --status --certificates --format json --output results.json +``` + +The JSON result keeps the existing `domain` and `hostnames` fields, adds an +`active` boolean to each hostname when `--resolve` is used, and adds a +`certificates` array containing each SHA-256 hash and its response from +`/api/v1/cert/{sha256}`. The extra certificate requests are opt-in. ## Options @@ -101,7 +157,8 @@ does not scan ports or send requests to discovered hostnames. ``` Available controls include HTTP timeout, retry count, concurrent DNS workers, -input files, output files, and TXT or JSON formatting. +input files, output files, TXT or JSON formatting, active-node status, color +control, and optional certificate enrichment. ## Project Structure diff --git a/pyproject.toml b/pyproject.toml index cec60fa..72f837e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "recons101x" -version = "1.0.0" +version = "1.0.5" description = "Portable passive hostname reconnaissance through Shodan CTL" readme = "README.md" requires-python = ">=3.10" diff --git a/src/recons101x.py b/src/recons101x.py index fc0671f..00f77c6 100644 --- a/src/recons101x.py +++ b/src/recons101x.py @@ -9,6 +9,7 @@ import os import re import socket +import shutil import sys import time import urllib.error @@ -17,9 +18,13 @@ from typing import Iterable -API_URL = "https://ctl.shodan.io/api/v1/domain/{domain}/hostnames" -__version__ = "1.0.0" +DOMAIN_API_URL = "https://ctl.shodan.io/api/v1/domain/{domain}" +HOSTNAMES_API_URL = "https://ctl.shodan.io/api/v1/domain/{domain}/hostnames" +CERT_API_URL = "https://ctl.shodan.io/api/v1/cert/{sha256}" +API_URL = HOSTNAMES_API_URL # Backwards-compatible name for integrations. +__version__ = "1.0.5" LABEL_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$") +SHA256_RE = re.compile(r"^[0-9a-f]{64}$", re.IGNORECASE) BANNER = r""" ██▀███ ▓█████ ▄████▄ ▒█████ ███▄ █ ██████ ████ █████ ████ ▒██ ██▒ ▓██ ▒ ██▒▓█ ▀ ▒██▀ ▀█ ▒██▒ ██▒ ██ ▀█ █ ▒██ ▒ ░░███ ███░░░███ ░░███ ▒▒ █ █ ▒░ @@ -64,26 +69,72 @@ def normalize_domain(value: str) -> str: def fetch_hostnames(domain: str, timeout: float, retries: int) -> list[str]: + payload = fetch_json( + HOSTNAMES_API_URL.format(domain=domain), timeout=timeout, retries=retries + ) + if not isinstance(payload, list) or not all(isinstance(item, str) for item in payload): + raise RuntimeError(f"could not query {domain}: unexpected API response") + return sorted(set(payload)) + + +def fetch_json(url: str, timeout: float, retries: int) -> object: + """Fetch and decode one CTL JSON response with the same retry policy.""" request = urllib.request.Request( - API_URL.format(domain=domain), - headers={"Accept": "application/json", "User-Agent": "shodan-domain-recon/1.0"}, + url, + headers={"Accept": "application/json", "User-Agent": "shodan-domain-recon/1.0.5"}, ) last_error: Exception | None = None for attempt in range(retries + 1): try: - # The request always uses the fixed HTTPS API_URL and a validated domain. + # URLs are assembled only from validated domains or SHA-256 hashes. with urllib.request.urlopen(request, timeout=timeout) as response: # nosec B310 payload = json.load(response) - if not isinstance(payload, list) or not all(isinstance(item, str) for item in payload): - raise ValueError("unexpected API response") - return sorted(set(payload)) + return payload except (urllib.error.URLError, TimeoutError, ValueError, json.JSONDecodeError) as exc: last_error = exc if attempt < retries: time.sleep(2**attempt) - raise RuntimeError(f"could not query {domain}: {last_error}") + raise RuntimeError(f"could not query {url}: {last_error}") + + +def _certificate_hash(item: object) -> str | None: + if isinstance(item, str): + candidate = item.strip().lower() + elif isinstance(item, dict): + candidate = next( + ( + str(item[key]).strip().lower() + for key in ("sha256", "sha-256", "hash", "id") + if item.get(key) + ), + "", + ) + else: + candidate = "" + return candidate if SHA256_RE.fullmatch(candidate) else None + + +def fetch_certificate_hashes(domain: str, timeout: float, retries: int) -> list[str]: + payload = fetch_json( + DOMAIN_API_URL.format(domain=domain), timeout=timeout, retries=retries + ) + items: object = payload + if isinstance(payload, dict): + for key in ("certificates", "hashes", "data", "results"): + if isinstance(payload.get(key), list): + items = payload[key] + break + if not isinstance(items, list): + raise RuntimeError(f"could not query certificates for {domain}: unexpected API response") + return sorted({certificate_hash for item in items if (certificate_hash := _certificate_hash(item))}) + + +def fetch_certificate(sha256: str, timeout: float, retries: int) -> object: + if not SHA256_RE.fullmatch(sha256): + raise ValueError("invalid certificate SHA-256") + return fetch_json(CERT_API_URL.format(sha256=sha256.lower()), timeout=timeout, retries=retries) def resolve_hostname(hostname: str) -> list[str]: @@ -94,6 +145,87 @@ def resolve_hostname(hostname: str) -> list[str]: return sorted({address[4][0] for address in addresses}) +def probe_hostname(hostname: str, timeout: float) -> list[int]: + """Return reachable web ports; this is opt-in because it makes network probes.""" + reachable_ports = [] + for port in (443, 80): + try: + with socket.create_connection((hostname, port), timeout=timeout): + reachable_ports.append(port) + except (OSError, TimeoutError): + continue + return reachable_ports + + +def probe_http_services(hostname: str, ports: list[int], timeout: float) -> dict[str, int]: + """Return HTTP status codes for reachable HTTPS/HTTP services.""" + statuses: dict[str, int] = {} + for port in ports: + scheme = "https" if port == 443 else "http" + request = urllib.request.Request( + f"{scheme}://{hostname}/", + method="HEAD", + headers={"User-Agent": "shodan-domain-recon/1.0.5"}, + ) + try: + with urllib.request.urlopen(request, timeout=timeout) as response: # nosec B310 + statuses[scheme] = int(getattr(response, "status", 200)) + except urllib.error.HTTPError as exc: + # An HTTP error still proves that an HTTP server answered. + statuses[scheme] = exc.code + except (urllib.error.URLError, OSError, TimeoutError): + continue + return statuses + + +def classify_state(ips: list[str], ports: list[int], http: dict[str, int]) -> str: + if http: + return "HTTP_REACHABLE" + if ports: + return "TCP_REACHABLE" + if ips: + return "DNS_ONLY" + return "STALE" + + +def summarize_entries(entries: list[dict[str, object]]) -> dict[str, int]: + states = [str(entry.get("state", "STALE")) for entry in entries] + return { + "total": len(entries), + "dns": sum(state != "STALE" for state in states), + "tcp": sum(state in {"TCP_REACHABLE", "HTTP_REACHABLE"} for state in states), + "http": sum(state == "HTTP_REACHABLE" for state in states), + "stale": sum(state == "STALE" for state in states), + } + + +def build_hostname_entry( + hostname: str, + ips: list[str], + ports: list[int], + http: dict[str, int], + include_ips: bool, + resolve: bool, + probe: bool, +) -> dict[str, object]: + entry: dict[str, object] = { + "hostname": hostname, + "ips": ips if include_ips else [], + } + if resolve or probe: + state = classify_state(ips, ports, http) + entry["state"] = state + entry["active"] = ( + state in {"TCP_REACHABLE", "HTTP_REACHABLE"} + if probe + else bool(ips) + ) + if probe: + entry["ports"] = ports + entry["http"] = http + return entry + + def read_domains(values: Iterable[str], input_file: Path | None) -> list[str]: candidates = list(values) if input_file: @@ -111,7 +243,17 @@ def read_domains(values: Iterable[str], input_file: Path | None) -> list[str]: return list(dict.fromkeys(normalize_domain(value) for value in candidates)) -def format_text(results: list[dict[str, object]], include_ips: bool) -> str: +def format_text( + results: list[dict[str, object]], + include_ips: bool, + include_status: bool = False, + color: bool = False, + pretty: bool = False, + width: int = 120, +) -> str: + if pretty: + return format_pretty(results, include_ips, include_status, color, width) + lines = [] for result in results: domain = str(result["domain"]) @@ -120,10 +262,106 @@ def format_text(results: list[dict[str, object]], include_ips: bool) -> str: fields = [domain, hostname] if include_ips: fields.append(",".join(entry["ips"])) + if include_status: + active = bool(entry.get("active")) + status = str(entry.get("state", "ACTIVE" if active else "INACTIVE")) + state_colors = { + "HTTP_REACHABLE": "1;32", + "TCP_REACHABLE": "1;32", + "DNS_ONLY": "1;33", + "STALE": "2;31", + "ACTIVE": "1;32", + "INACTIVE": "2;31", + } + if color and status in state_colors: + status = f"\033[{state_colors[status]}m{status}\033[0m" + fields.append(status) lines.append("\t".join(fields)) return "\n".join(lines) + ("\n" if lines else "") +def shorten(value: str, width: int) -> str: + if width <= 0: + return "" + if len(value) <= width: + return value + if width <= 3: + return value[:width] + return value[: width - 3] + "..." + + +def format_pretty( + results: list[dict[str, object]], + include_ips: bool, + include_status: bool, + color: bool, + width: int, +) -> str: + """Render a fixed-width human table without wrapping long hostnames/IPs.""" + rows: list[tuple[str, str, str, str]] = [] + for result in results: + domain = str(result["domain"]) + for entry in result["hostnames"]: + ips = ",".join(str(ip) for ip in entry.get("ips", [])) if include_ips else "-" + if len(ips) > 1 and "," in ips: + ips = ips.split(",", 1)[0] + ",..." + status = "" + if include_status: + active = bool(entry.get("active")) + status = str(entry.get("state", "ACTIVE" if active else "INACTIVE")) + rows.append((domain, str(entry["hostname"]), ips, status)) + + if not rows: + return "" + + state_width = max(7, max(len(row[3]) for row in rows)) if include_status else 0 + domain_width = min(24, max(len("DOMAIN"), max(len(row[0]) for row in rows))) + ip_width = min(38, max(len("IP / PORT"), max(len(row[2]) for row in rows))) + separators = 9 if include_status else 7 + host_width = max(18, width - domain_width - ip_width - state_width - separators) + if include_status: + header = ( + f"{'DOMAIN':<{domain_width}} {'HOSTNAME':<{host_width}} " + f"{'IP / PORT':<{ip_width}} {'STATE':<{state_width}}" + ) + else: + header = f"{'DOMAIN':<{domain_width}} {'HOSTNAME':<{host_width}} {'IP / PORT':<{ip_width}}" + + lines = [header, "-" * min(width, len(header))] + state_colors = { + "HTTP_REACHABLE": "1;32", + "TCP_REACHABLE": "1;32", + "DNS_ONLY": "1;33", + "STALE": "2;31", + "ACTIVE": "1;32", + "INACTIVE": "2;31", + } + for domain, hostname, ips, status in rows: + fields = [ + shorten(domain, domain_width).ljust(domain_width), + shorten(hostname, host_width).ljust(host_width), + shorten(ips, ip_width).ljust(ip_width), + ] + if include_status: + visible_status = shorten(status, state_width).ljust(state_width) + if color and status in state_colors: + visible_status = f"\033[{state_colors[status]}m{visible_status}\033[0m" + fields.append(visible_status) + lines.append(" ".join(fields).rstrip()) + return "\n".join(lines) + "\n" + + +def color_enabled(force: bool = False, stream: object = sys.stdout) -> bool: + return bool( + "NO_COLOR" not in os.environ + and (force or (hasattr(stream, "isatty") and stream.isatty())) + ) + + +def colorize(text: str, code: str, enabled: bool) -> str: + return f"\033[{code}m{text}\033[0m" if enabled else text + + def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Enumerate hostnames for one or more domains through Shodan CTL." @@ -132,19 +370,69 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument("-i", "--input", type=Path, help="file containing one domain per line") parser.add_argument("-o", "--output", type=Path, help="output file (default: stdout)") parser.add_argument("-f", "--format", choices=("txt", "json"), default="txt") + parser.add_argument( + "--live", + action="store_true", + help="show only hostnames with a reachable web service", + ) parser.add_argument("--resolve", action="store_true", help="resolve A/AAAA records for each hostname") + parser.add_argument( + "--probe", + action="store_true", + help="probe TCP and HTTP(S) on ports 443/80 (opt-in)", + ) + parser.add_argument( + "--active-only", + action="store_true", + help="output only active hostnames (requires --resolve or --probe)", + ) + parser.add_argument( + "--status", + action="store_true", + help="add reachability state to text output (requires --resolve or --probe)", + ) + parser.add_argument( + "--certificates", + action="store_true", + help="include matching certificate hashes and CTL certificate data in JSON output", + ) + color = parser.add_mutually_exclusive_group() + color.add_argument("--color", action="store_true", help="force color in text status output") + color.add_argument("--no-color", action="store_true", help="disable ANSI colors") parser.add_argument("--workers", type=int, default=10, help="concurrent DNS lookups (10)") parser.add_argument("--timeout", type=float, default=15, help="HTTP timeout in seconds (15)") + parser.add_argument( + "--probe-timeout", + type=float, + default=3, + help="network timeout per probe in seconds (3)", + ) parser.add_argument("--retries", type=int, default=2, help="HTTP retries (2)") parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}") return parser +def apply_mode(args: argparse.Namespace) -> argparse.Namespace: + """Expand user-facing modes into the existing low-level switches.""" + if args.live: + args.resolve = True + args.probe = True + args.active_only = True + args.status = True + return args + + def main(argv: list[str] | None = None) -> int: print_banner() - args = build_parser().parse_args(argv) - if args.workers < 1 or args.timeout <= 0 or args.retries < 0: - print("error: workers must be >= 1, timeout > 0, and retries >= 0", file=sys.stderr) + args = apply_mode(build_parser().parse_args(argv)) + if args.workers < 1 or args.timeout <= 0 or args.probe_timeout <= 0 or args.retries < 0: + print("error: workers must be >= 1, timeouts > 0, and retries >= 0", file=sys.stderr) + return 2 + if args.status and not (args.resolve or args.probe): + print("error: --status requires --resolve or --probe", file=sys.stderr) + return 2 + if args.active_only and not (args.resolve or args.probe): + print("error: --active-only requires --resolve or --probe", file=sys.stderr) return 2 try: @@ -165,25 +453,118 @@ def main(argv: list[str] | None = None) -> int: continue ips_by_hostname: dict[str, list[str]] = {} - if args.resolve and hostnames: + if (args.resolve or args.probe) and hostnames: with concurrent.futures.ThreadPoolExecutor(max_workers=args.workers) as executor: ips_by_hostname = dict(zip(hostnames, executor.map(resolve_hostname, hostnames))) - results.append( - { - "domain": domain, - "hostnames": [ - {"hostname": hostname, "ips": ips_by_hostname.get(hostname, [])} - for hostname in hostnames - ], - } + ports_by_hostname: dict[str, list[int]] = {} + if args.probe and hostnames: + with concurrent.futures.ThreadPoolExecutor(max_workers=args.workers) as executor: + ports_by_hostname = dict( + zip( + hostnames, + executor.map( + lambda hostname: probe_hostname(hostname, args.probe_timeout), + hostnames, + ), + ) + ) + + http_by_hostname: dict[str, dict[str, int]] = {} + if args.probe and hostnames: + with concurrent.futures.ThreadPoolExecutor(max_workers=args.workers) as executor: + http_by_hostname = dict( + zip( + hostnames, + executor.map( + lambda hostname: probe_http_services( + hostname, + ports_by_hostname.get(hostname, []), + args.probe_timeout, + ), + hostnames, + ), + ) + ) + + all_hostname_entries = [ + build_hostname_entry( + hostname, + ips_by_hostname.get(hostname, []), + ports_by_hostname.get(hostname, []), + http_by_hostname.get(hostname, {}), + include_ips=args.resolve, + resolve=args.resolve, + probe=args.probe, + ) + for hostname in hostnames + ] + summary = summarize_entries(all_hostname_entries) + active_count = sum(1 for entry in all_hostname_entries if entry.get("active")) + hostname_entries = ( + [entry for entry in all_hostname_entries if entry.get("active")] + if args.active_only + else all_hostname_entries ) - print(f"[+] Found {len(hostnames)} hostnames", file=sys.stderr) + result: dict[str, object] = {"domain": domain, "hostnames": hostname_entries} + if args.resolve or args.probe: + result["summary"] = summary + + if args.certificates: + try: + certificate_hashes = fetch_certificate_hashes(domain, args.timeout, args.retries) + certificates = [] + for certificate_hash in certificate_hashes: + try: + certificates.append( + {"sha256": certificate_hash, "data": fetch_certificate(certificate_hash, args.timeout, args.retries)} + ) + except (RuntimeError, ValueError) as exc: + print(f"[!] Could not fetch certificate {certificate_hash}: {exc}", file=sys.stderr) + failed = True + result["certificates"] = certificates + print(f"[+] Found {len(certificates)} related certificates", file=sys.stderr) + except RuntimeError as exc: + print(f"[!] Certificate enrichment failed for {domain}: {exc}", file=sys.stderr) + result["certificates"] = [] + failed = True + + results.append(result) + if args.resolve or args.probe: + active_label = colorize( + f"{active_count} active", + "1;32", + color_enabled(args.color and not args.no_color, sys.stderr), + ) + filtered_label = ( + f"; kept {len(hostname_entries)}" + if args.active_only + else "" + ) + print( + f"[+] Found {len(hostnames)} hostnames ({active_label}{filtered_label})", + file=sys.stderr, + ) + print( + "[+] Summary: " + f"{summary['dns']} DNS | {summary['tcp']} TCP | " + f"{summary['http']} HTTP | {summary['stale']} stale", + file=sys.stderr, + ) + else: + print(f"[+] Found {len(hostnames)} hostnames", file=sys.stderr) content = ( json.dumps(results, ensure_ascii=True, indent=2) + "\n" if args.format == "json" - else format_text(results, args.resolve) + else format_text( + results, + args.resolve, + include_status=args.status, + color=(args.color or color_enabled(False)) and not args.no_color, + pretty=(args.output is None and args.status and sys.stdout.isatty()), + width=shutil.get_terminal_size(fallback=(120, 24)).columns, + ) ) if args.output: try: diff --git a/tests/test_recons101x.py b/tests/test_recons101x.py index 3e7d5c2..a4485a5 100644 --- a/tests/test_recons101x.py +++ b/tests/test_recons101x.py @@ -12,7 +12,7 @@ class DomainTests(unittest.TestCase): def test_version(self): - self.assertEqual(recons101x.__version__, "1.0.0") + self.assertEqual(recons101x.__version__, "1.0.5") def test_normalizes_domain(self): self.assertEqual(recons101x.normalize_domain(" Example.COM. "), "example.com") @@ -59,6 +59,94 @@ def test_formats_text_with_ips(self): "example.com\twww.example.com\t1.2.3.4\n", ) + @mock.patch("recons101x.fetch_json") + def test_fetches_certificate_hashes_from_domain_endpoint(self, fetch_json): + fetch_json.return_value = [ + "A" * 64, + {"sha256": "b" * 64}, + "not-a-sha256", + "A" * 64, + ] + self.assertEqual( + recons101x.fetch_certificate_hashes("example.com", timeout=1, retries=0), + ["a" * 64, "b" * 64], + ) + + def test_formats_active_status_and_keeps_green_for_active_nodes(self): + results = [ + { + "domain": "example.com", + "hostnames": [ + {"hostname": "online.example.com", "ips": ["1.2.3.4"], "active": True}, + {"hostname": "offline.example.com", "ips": [], "active": False}, + ], + } + ] + output = recons101x.format_text(results, include_ips=True, include_status=True, color=True) + self.assertIn("\033[1;32mACTIVE\033[0m", output) + self.assertIn("INACTIVE", output) + + def test_pretty_output_is_fixed_width_and_truncates_long_values(self): + results = [ + { + "domain": "example.com", + "hostnames": [ + { + "hostname": "very-long-hostname-" + "x" * 100 + ".example.com", + "ips": ["2001:db8::1", "2001:db8::2"], + "state": "DNS_ONLY", + "active": True, + } + ], + } + ] + output = recons101x.format_text( + results, include_ips=True, include_status=True, pretty=True, width=80 + ) + self.assertIn("DOMAIN", output) + self.assertIn("DNS_ONLY", output) + self.assertIn("...", output) + self.assertLessEqual(max(map(len, output.rstrip().splitlines())), 80) + + def test_classifies_reachability_and_builds_summary(self): + entries = [ + recons101x.build_hostname_entry("stale.example.com", [], [], {}, False, True, True), + recons101x.build_hostname_entry("dns.example.com", ["1.2.3.4"], [], {}, True, True, True), + recons101x.build_hostname_entry("tcp.example.com", ["1.2.3.5"], [443], {}, True, True, True), + recons101x.build_hostname_entry( + "http.example.com", ["1.2.3.6"], [443], {"https": 200}, True, True, True + ), + ] + self.assertEqual( + [entry["state"] for entry in entries], + ["STALE", "DNS_ONLY", "TCP_REACHABLE", "HTTP_REACHABLE"], + ) + self.assertEqual( + recons101x.summarize_entries(entries), + {"total": 4, "dns": 3, "tcp": 2, "http": 1, "stale": 1}, + ) + + @mock.patch("recons101x.socket.create_connection") + def test_probe_hostname_reports_reachable_web_ports(self, create_connection): + connection = mock.MagicMock() + connection.__enter__.return_value = connection + create_connection.side_effect = [connection, OSError("closed")] + + self.assertEqual(recons101x.probe_hostname("online.example.com", timeout=1), [443]) + self.assertEqual( + [call.args for call in create_connection.call_args_list], + [(('online.example.com', 443),), (('online.example.com', 80),)], + ) + + def test_live_mode_is_a_single_high_level_switch(self): + args = recons101x.apply_mode( + recons101x.build_parser().parse_args(["example.com", "--live"]) + ) + self.assertTrue(args.resolve) + self.assertTrue(args.probe) + self.assertTrue(args.active_only) + self.assertTrue(args.status) + if __name__ == "__main__": unittest.main()