scry.py

#!/usr/bin/env python3
"""
scry — quick infrastructure fingerprinting for an SRE.

SCRY: Stack, CDN & Routing Yield-er — "scry the unseen infrastructure."

Given a URL or domain, it figures out (best-effort, from public signals):
  * DNS records (A/AAAA/NS/MX/TXT/CNAME/SOA) via `dig`
  * Hosting / DNS provider inferred from nameservers + IP whois
  * CDN / edge presence (CNAME patterns, IP ownership, response headers)
  * Web stack hints (Server, X-Powered-By, cookies, HTML signatures)
  * TLS certificate issuer + SANs (via openssl)
  * Email security posture (SPF / DMARC / MX provider)
  * Best-effort OS guess (ping TTL, Server-header tokens, optional nmap -O)

NOTE on OS detection: from the outside it is inherently unreliable. If the
site sits behind a CDN/load balancer you are fingerprinting the edge node,
not the origin. Treat every OS guess as low confidence unless corroborated.

Stdlib only. Shells out to `dig`, `whois`, `openssl`, `ping`, and `nmap`
(optional, for -O) if present.

Usage:
    python3 scry.py example.com
    python3 scry.py https://www.example.com --json
    python3 scry.py example.com --no-whois     # faster, skips whois
"""

import argparse
import json
import re
import shutil
import socket
import ssl
import subprocess
import sys
from urllib.parse import urlparse
from urllib.request import Request, urlopen

# ---------------------------------------------------------------------------
# Signature tables. Add to these as you learn new patterns.
# ---------------------------------------------------------------------------

# CNAME / hostname substrings -> provider name
CNAME_SIGNATURES = {
    "cloudfront.net": "AWS CloudFront (CDN)",
    "akamai": "Akamai (CDN)",
    "akamaiedge.net": "Akamai (CDN)",
    "akadns.net": "Akamai",
    "edgekey.net": "Akamai (CDN)",
    "edgesuite.net": "Akamai (CDN)",
    "fastly.net": "Fastly (CDN)",
    "cloudflare": "Cloudflare (CDN/WAF)",
    "cdn.cloudflare.net": "Cloudflare (CDN)",
    "azureedge.net": "Azure CDN",
    "azurefd.net": "Azure Front Door",
    "trafficmanager.net": "Azure Traffic Manager",
    "googleusercontent.com": "Google",
    "googlehosted.com": "Google",
    "ghs.google.com": "Google Hosting",
    "herokudns.com": "Heroku",
    "herokuapp.com": "Heroku",
    "netlify": "Netlify",
    "vercel-dns.com": "Vercel",
    "vercel.app": "Vercel",
    "github.io": "GitHub Pages",
    "gitlab.io": "GitLab Pages",
    "wpengine": "WP Engine",
    "pantheonsite.io": "Pantheon",
    "wixdns.net": "Wix",
    "squarespace.com": "Squarespace",
    "shopify": "Shopify",
    "myshopify.com": "Shopify",
    "incapdns.net": "Imperva Incapsula (CDN/WAF)",
    "stackpathdns.com": "StackPath (CDN)",
    "bunnycdn.com": "Bunny CDN",
    "b-cdn.net": "Bunny CDN",
    "kxcdn.com": "KeyCDN",
    "cdn77": "CDN77",
    "edgecastcdn.net": "Edgecast (CDN)",
    "llnwd.net": "Limelight (CDN)",
    "elasticbeanstalk.com": "AWS Elastic Beanstalk",
    "amazonaws.com": "AWS",
    "awsglobalaccelerator.com": "AWS Global Accelerator",
}

# Nameserver substrings -> DNS/hosting provider
NS_SIGNATURES = {
    "awsdns": "AWS Route 53",
    "cloudflare": "Cloudflare DNS",
    "azure-dns": "Azure DNS",
    "googledomains": "Google Domains",
    "google.com": "Google Cloud DNS",
    "domaincontrol.com": "GoDaddy",
    "dnsmadeeasy": "DNS Made Easy",
    "nsone.net": "NS1",
    "ultradns": "UltraDNS (Neustar)",
    "dyn.com": "Oracle Dyn",
    "akam.net": "Akamai DNS",
    "name-services.com": "eNom",
    "registrar-servers.com": "Namecheap",
    "digitalocean.com": "DigitalOcean DNS",
    "linode.com": "Linode/Akamai DNS",
    "vultr.com": "Vultr DNS",
    "hetzner": "Hetzner DNS",
}

# MX substrings -> email provider
MX_SIGNATURES = {
    "google.com": "Google Workspace",
    "googlemail.com": "Google Workspace",
    "outlook.com": "Microsoft 365",
    "protection.outlook.com": "Microsoft 365 / Exchange Online",
    "pphosted.com": "Proofpoint",
    "mimecast": "Mimecast",
    "messagelabs.com": "Symantec/Broadcom Email",
    "zoho": "Zoho Mail",
    "amazonaws.com": "Amazon SES/WorkMail",
    "mailgun": "Mailgun",
    "sendgrid": "SendGrid",
    "protonmail": "Proton Mail",
    "fastmail": "Fastmail",
}

# Response-header signals -> meaning
HEADER_SIGNATURES = {
    "server": {
        "cloudflare": "Cloudflare (CDN/WAF)",
        "awselb": "AWS Elastic Load Balancer",
        "nginx": "nginx",
        "apache": "Apache httpd",
        "microsoft-iis": "Microsoft IIS",
        "gunicorn": "Gunicorn (Python WSGI)",
        "openresty": "OpenResty (nginx+Lua)",
        "litespeed": "LiteSpeed",
        "envoy": "Envoy proxy",
        "gws": "Google Web Server",
        "ecacc": "Akamai (CDN)",
        "akamaighost": "Akamai (CDN)",
        "fastly": "Fastly (CDN)",
        "vercel": "Vercel",
        "netlify": "Netlify",
        "caddy": "Caddy",
    },
    "x-powered-by": {
        "php": "PHP",
        "express": "Node.js / Express",
        "asp.net": "ASP.NET",
        "next.js": "Next.js",
        "servlet": "Java Servlet",
        "plesk": "Plesk",
    },
}

# Header *presence* -> what it implies
HEADER_PRESENCE = {
    "cf-ray": "Cloudflare (CDN/WAF)",
    "cf-cache-status": "Cloudflare (CDN)",
    "x-amz-cf-id": "AWS CloudFront (CDN)",
    "x-amz-request-id": "AWS S3 / API",
    "x-served-by": "Fastly/Varnish edge",
    "x-cache": "CDN/edge cache present",
    "x-fastly-request-id": "Fastly (CDN)",
    "x-vercel-id": "Vercel",
    "x-nf-request-id": "Netlify",
    "x-github-request-id": "GitHub Pages",
    "x-azure-ref": "Azure Front Door",
    "x-akamai-transformed": "Akamai (CDN)",
    "x-drupal-cache": "Drupal",
    "x-generator": "see value (CMS/framework)",
    "x-shopify-stage": "Shopify",
    "fly-request-id": "Fly.io",
}

# Cookie name -> framework/CMS
COOKIE_SIGNATURES = {
    "wordpress": "WordPress",
    "wp-": "WordPress",
    "phpsessid": "PHP",
    "jsessionid": "Java",
    "asp.net": "ASP.NET",
    "laravel_session": "Laravel (PHP)",
    "_shopify": "Shopify",
    "ci_session": "CodeIgniter (PHP)",
    "django": "Django",
    "csrftoken": "Django",
    "rack.session": "Ruby Rack/Rails",
    "_rails": "Ruby on Rails",
}

# HTML body substrings -> CMS/framework
HTML_SIGNATURES = {
    "wp-content": "WordPress",
    "wp-includes": "WordPress",
    "/sites/default/files": "Drupal",
    "Drupal.settings": "Drupal",
    "data-drupal": "Drupal",
    "cdn.shopify.com": "Shopify",
    "Joomla!": "Joomla",
    "/_next/static": "Next.js (React)",
    "__NEXT_DATA__": "Next.js (React)",
    "window.__NUXT__": "Nuxt.js (Vue)",
    "ng-version": "Angular",
    "data-reactroot": "React",
    "wix.com": "Wix",
    "squarespace": "Squarespace",
    "gatsby": "Gatsby",
    "hubspot": "HubSpot",
    "static.parastorage.com": "Wix",
}


# Common subdomain prefixes that frequently expose the true origin behind a
# CDN. The CDN must pull from a real backend, and teams leave a directly
# resolving record for it. Ordered roughly by hit-rate.
ORIGIN_PREFIXES = [
    "origin", "origin-www", "origin-api", "origin-app", "www-origin",
    "direct", "direct-connect", "real", "backend", "back-end",
    "www2", "www1", "web", "web1", "web2", "app", "api", "secure",
    "cpanel", "whm", "webmail", "mail", "smtp", "ftp", "ssh", "vpn",
    "remote", "gateway", "gw", "edge", "lb",
    "dev", "staging", "stage", "test", "uat", "qa", "preprod",
    "old", "legacy", "portal", "admin", "internal", "intranet", "corp",
    "m", "mobile", "cdn-origin", "ords", "host",
]

# IP/owner substrings that mean "still the CDN, not the origin".
CDN_OWNER_NEEDLES = [
    "cloudflare", "akamai", "fastly", "incapsula", "imperva", "edgecast",
    "stackpath", "sucuri", "azure front", "cloudfront", "limelight",
    "section.io", "bunny", "keycdn", "g-core", "qrator",
]


def have(binary):
    return shutil.which(binary) is not None


def run(cmd, timeout=15):
    try:
        out = subprocess.run(
            cmd, capture_output=True, text=True, timeout=timeout
        )
        return out.stdout.strip()
    except Exception:
        return ""


def normalize(target):
    """Return (hostname, scheme_url)."""
    if "://" not in target:
        target = "https://" + target
    p = urlparse(target)
    host = p.hostname or target
    return host, target


def dig(name, rtype):
    if not have("dig"):
        return []
    out = run(["dig", "+short", name, rtype])
    return [line for line in out.splitlines() if line.strip()]


def gather_dns(host):
    """Resolve common records. Strip a leading www to also probe apex."""
    apex = host[4:] if host.startswith("www.") else host
    records = {}
    for rtype in ("A", "AAAA", "CNAME", "NS", "MX", "TXT", "SOA"):
        # NS/SOA/MX/TXT belong to the apex; A/AAAA/CNAME to the host as given
        query = apex if rtype in ("NS", "SOA", "MX", "TXT") else host
        records[rtype] = dig(query, rtype)
    # Fall back to apex A records if host had none
    if not records["A"] and host != apex:
        records["A"] = dig(apex, "A")
    return records, apex


def match_table(value, table):
    v = value.lower()
    hits = []
    for needle, label in table.items():
        if needle in v:
            hits.append(label)
    return hits


_ASN_CACHE = {}


def asn_lookup(ip):
    """Resolve an IP to ASN + network name + BGP prefix via Team Cymru's
    whois service. Passive (no packets to the target) and needs no install —
    it's just a whois query to a public reflector. Returns a dict or {}.
    """
    if ip in _ASN_CACHE:
        return _ASN_CACHE[ip]
    result = {}
    if have("whois"):
        out = run(["whois", "-h", "whois.cymru.com", f" -v {ip}"], timeout=15)
        # Skip the header line; parse the data row.
        for line in out.splitlines():
            if "|" in line and "AS Name" not in line and "BGP Prefix" not in line:
                cols = [c.strip() for c in line.split("|")]
                if len(cols) >= 7 and cols[0].isdigit():
                    result = {"asn": cols[0], "prefix": cols[2],
                              "cc": cols[3], "name": cols[6]}
                    break
    _ASN_CACHE[ip] = result
    return result


def ip_owner(ip, do_whois=True):
    if not do_whois or not have("whois"):
        try:
            return socket.gethostbyaddr(ip)[0]
        except Exception:
            return ""
    out = run(["whois", ip], timeout=20)
    for key in ("OrgName", "org-name", "owner", "netname", "Organization", "descr"):
        m = re.search(rf"^{key}:\s*(.+)$", out, re.IGNORECASE | re.MULTILINE)
        if m:
            return m.group(1).strip()
    return ""


UA = "Mozilla/5.0 (compatible; scry/1.0; +SRE-tool)"

# Headers an SRE actually cares about, for the condensed summary.
NOTABLE_HEADERS = [
    "server", "x-powered-by", "via", "x-cache", "cf-ray", "cf-cache-status",
    "x-served-by", "x-amz-cf-id", "x-vercel-id", "x-nf-request-id",
    "x-azure-ref", "strict-transport-security", "content-security-policy",
    "x-frame-options", "x-content-type-options", "x-generator",
    "x-aspnet-version", "x-runtime",
]


def curl_headers(url):
    """Grab response headers with curl, following redirects. Returns
    (final_headers_dict, raw_header_text, redirect_chain). Falls back to {}.

    curl is preferred over urllib here because it preserves every Set-Cookie,
    shows the full redirect chain, and is the tool SREs already trust.
    """
    if not have("curl"):
        return {}, "", []
    raw = run(["curl", "-sSL", "-D", "-", "-o", "/dev/null",
               "-A", UA, "--max-time", "15", url], timeout=20)
    if not raw:
        return {}, "", []
    # Split into per-response blocks (one per redirect hop).
    blocks = re.split(r"\r?\n\r?\n", raw)
    chain, last = [], {}
    for block in blocks:
        lines = block.splitlines()
        if not lines or not lines[0].startswith("HTTP/"):
            continue
        last = {}
        for line in lines[1:]:
            if ":" in line:
                k, v = line.split(":", 1)
                last[k.strip().lower()] = v.strip()
        status = lines[0].strip()
        loc = last.get("location")
        chain.append(f"{status} → {loc}" if loc else status)
    return last, raw, chain


def fetch_http(url):
    """Return (final_url, status, headers_dict, body_snippet)."""
    headers = {
        "User-Agent": "Mozilla/5.0 (compatible; scry/1.0; +SRE-tool)"
    }
    ctx = ssl.create_default_context()
    try:
        req = Request(url, headers=headers)
        with urlopen(req, timeout=15, context=ctx) as resp:
            raw = resp.read(200_000)
            body = raw.decode("utf-8", errors="replace")
            hdrs = {k.lower(): v for k, v in resp.headers.items()}
            return resp.geturl(), resp.status, hdrs, body
    except Exception as e:
        return url, None, {"_error": str(e)}, ""


def tls_cert(host, port=443):
    if not have("openssl"):
        return {}
    cmd = ["openssl", "s_client", "-connect", f"{host}:{port}",
           "-servername", host]
    try:
        proc = subprocess.run(cmd, input="", capture_output=True,
                              text=True, timeout=15)
        raw = proc.stdout
    except Exception:
        return {}
    info = {}
    m = re.search(r"issuer=(.+)", raw)
    if m:
        info["issuer"] = m.group(1).strip()
    m = re.search(r"subject=(.+)", raw)
    if m:
        info["subject"] = m.group(1).strip()
    # SANs via x509 text
    try:
        cert = re.search(r"-----BEGIN CERTIFICATE-----.+?-----END CERTIFICATE-----",
                         raw, re.DOTALL)
        if cert:
            x = subprocess.run(["openssl", "x509", "-noout", "-text"],
                               input=cert.group(0), capture_output=True,
                               text=True, timeout=10).stdout
            sans = re.search(r"X509v3 Subject Alternative Name:\s*\n\s*(.+)", x)
            if sans:
                info["sans"] = [s.strip().replace("DNS:", "")
                                for s in sans.group(1).split(",")][:25]
            dates = re.search(r"Not After : (.+)", x)
            if dates:
                info["expires"] = dates.group(1).strip()
    except Exception:
        pass
    return info


# OS-token substrings that distros/ports leak in the Server header
SERVER_OS_TOKENS = {
    "ubuntu": "Ubuntu Linux",
    "debian": "Debian Linux",
    "centos": "CentOS Linux",
    "red hat": "Red Hat Enterprise Linux",
    "rhel": "Red Hat Enterprise Linux",
    "fedora": "Fedora Linux",
    "amazon": "Amazon Linux",
    "alpine": "Alpine Linux",
    "suse": "SUSE Linux",
    "win64": "Windows",
    "win32": "Windows",
    "windows": "Windows",
    "(unix)": "Unix",
    "freebsd": "FreeBSD",
    "darwin": "macOS/Darwin",
}


def ttl_to_os(ttl):
    """Map an observed TTL back to a likely initial TTL / OS family.

    Hosts decrement TTL by 1 per hop, so the observed value is a bit below
    the initial. We bucket to the nearest common initial TTL.
    """
    if ttl is None:
        return None, None
    if ttl <= 64:
        return 64, "Linux / Unix / macOS / BSD (initial TTL 64)"
    if ttl <= 128:
        return 128, "Windows (initial TTL 128)"
    return 255, "network gear / Solaris / Cisco (initial TTL 255)"


def ping_ttl(host):
    """Return observed TTL from a single ping, or None."""
    if not have("ping"):
        return None
    # macOS/BSD: -c count -t timeout(sec) ; Linux: -c -W. Try BSD form first.
    for cmd in (["ping", "-c", "1", "-t", "5", host],
                ["ping", "-c", "1", "-W", "5", host]):
        out = run(cmd, timeout=8)
        m = re.search(r"ttl[=\s](\d+)", out, re.IGNORECASE)
        if m:
            return int(m.group(1))
    return None


def nmap_os(host):
    """Real stack fingerprint via nmap -O. Needs root + open ports + nmap."""
    if not have("nmap"):
        return None
    out = run(["nmap", "-O", "-Pn", "--osscan-guess", host], timeout=120)
    guesses = re.findall(r"(?:OS details|Running|Aggressive OS guesses):\s*(.+)",
                         out)
    if "requires root privileges" in out.lower():
        return "nmap -O requires root (run with sudo)"
    return guesses[0].strip() if guesses else None


def grab_ssh_banner(host, port=22, timeout=4):
    """Open TCP/22 and read the SSH identification string the server sends
    first. NOT passive — it's a real connection. The banner frequently names
    the distro, e.g. 'SSH-2.0-OpenSSH_8.9p1 Ubuntu-3ubuntu0.1'.

    Returns (banner, os_guess) or (None, None).
    """
    try:
        with socket.create_connection((host, port), timeout=timeout) as s:
            s.settimeout(timeout)
            banner = s.recv(256).decode("utf-8", errors="replace").strip()
    except Exception:
        return None, None
    if not banner.startswith("SSH-"):
        return banner or None, None
    # Pull a distro/OS hint out of the comment field after the version.
    b = banner.lower()
    for token, label in SERVER_OS_TOKENS.items():
        if token in b:
            return banner, label
    # OpenSSH without a distro tag still implies a Unix-like host.
    if "openssh" in b:
        return banner, "Unix-like (OpenSSH, no distro tag)"
    if "windows" in b or "for_windows" in b:
        return banner, "Windows"
    return banner, None


def os_signals_for(target, label_prefix, behind_cdn, use_nmap, check_ssh,
                   server_header=""):
    """Collect OS signals for a single host/IP: Server header, SSH banner,
    ping TTL, optional nmap. Returns (signals, ssh_banner).

    `label_prefix` distinguishes edge vs origin in the source string. Probing
    an origin directly is far more meaningful than the CDN edge, so origin
    signals are not confidence-penalised for the CDN.
    """
    signals, ssh_banner = [], None
    penalise = behind_cdn and label_prefix == "edge"

    if check_ssh:
        banner, ssh_os = grab_ssh_banner(target)
        if banner:
            ssh_banner = banner
            if ssh_os:
                signals.append({"source": f"{label_prefix} SSH banner",
                                "guess": ssh_os, "confidence": "high"})

    if server_header:
        for token, lbl in SERVER_OS_TOKENS.items():
            if token in server_header.lower():
                signals.append({"source": f"{label_prefix} Server header",
                                "guess": lbl, "confidence": "medium"})

    ttl = ping_ttl(target)
    if ttl is not None:
        _, lbl = ttl_to_os(ttl)
        signals.append({"source": f"{label_prefix} ping TTL={ttl}",
                        "guess": lbl,
                        "confidence": "low" if penalise else "medium"})

    if use_nmap:
        n = nmap_os(target)
        if n:
            signals.append({"source": f"{label_prefix} nmap -O",
                            "guess": n, "confidence": "high"})

    return signals, ssh_banner


def detect_os(host, server_header, behind_cdn, use_nmap=False, check_ssh=True,
              origins=None):
    """Fingerprint the OS. Probes the public host (edge) and, more usefully,
    any discovered web origins directly — the origin reflects the real backend
    OS rather than the CDN edge node.
    """
    os_info = {"signals": [], "behind_cdn": behind_cdn}

    # Edge / public host.
    sigs, banner = os_signals_for(host, "edge", behind_cdn, use_nmap,
                                  check_ssh, server_header)
    os_info["signals"].extend(sigs)
    if banner:
        os_info["ssh_banner"] = banner

    # Discovered web origins — the real prize. Probe their IPs directly.
    seen_ips = set()
    for o in (origins or []):
        if not (o.get("likely_origin") and o.get("category") == "web-origin"):
            continue
        for ip in o.get("ips", [])[:1]:
            if ip in seen_ips:
                continue
            seen_ips.add(ip)
            tag = f"origin {o['host']}"
            sigs, banner = os_signals_for(ip, tag, behind_cdn, use_nmap,
                                          check_ssh)
            os_info["signals"].extend(sigs)
            if banner and "ssh_banner" not in os_info:
                os_info["ssh_banner"] = f"{o['host']}: {banner}"

    return os_info


def crtsh_subdomains(apex, limit=60):
    """Mine subdomains from Certificate Transparency logs via crt.sh.

    CT logs record every cert ever issued, so they expose hostnames that
    never appear in normal DNS browsing — including origins, staging, and
    internal hosts whose certs leaked into the public logs. Passive: we query
    crt.sh's DB, not the target.
    """
    if not have("curl"):
        return []
    url = f"https://crt.sh/?q=%25.{apex}&output=json"
    raw = run(["curl", "-sS", "-A", UA, "--max-time", "25", url], timeout=30)
    if not raw:
        return []
    try:
        data = json.loads(raw)
    except Exception:
        return []
    names = set()
    for entry in data:
        for field in ("name_value", "common_name"):
            val = entry.get(field, "")
            for n in str(val).splitlines():
                n = n.strip().lower().lstrip("*.")
                if n.endswith(apex) and n != apex and "@" not in n:
                    names.add(n)
    return sorted(names)[:limit]


def check_origin_exposure(public_host, origin_ip, scheme="https"):
    """Connect straight to the origin IP while presenting the real hostname
    (correct SNI + Host via curl --resolve). If it answers 200 with real
    content, the origin is reachable directly — i.e. the CDN/WAF can be
    bypassed. ACTIVE: this opens a connection to the origin.
    """
    if not have("curl"):
        return None
    port = "443" if scheme == "https" else "80"
    out = run(["curl", "-sS", "-k", "-o", "/dev/null", "-A", UA,
               "--max-time", "12",
               "--resolve", f"{public_host}:{port}:{origin_ip}",
               "-w", "%{http_code} %{size_download} %{ssl_verify_result}",
               f"{scheme}://{public_host}/"], timeout=15)
    parts = out.split()
    if len(parts) < 2 or not parts[0].isdigit():
        return {"reachable": False, "raw": out}
    code, size = int(parts[0]), int(parts[1])
    # HTTP 0 = curl never got a response (refused/timeout/TLS reset) -> the
    # origin is firewalled to the CDN, the good outcome.
    if code == 0:
        return {"reachable": False, "status": 0}
    # 2xx/3xx with a real body served straight off the origin = exposed.
    exposed = code in (200, 301, 302) and size > 256
    return {"reachable": True, "status": code, "bytes": size,
            "exposed": exposed}


def detect_waf(url):
    """Identify a WAF with wafw00f if available. Active (sends probes)."""
    if not have("wafw00f"):
        return None
    out = run(["wafw00f", "-a", url], timeout=60)
    wafs = []
    for line in out.splitlines():
        m = re.search(r"is behind (.+?)(?: WAF| \(|$)", line)
        if m and "No WAF" not in line:
            wafs.append(m.group(1).strip())
    if "No WAF detected" in out or "seems to be behind a WAF" not in out \
            and not wafs:
        # fall through; wafs may still be empty
        pass
    return wafs or None


def tls_posture(host):
    """Richer TLS analysis via testssl.sh or sslscan if present. Active.
    Returns a short list of headline findings, not the full dump.
    """
    findings = []
    if have("testssl.sh"):
        out = run(["testssl.sh", "--quiet", "--color", "0",
                   "--protocols", "--vulnerable", f"{host}:443"], timeout=180)
        for proto in ("SSLv2", "SSLv3", "TLS 1 ", "TLS 1.1"):
            if re.search(rf"{re.escape(proto)}.*offered", out) and \
                    not re.search(rf"{re.escape(proto)}.*not offered", out):
                findings.append(f"legacy {proto.strip()} offered")
        for vuln in ("Heartbleed", "ROBOT", "POODLE", "BEAST", "FREAK",
                     "LOGJAM", "CCS", "Ticketbleed"):
            if re.search(rf"{vuln}.*(VULNERABLE|potentially)", out, re.I):
                findings.append(f"VULNERABLE: {vuln}")
    elif have("sslscan"):
        out = run(["sslscan", "--no-colour", host], timeout=60)
        for proto in ("SSLv2", "SSLv3", "TLSv1.0", "TLSv1.1"):
            if re.search(rf"{re.escape(proto)}\s+enabled", out):
                findings.append(f"legacy {proto} enabled")
    return findings or None


def trace_path(target, max_hops=20):
    """traceroute to a target, returning a compact list of hops with the
    ASN/network for each (Team Cymru). Most useful pointed at an *origin* IP:
    against a CDN edge it mostly reflects your own ISP's transit.
    """
    if not have("traceroute"):
        return None
    out = run(["traceroute", "-n", "-w", "2", "-q", "1", "-m",
               str(max_hops), target], timeout=60)
    hops = []
    for line in out.splitlines()[1:]:
        m = re.match(r"\s*(\d+)\s+(\d+\.\d+\.\d+\.\d+)", line)
        if m:
            ip = m.group(2)
            asn = asn_lookup(ip)
            net = asn.get("name", "").split(" - ")[0] if asn else ""
            hops.append({"hop": m.group(1), "ip": ip, "net": net})
        elif re.match(r"\s*\d+\s+\*", line):
            hops.append({"hop": line.split()[0], "ip": "*", "net": ""})
    return hops or None


def looks_like_cdn(cname_chain, owner, ips=None):
    blob = (" ".join(cname_chain) + " " + (owner or "")).lower()
    if any(n in blob for n in CDN_OWNER_NEEDLES) or \
            any(n in blob for n in CNAME_SIGNATURES):
        return True
    # PTR catches CDNs whose whois owner is generic (e.g. CloudFront IPs are
    # owned by "Amazon" but reverse-resolve to *.cloudfront.net).
    for ip in (ips or [])[:1]:
        ptr = " ".join(dig_x(ip)).lower()
        if any(n in ptr for n in CNAME_SIGNATURES) or \
                any(n in ptr for n in CDN_OWNER_NEEDLES):
            return True
    return False


def dig_x(ip):
    if not have("dig"):
        return []
    return [l for l in run(["dig", "+short", "-x", ip]).splitlines() if l]


# Map the leftmost DNS label to a category so web origins aren't buried under
# corporate/mail/VPN edge. First matching group wins.
CATEGORY_RULES = [
    ("web-origin", ("origin", "www", "web", "secure", "edge", "app", "api",
                    "direct", "backend", "back-end", "real", "cdn-origin",
                    "lb", "host", "ords")),
    ("remote-access", ("vpn", "remote", "gateway", "gw", "ssh")),
    ("mail/ftp", ("mail", "smtp", "webmail", "ftp", "imap", "pop", "mx")),
    ("non-prod", ("dev", "staging", "stage", "test", "uat", "qa", "preprod",
                  "sandbox", "demo")),
    ("admin/internal", ("admin", "portal", "internal", "intranet", "corp",
                        "old", "legacy", "vpc", "cpanel", "whm")),
]


def categorize(fqdn, apex):
    label = fqdn[:-(len(apex) + 1)] if fqdn.endswith("." + apex) else fqdn
    first = label.split(".")[0]  # leftmost label, e.g. "origin" in origin-www
    for cat, prefixes in CATEGORY_RULES:
        for p in prefixes:
            if first == p or first.startswith(p):
                return cat
    return "other"


def discover_origins(apex, edge_ips, do_whois=True, extra_hosts=None,
                     check_exposure=False, public_host=None):
    """Probe origin-revealing subdomains (static prefix list + any extra hosts
    mined from CT logs). A candidate is a likely origin when it resolves to an
    IP that (a) isn't one of the CDN edge IPs and (b) isn't owned by a CDN.
    Each hit is categorized (web-origin vs remote-access/mail/etc.), and
    likely web origins are optionally probed for direct exposure.
    """
    edge = set(edge_ips)
    candidates = {f"{p}.{apex}" for p in ORIGIN_PREFIXES}
    candidates.update(extra_hosts or [])
    results = []
    seen = set()
    for fqdn in candidates:
        if fqdn in seen:
            continue
        seen.add(fqdn)
        cnames = dig(fqdn, "CNAME")
        a = [ip for ip in dig(fqdn, "A")
             if re.match(r"^\d+\.\d+\.\d+\.\d+$", ip)]
        if not a:
            continue
        owner = ip_owner(a[0], do_whois)
        is_cdn = looks_like_cdn(cnames, owner, a)
        likely_origin = bool(set(a) - edge) and not is_cdn
        cat = categorize(fqdn, apex)
        entry = {
            "host": fqdn, "ips": a, "owner": owner, "cname": cnames,
            "likely_origin": likely_origin, "category": cat,
            "asn": asn_lookup(a[0]),
        }
        # Exposure check only for likely web origins, and only on request.
        if (check_exposure and likely_origin and cat == "web-origin"
                and public_host):
            entry["exposure"] = check_origin_exposure(public_host, a[0])
        results.append(entry)
    results.sort(key=lambda r: (not r["likely_origin"], r["category"],
                                r["host"]))
    return results


def analyze(host, do_whois=True, use_nmap=False, check_ssh=True, url=None,
            find_origin=False, use_crtsh=True, check_exposure=False,
            trace=False):
    findings = {
        "target": host,
        "dns": {},
        "providers": {"dns": set(), "cdn": set(), "hosting": set(),
                      "email": set(), "stack": set()},
        "ips": [],
        "http": {},
        "tls": {},
        "email_security": {},
    }

    dns_records, apex = gather_dns(host)
    findings["dns"] = dns_records
    findings["apex"] = apex

    # Nameservers -> DNS provider
    for ns in dns_records.get("NS", []):
        findings["providers"]["dns"].update(match_table(ns, NS_SIGNATURES))

    # CNAME chain -> CDN/hosting
    for cn in dns_records.get("CNAME", []):
        findings["providers"]["cdn"].update(match_table(cn, CNAME_SIGNATURES))

    # MX -> email
    for mx in dns_records.get("MX", []):
        findings["providers"]["email"].update(match_table(mx, MX_SIGNATURES))

    # IP ownership -> hosting/CDN
    for ip in dns_records.get("A", [])[:4]:
        if re.match(r"^\d+\.\d+\.\d+\.\d+$", ip):
            owner = ip_owner(ip, do_whois)
            findings["ips"].append({"ip": ip, "owner": owner,
                                    "asn": asn_lookup(ip)})
            # crude: owner string often reveals AWS/Google/Cloudflare/etc.
            for needle, label in {
                "amazon": "AWS", "google": "Google Cloud",
                "cloudflare": "Cloudflare", "microsoft": "Azure",
                "fastly": "Fastly (CDN)", "akamai": "Akamai (CDN)",
                "digitalocean": "DigitalOcean", "linode": "Linode/Akamai",
                "hetzner": "Hetzner", "ovh": "OVH", "vultr": "Vultr",
                "incapsula": "Imperva Incapsula", "godaddy": "GoDaddy",
            }.items():
                if needle in owner.lower():
                    findings["providers"]["hosting"].add(label)

    # Email security from TXT
    txt = " ".join(dns_records.get("TXT", []))
    findings["email_security"]["spf"] = bool(re.search(r"v=spf1", txt, re.I))
    dmarc = dig("_dmarc." + apex, "TXT")
    findings["email_security"]["dmarc"] = bool(
        any("v=DMARC1" in d for d in dmarc))
    findings["email_security"]["dmarc_record"] = dmarc

    # HTTP fetch + header/body fingerprinting.
    # curl gives the best headers (all Set-Cookies, redirect chain); urllib
    # still supplies the HTML body for content fingerprinting.
    fetch_url = url or ("https://" + host)
    final_url, status, hdrs, body = fetch_http(fetch_url)
    curl_hdrs, _, redirect_chain = curl_headers(fetch_url)
    if curl_hdrs:
        merged = dict(hdrs)
        merged.update(curl_hdrs)  # curl wins on conflicts
        hdrs = merged
    notable = {h: hdrs[h] for h in NOTABLE_HEADERS if h in hdrs}
    findings["http"] = {"final_url": final_url, "status": status,
                        "server": hdrs.get("server", ""),
                        "x_powered_by": hdrs.get("x-powered-by", ""),
                        "redirect_chain": redirect_chain,
                        "notable_headers": notable,
                        "error": hdrs.get("_error", "")}

    for hname, table in HEADER_SIGNATURES.items():
        val = hdrs.get(hname, "")
        if val:
            for label in match_table(val, table):
                bucket = "cdn" if "CDN" in label else "stack"
                findings["providers"][bucket].add(label)

    for hname, label in HEADER_PRESENCE.items():
        if hname in hdrs:
            bucket = "cdn" if "CDN" in label or "edge" in label.lower() else "stack"
            findings["providers"][bucket].add(f"{label} [{hname}]")

    cookies = hdrs.get("set-cookie", "")
    for label in match_table(cookies, COOKIE_SIGNATURES):
        findings["providers"]["stack"].add(label)

    for label in match_table(body, HTML_SIGNATURES):
        findings["providers"]["stack"].add(label)

    gen = re.search(r'<meta[^>]+name=["\']generator["\'][^>]+content=["\']([^"\']+)',
                    body, re.I)
    if gen:
        findings["providers"]["stack"].add("generator: " + gen.group(1))

    findings["tls"] = tls_cert(host)
    findings["tls_posture"] = tls_posture(host)  # testssl.sh/sslscan if present
    findings["waf"] = detect_waf("https://" + host)  # wafw00f if present

    # Origin discovery: most valuable when the site is CDN-fronted. Runs
    # automatically behind a CDN, or on demand via --find-origin.
    behind_cdn = bool(findings["providers"]["cdn"])
    edge_ips = dns_records.get("A", [])
    if find_origin or behind_cdn:
        extra = []
        if use_crtsh:
            # CT-log subdomains, plus this cert's own SANs if we grabbed them.
            extra = crtsh_subdomains(apex)
            extra += [s for s in findings.get("tls", {}).get("sans", [])
                      if s.endswith(apex)]
        findings["crtsh_count"] = len(set(extra))
        findings["origins"] = discover_origins(
            apex, edge_ips, do_whois, extra_hosts=extra,
            check_exposure=check_exposure, public_host=host)
    else:
        findings["origins"] = []

    # Traceroute (opt-in). Point it at a discovered origin if we have one —
    # tracing the CDN edge mostly reveals your own ISP's transit.
    findings["trace"] = None
    if trace:
        web = [o for o in findings["origins"]
               if o.get("likely_origin") and o.get("category") == "web-origin"]
        tgt_ip = web[0]["ips"][0] if web else host
        tgt_label = web[0]["host"] if web else host
        findings["trace"] = {"target": tgt_label,
                             "hops": trace_path(tgt_ip)}

    # OS detection (best-effort). Flag low confidence if a CDN is in play.
    findings["os"] = detect_os(host, hdrs.get("server", ""),
                               behind_cdn, use_nmap=use_nmap,
                               check_ssh=check_ssh,
                               origins=findings["origins"])

    # Convert sets to sorted lists for output
    for k in findings["providers"]:
        findings["providers"][k] = sorted(findings["providers"][k])
    return findings


# ---------------------------------------------------------------------------
# Output
# ---------------------------------------------------------------------------

C = {
    "h": "\033[1;36m", "k": "\033[1;33m", "g": "\033[32m",
    "d": "\033[2m", "r": "\033[31m", "x": "\033[0m",
}


def color(enabled):
    return C if enabled else {k: "" for k in C}


def section(title, c):
    print(f"\n{c['h']}== {title} =={c['x']}")


def print_report(f, use_color=True):
    c = color(use_color and sys.stdout.isatty())
    print(f"\n{c['h']}╔══ scry: {f['target']} ══╗{c['x']}")

    section("Inferred providers", c)
    labels = [("DNS", "dns"), ("CDN / edge", "cdn"), ("Hosting", "hosting"),
              ("Email", "email"), ("Web stack / CMS", "stack")]
    for name, key in labels:
        vals = f["providers"][key]
        if vals:
            print(f"  {c['k']}{name:16}{c['x']} {c['g']}" +
                  ", ".join(vals) + c['x'])
        else:
            print(f"  {c['k']}{name:16}{c['x']} {c['d']}—{c['x']}")
    waf = f.get("waf")
    if waf:
        print(f"  {c['k']}{'WAF':16}{c['x']} {c['g']}" +
              ", ".join(waf) + c['x'])

    section("DNS records", c)
    for rtype, vals in f["dns"].items():
        if vals:
            print(f"  {c['k']}{rtype:6}{c['x']} " +
                  ("\n         ".join(vals) if len(vals) > 1 else vals[0]))

    if f["ips"]:
        section("IP ownership", c)
        for ip in f["ips"]:
            asn = ip.get("asn") or {}
            asn_str = (f"AS{asn['asn']} {asn['name']} [{asn['prefix']} "
                       f"{asn['cc']}]") if asn else ip.get("owner", "")
            print(f"  {ip['ip']:18} {c['d']}{asn_str}{c['x']}")

    section("HTTP", c)
    h = f["http"]
    if h.get("error"):
        print(f"  {c['r']}error: {h['error']}{c['x']}")
    else:
        print(f"  status     {h['status']}")
        print(f"  final url  {h['final_url']}")
        chain = h.get("redirect_chain") or []
        if len(chain) > 1:
            print(f"  {c['k']}redirect chain{c['x']}")
            for hop in chain:
                print(f"    {c['d']}{hop}{c['x']}")
        notable = h.get("notable_headers") or {}
        if notable:
            print(f"  {c['k']}notable headers{c['x']}")
            for k, v in notable.items():
                if len(v) > 110:
                    v = v[:107] + "…"
                print(f"    {c['d']}{k}:{c['x']} {v}")

    if f["tls"]:
        section("TLS certificate", c)
        for k in ("subject", "issuer", "expires"):
            if f["tls"].get(k):
                print(f"  {c['k']}{k:8}{c['x']} {f['tls'][k]}")
        if f["tls"].get("sans"):
            print(f"  {c['k']}sans{c['x']}     " +
                  ", ".join(f["tls"]["sans"]))
        posture = f.get("tls_posture")
        if posture:
            for p in posture:
                col = c['r'] if p.startswith("VULNERABLE") else c['k']
                print(f"  {col}{p}{c['x']}")

    origins = f.get("origins") or []
    if origins:
        section("Origin discovery (behind CDN)", c)
        if f.get("crtsh_count"):
            print(f"  {c['d']}({f['crtsh_count']} extra hostnames mined from "
                  f"CT logs / cert SANs){c['x']}")
        likely = [o for o in origins if o["likely_origin"]]

        # Web origins first — the storefront-bypass targets.
        web = [o for o in likely if o["category"] == "web-origin"]
        if web:
            print(f"\n  {c['k']}▸ Likely WEB ORIGINS "
                  f"(direct backend, CDN-bypass risk){c['x']}")
            for o in web:
                asn = o.get("asn") or {}
                tag = (f"AS{asn['asn']} {asn['name'].split(' - ')[0]}"
                       if asn else o["owner"])
                line = (f"  {c['g']}● {o['host']:30}{c['x']} "
                        f"{', '.join(o['ips'])}  {c['d']}{tag}{c['x']}")
                print(line)
                ex = o.get("exposure")
                if ex and ex.get("reachable"):
                    if ex.get("exposed"):
                        print(f"      {c['r']}⚠ EXPOSED: origin served the "
                              f"site directly (HTTP {ex['status']}, "
                              f"{ex['bytes']} bytes) — WAF/CDN bypassable"
                              f"{c['x']}")
                    else:
                        print(f"      {c['d']}origin reachable but did not "
                              f"serve the site (HTTP {ex['status']})"
                              f"{c['x']}")
                elif ex:
                    print(f"      {c['d']}origin not directly reachable"
                          f"{c['x']}")

        # Everything else off-CDN, grouped by category.
        infra = [o for o in likely if o["category"] != "web-origin"]
        if infra:
            print(f"\n  {c['k']}▸ Other EXPOSED INFRASTRUCTURE "
                  f"(off-CDN, not the storefront){c['x']}")
            for cat in ("remote-access", "mail/ftp", "non-prod",
                        "admin/internal", "other"):
                group = [o for o in infra if o["category"] == cat]
                for o in group:
                    print(f"  {c['k']}● {o['host']:30}{c['x']} "
                          f"{', '.join(o['ips'])}  "
                          f"{c['d']}[{cat}] {o['owner']}{c['x']}")

        other = [o for o in origins if not o["likely_origin"]]
        if other:
            print(f"\n  {c['d']}○ on CDN/edge (not origins): " +
                  ", ".join(o["host"].split(".")[0] for o in other) + c['x'])
        if not likely:
            print(f"  {c['d']}no obvious off-CDN origin found.{c['x']}")

    tr = f.get("trace")
    if tr and tr.get("hops"):
        section(f"Transit / backbone networks → {tr['target']}", c)
        # Collapse hops into an ordered, de-duplicated list of networks.
        backbones, seen = [], set()
        for h in tr["hops"]:
            net = h.get("net")
            if net and net not in seen:
                seen.add(net)
                backbones.append(net)
        if backbones:
            print(f"  {c['d']}(roles are vantage-dependent: run from another "
                  f"location and the your-ISP end will differ){c['x']}")
            n = len(backbones)
            for i, net in enumerate(backbones, 1):
                if i == 1 and n > 1:
                    role, col = "your ISP", c['d']
                elif i == n:
                    role, col = "target-side", c['g']
                else:
                    role, col = "transit backbone", c['k']
                print(f"  {col}{i}. {net:42}{c['x']} "
                      f"{c['d']}[{role}]{c['x']}")
        else:
            print(f"  {c['d']}no networks resolved (path filtered){c['x']}")

    osd = f.get("os", {})
    if osd:
        section("Operating system (best-effort)", c)
        sigs = osd.get("signals", [])
        edge_sigs = [s for s in sigs if s["source"].startswith("edge")]
        origin_sigs = [s for s in sigs if s["source"].startswith("origin")]
        if not sigs:
            print(f"  {c['d']}no usable signals{c['x']}")

        def emit(s):
            conf = s["confidence"]
            col = {"high": c['g'], "medium": c['k'],
                   "low": c['d']}.get(conf, "")
            print(f"    {col}{s['guess']}{c['x']} "
                  f"{c['d']}({s['source']}, {conf} confidence){c['x']}")

        if origin_sigs:
            print(f"  {c['k']}▸ origin backend (probed directly){c['x']}")
            for s in origin_sigs:
                emit(s)
        if edge_sigs:
            note = " — CDN edge, not the real backend" \
                if osd.get("behind_cdn") else ""
            print(f"  {c['k']}▸ public host{note}{c['x']}")
            for s in edge_sigs:
                emit(s)
        if osd.get("ssh_banner"):
            print(f"  {c['d']}ssh: {osd['ssh_banner']}{c['x']}")

    section("Email security", c)
    es = f["email_security"]
    spf = f"{c['g']}present{c['x']}" if es["spf"] else f"{c['r']}missing{c['x']}"
    dm = f"{c['g']}present{c['x']}" if es["dmarc"] else f"{c['r']}missing{c['x']}"
    print(f"  SPF   {spf}")
    print(f"  DMARC {dm}")
    print()


def main():
    ap = argparse.ArgumentParser(
        prog="scry",
        description="scry — fingerprint a site's infrastructure from public "
                    "signals (Stack, CDN & Routing Yield-er).")
    ap.add_argument("target", help="URL or domain, e.g. example.com")
    ap.add_argument("--json", action="store_true", help="emit JSON")
    ap.add_argument("--no-whois", action="store_true",
                    help="skip whois IP lookups (faster)")
    ap.add_argument("--nmap", action="store_true",
                    help="run `nmap -O` for OS fingerprint (slow, needs sudo, "
                         "usually CDN-blocked)")
    ap.add_argument("--no-ssh", action="store_true",
                    help="skip the TCP/22 SSH banner grab")
    ap.add_argument("--find-origin", action="store_true",
                    help="probe common origin subdomains even if no CDN is "
                         "detected (auto-runs when a CDN is found)")
    ap.add_argument("--trace", action="store_true",
                    help="traceroute the network path (targets a discovered "
                         "origin if found, else the host) and annotate each "
                         "hop with its ASN/network. ACTIVE.")
    ap.add_argument("--no-crtsh", action="store_true",
                    help="skip Certificate Transparency (crt.sh) subdomain "
                         "mining during origin discovery")
    ap.add_argument("--no-exposure", action="store_true",
                    help="skip the origin exposure check. By default, "
                         "discovered web origins are probed directly (correct "
                         "SNI/Host) to test CDN/WAF bypass — an ACTIVE "
                         "connection to the origin. Only run against targets "
                         "you are authorized to test.")
    ap.add_argument("--no-color", action="store_true")
    args = ap.parse_args()

    host, url = normalize(args.target)
    findings = analyze(host, do_whois=not args.no_whois, use_nmap=args.nmap,
                       check_ssh=not args.no_ssh, url=url,
                       find_origin=args.find_origin,
                       use_crtsh=not args.no_crtsh,
                       check_exposure=not args.no_exposure,
                       trace=args.trace)

    if args.json:
        print(json.dumps(findings, indent=2, default=str))
    else:
        print_report(findings, use_color=not args.no_color)


if __name__ == "__main__":
    main()