#!/usr/bin/env python3
"""Find websites that do not support TLS 1.3.

Reads domains (one per line) and performs a normal TLS handshake with each,
offering every protocol version from TLS 1.0 to 1.3. The server picks the
highest version it supports, so a negotiated version below 1.3 means the
server has no TLS 1.3 support.

Each host is also classified by where it is hosted (CDN, hosted platform,
cloud, or the company's own network / colocation), and --subdomains expands
every input domain with hostnames found in Certificate Transparency logs.
The server certificate is inspected too, so results can be filtered on
certificates that are breakable today or by a future quantum computer.
For each host written, the owner is looked up for free from the certificate
subject and the domain's registration record (RDAP).

Usage:
    python3 tls_scan.py domains.txt -o results.csv
    python3 tls_scan.py domains.txt -o results.csv --subdomains --legacy --all
    python3 tls_scan.py domains.txt -o weak_certs.csv --filter weak-cert
"""

import argparse
import csv
import functools
import json
import re
import socket
import ssl
import subprocess
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
import warnings
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone

warnings.filterwarnings("ignore", category=DeprecationWarning)

LEGACY_VERSIONS = {
    "tls1.0": ssl.TLSVersion.TLSv1,
    "tls1.1": ssl.TLSVersion.TLSv1_1,
}
FIELDS = ["domain", "parent", "ip", "asn", "as_name", "cname", "hosting", "lead",
          "status", "max_version", "cipher", "bits", "tls1.0", "tls1.1",
          "key_type", "key_bits", "sig_alg", "cert_expires", "cert_breakable",
          "cert_issues", "cert_org", "cert_location", "registrant_org",
          "registrant_location", "error"]

# Hosting categories where the site owner cannot change the TLS configuration.
NOT_CUSTOMER_CONTROLLED = {"cdn", "platform"}

# Matched against the end of the host's CNAME chain. Checked before the AS name
# because CDNs and platforms often sit on a cloud provider's network.
CNAME_RULES = [
    ("cdn", ("cloudflare.net", "cloudfront.net", "akamaiedge.net", "akamai.net",
             "edgekey.net", "edgesuite.net", "fastly.net", "azureedge.net",
             "azurefd.net", "incapdns.net", "sucuri.net", "b-cdn.net")),
    ("platform", ("myshopify.com", "shopify.com", "wixdns.net", "squarespace.com",
                  "wpengine.com", "wordpress.com", "github.io", "netlify.app",
                  "netlify.com", "vercel-dns.com", "hubspot.net", "hs-sites.com",
                  "webflow.com", "webflow.io", "godaddysites.com", "secureserver.net",
                  "weebly.com", "kinsta.cloud", "pantheonsite.io")),
]

# Matched in order against the upper-cased AS name. Linode comes first because
# its AS name also contains AKAMAI.
AS_RULES = [
    ("cloud", ("LINODE",)),
    ("cdn", ("CLOUDFLARE", "AKAMAI", "FASTLY", "INCAPSULA", "IMPERVA", "SUCURI",
             "EDGECAST", "EDGIO", "CDN77", "HIGHWINDS", "STACKPATH", "BUNNY")),
    ("platform", ("GO-DADDY", "GODADDY", "WIX_COM", "WIX-COM", "SQUARESPACE", "SHOPIFY",
                  "WPENGINE", "AUTOMATTIC", "UNIFIEDLAYER", "NEWFOLD", "ENDURANCE",
                  "HOSTGATOR", "BLUEHOST", "WEEBLY", "HUBSPOT", "NAMECHEAP",
                  "DREAMHOST", "SITEGROUND", "HOSTINGER", "NETLIFY", "VERCEL", "GITHUB")),
    ("cloud", ("AMAZON", "MICROSOFT", "GOOGLE", "ORACLE", "ALIBABA", "DIGITALOCEAN",
               "OVH", "HETZNER", "CHOOPA", "VULTR", "RACKSPACE", "SOFTLAYER", "LEASEWEB")),
]


def classify_hosting(cnames, as_name):
    for category, suffixes in CNAME_RULES:
        if any(name == s or name.endswith("." + s) for name in cnames for s in suffixes):
            return category
    if not as_name:
        return "unknown"
    upper = as_name.upper()
    for category, keywords in AS_RULES:
        if any(k in upper for k in keywords):
            return category
    # Not a known provider: the company's own network, an ISP, or colocation.
    return "self_or_colo"


def fetch_subdomains(parent, limit, timeout=30, attempts=3):
    """Return hostnames under `parent` seen in Certificate Transparency logs."""
    url = "https://crt.sh/?" + urllib.parse.urlencode({"q": f"%.{parent}", "output": "json"})
    request = urllib.request.Request(url, headers={"User-Agent": "tls-scan/1.0"})
    for attempt in range(attempts):
        try:
            with urllib.request.urlopen(request, timeout=timeout) as response:
                entries = json.load(response)
            break
        except (urllib.error.URLError, TimeoutError, ValueError, OSError):
            # crt.sh is frequently overloaded; back off and retry.
            time.sleep(2 * (attempt + 1))
    else:
        return None

    names = set()
    for entry in entries:
        for name in entry.get("name_value", "").split("\n"):
            name = name.strip().lower().rstrip(".")
            if "*" in name or "@" in name or name == parent:
                continue
            if name.endswith("." + parent):
                names.add(name)
    return sorted(names)[:limit]


def resolve(host):
    canonical, aliases, addresses = socket.gethostbyname_ex(host)
    cnames = [n.lower().rstrip(".") for n in [canonical, *aliases] if n.lower().rstrip(".") != host]
    return addresses[0], cnames


def system_nameservers():
    try:
        with open("/etc/resolv.conf") as fh:
            servers = [line.split()[1] for line in fh if line.startswith("nameserver")]
    except OSError:
        servers = []
    return servers or ["1.1.1.1"]


def skip_dns_name(packet, offset):
    while True:
        length = packet[offset]
        if length >= 0xC0:  # compression pointer
            return offset + 2
        if length == 0:
            return offset + 1
        offset += length + 1


def txt_query(name, nameservers, timeout=5):
    """Return the first TXT record for `name`, or None. The stdlib has no TXT resolver."""
    question = b"".join(bytes([len(p)]) + p.encode() for p in name.split(".")) + b"\0"
    packet = b"\x12\x34\x01\x00\x00\x01\x00\x00\x00\x00\x00\x00" + question + b"\x00\x10\x00\x01"
    # UDP queries get dropped under load, so go round the nameservers three times.
    for server in nameservers * 3:
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
                sock.settimeout(timeout)
                sock.sendto(packet, (server, 53))
                reply = sock.recv(4096)
            offset = skip_dns_name(reply, 12) + 4
            for _ in range(int.from_bytes(reply[6:8], "big")):
                offset = skip_dns_name(reply, offset)
                rtype = int.from_bytes(reply[offset:offset + 2], "big")
                rdlen = int.from_bytes(reply[offset + 8:offset + 10], "big")
                offset += 10
                if rtype == 16:
                    return reply[offset + 1:offset + 1 + reply[offset]].decode(errors="replace")
                offset += rdlen
            return None
        except (OSError, IndexError):
            continue
    return None


def lookup_asns_dns(ips, workers=20):
    """Per-IP fallback for lookup_asns using Team Cymru's DNS interface."""
    nameservers = system_nameservers()

    def origin(ip):
        reversed_ip = ".".join(reversed(ip.split(".")))
        answer = txt_query(f"{reversed_ip}.origin.asn.cymru.com", nameservers)
        # An IP announced by several networks lists them all; take the first.
        return answer.split("|")[0].split()[0] if answer and answer.split("|")[0].strip() else None

    def as_name(asn):
        answer = txt_query(f"AS{asn}.asn.cymru.com", nameservers)
        return answer.split("|")[-1].strip() if answer else ""

    ips = sorted(ips)
    with ThreadPoolExecutor(max_workers=workers) as pool:
        origins = dict(zip(ips, pool.map(origin, ips)))
        unique = sorted({asn for asn in origins.values() if asn})
        names = dict(zip(unique, pool.map(as_name, unique)))
    return {ip: (asn, names[asn]) for ip, asn in origins.items() if asn}


def lookup_asns(ips, timeout=10, batch=2000):
    """Map each IP to (asn, as_name) using Team Cymru's IP-to-ASN service."""
    try:
        return lookup_asns_whois(ips, timeout, batch)
    except OSError:
        # Outbound whois (port 43) is often blocked; DNS is slower but gets through.
        return lookup_asns_dns(ips)


def lookup_asns_whois(ips, timeout, batch):
    result = {}
    ips = sorted(ips)
    for start in range(0, len(ips), batch):
        query = "begin\nverbose\n" + "\n".join(ips[start:start + batch]) + "\nend\n"
        chunks = []
        with socket.create_connection(("whois.cymru.com", 43), timeout=timeout) as sock:
            sock.sendall(query.encode())
            while data := sock.recv(65536):
                chunks.append(data)
        for line in b"".join(chunks).decode(errors="replace").splitlines():
            parts = [p.strip() for p in line.split("|")]
            if len(parts) == 7 and parts[0] != "NA":
                result[parts[1]] = (parts[0], parts[6])
    return result


# Public-key algorithms a quantum computer running Shor's algorithm can break,
# as named in `openssl x509 -text` output.
KEY_TYPES = {
    "rsaEncryption": "RSA",
    "rsassaPss": "RSA",
    "id-ecPublicKey": "EC",
    "dsaEncryption": "DSA",
    "ED25519": "Ed25519",
    "ED448": "Ed448",
}
QUANTUM_SAFE_PREFIXES = ("ML-DSA", "SLH-DSA")
WEAK_HASHES = ("sha1", "md5", "md2")
# Minimum key sizes still considered secure against classical attack.
MIN_KEY_BITS = {"RSA": 2048, "DSA": 2048, "EC": 224}


def inspect_cert(der):
    """Return certificate columns for a DER certificate, using the openssl CLI.

    cert_breakable is "now" for a key or signature that is already weak,
    "quantum" for one only a quantum computer could break, and "no" for a
    post-quantum certificate.
    """
    text = subprocess.run(
        ["openssl", "x509", "-inform", "DER", "-noout", "-text",
         "-nameopt", "esc_2253,utf8,sep_comma_plus,sname"],
        input=der, capture_output=True, check=True, timeout=10,
    ).stdout.decode(errors="replace")

    def field(pattern):
        match = re.search(pattern, text)
        return match.group(1).strip() if match else ""

    algorithm = field(r"Public Key Algorithm: (.+)")
    key_type = KEY_TYPES.get(algorithm, algorithm)
    bits = field(r"Public-Key: \((\d+) bit\)")
    sig_alg = field(r"Signature Algorithm: (.+)")
    info = {"key_type": key_type, "key_bits": bits, "sig_alg": sig_alg}

    issues = []
    if bits and int(bits) < MIN_KEY_BITS.get(key_type, 0):
        issues.append(f"weak-key({key_type}-{bits})")
    if any(h in sig_alg.lower() for h in WEAK_HASHES):
        issues.append(f"weak-signature({sig_alg})")
    if issues:
        info["cert_breakable"] = "now"
    elif key_type.upper().startswith(QUANTUM_SAFE_PREFIXES) \
            and sig_alg.upper().startswith(QUANTUM_SAFE_PREFIXES):
        info["cert_breakable"] = "no"
    else:
        info["cert_breakable"] = "quantum"

    # Not breakable, but worth knowing about when contacting the owner.
    expires = field(r"Not After : (.+)")
    if expires:
        expiry = datetime.strptime(expires, "%b %d %H:%M:%S %Y %Z").replace(tzinfo=timezone.utc)
        info["cert_expires"] = expiry.date().isoformat()
        if expiry < datetime.now(timezone.utc):
            issues.append("expired")
    subject = field(r"Subject: (.+)")
    if field(r"Issuer: (.+)") == subject:
        issues.append("self-signed")
    info["cert_issues"] = "; ".join(issues)

    # Organisation-validated certificates carry the owner's verified legal name
    # and location; domain-validated ones only have a CN, leaving these blank.
    attrs = {}
    for part in re.split(r"(?<!\\),", subject):
        key, _, value = part.partition("=")
        attrs[key.strip()] = re.sub(r"\\(.)", r"\1", value)
    info["cert_org"] = attrs.get("O", "")
    info["cert_location"] = ", ".join(
        attrs[k] for k in ("L", "ST", "C") if attrs.get(k))
    return info


REDACTION_MARKERS = ("redacted", "privacy", "protect", "proxy", "withheld",
                     "whoisguard", "not disclosed", "masked", "statutory", "dnstination")


def rdap_get(url, timeout=15, attempts=3):
    request = urllib.request.Request(
        url, headers={"Accept": "application/rdap+json", "User-Agent": "tls-scan/1.0"})
    for attempt in range(attempts):
        try:
            with urllib.request.urlopen(request, timeout=timeout) as response:
                return json.load(response)
        except urllib.error.HTTPError as exc:
            # 404 means no such record; anything else may be rate limiting.
            if exc.code == 404 or attempt == attempts - 1:
                raise
            retry_after = exc.headers.get("Retry-After", "")
            time.sleep(min(int(retry_after), 30) + 1 if retry_after.isdigit() else 5)
        except OSError:
            if attempt == attempts - 1:
                raise
            time.sleep(3)


@functools.lru_cache(maxsize=None)
def rdap_servers():
    """Map each top-level domain to its registry's RDAP server, from IANA's list."""
    servers = {}
    for tlds, urls in rdap_get("https://data.iana.org/rdap/dns.json")["services"]:
        for tld in tlds:
            servers[tld.lower()] = urls[0].rstrip("/")
    return servers


def find_registrant(data):
    entities = list(data.get("entities", []))
    while entities:
        entity = entities.pop()
        if "registrant" in entity.get("roles", []):
            return entity
        entities.extend(entity.get("entities", []))
    return None


def registrant_columns(entity):
    props = {}
    for item in (entity.get("vcardArray") or [None, []])[1]:
        if len(item) >= 4:
            props.setdefault(item[0], (item[1], item[3]))

    org = props.get("org", props.get("fn", ({}, "")))[1]
    org = " ".join(org) if isinstance(org, list) else str(org or "")
    if any(marker in org.lower() for marker in REDACTION_MARKERS):
        return {"registrant_org": "redacted"}

    params, address = props.get("adr", ({}, []))
    # A vCard address is [po box, extension, street, city, region, postcode, country].
    address = address if isinstance(address, list) and len(address) == 7 else [""] * 7
    place = [address[3], address[4], address[6] or params.get("cc", "")]
    place = [p for p in place if isinstance(p, str) and p
             and not any(marker in p.lower() for marker in REDACTION_MARKERS)]
    return {"registrant_org": org.strip(), "registrant_location": ", ".join(place)}


def lookup_registrant(domain):
    """Return registrant columns from the domain's RDAP registration record."""
    labels = domain.split(".")
    try:
        server = rdap_servers().get(labels[-1])
    except (OSError, ValueError, KeyError):
        return {"registrant_org": "lookup failed"}
    if not server:
        # Some registries, such as .de and .edu, do not offer RDAP.
        return {}
    # Only the registrable domain has a record, so drop labels until one is found.
    for start in range(len(labels) - 1):
        name = ".".join(labels[start:])
        try:
            data = rdap_get(f"{server}/domain/{name}")
            entity = find_registrant(data)
            if not entity:
                # Registries such as .com hold no owner data; the registrar's
                # record, linked from the registry's, does.
                for link in data.get("links", []):
                    if link.get("rel") == "related" and "rdap" in link.get("type", ""):
                        entity = find_registrant(rdap_get(link["href"]))
                        break
        except urllib.error.HTTPError as exc:
            if exc.code == 404:
                continue
            return {"registrant_org": "lookup failed"}
        except (OSError, ValueError):
            return {"registrant_org": "lookup failed"}
        return registrant_columns(entity) if entity else {}
    return {}


def make_context(minimum, maximum):
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    # We are measuring protocol support, not trusting the connection.
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    # SECLEVEL=0 lets OpenSSL 3 offer TLS 1.0/1.1 and old ciphers at all.
    ctx.set_ciphers("ALL:@SECLEVEL=0")
    ctx.minimum_version = minimum
    ctx.maximum_version = maximum
    return ctx


def handshake(host, ip, port, timeout, minimum, maximum):
    ctx = make_context(minimum, maximum)
    with socket.create_connection((ip, port), timeout=timeout) as sock:
        with ctx.wrap_socket(sock, server_hostname=host) as tls:
            name, _, bits = tls.cipher()
            return tls.version(), name, bits, tls.getpeercert(binary_form=True)


def scan(row, port, timeout, legacy):
    host, ip = row["domain"].rpartition(":")[0] or row["domain"], row["ip"]
    try:
        version, cipher, bits, cert = handshake(
            host, ip, port, timeout, ssl.TLSVersion.TLSv1, ssl.TLSVersion.TLSv1_3
        )
    except Exception as exc:
        row["status"] = "error"
        row["error"] = f"{type(exc).__name__}: {exc}"
        return row

    row.update(max_version=version, cipher=cipher, bits=bits)
    row["status"] = "ok" if version == "TLSv1.3" else "no_tls13"
    try:
        row.update(inspect_cert(cert))
    except (OSError, subprocess.SubprocessError, ValueError) as exc:
        row["error"] = f"certificate not inspected: {type(exc).__name__}: {exc}"

    if legacy:
        for label, ver in LEGACY_VERSIONS.items():
            try:
                handshake(host, ip, port, timeout, ver, ver)
                row[label] = "yes"
            except Exception:
                row[label] = "no"
    return row


def read_domains(path, default_port):
    """Yield (host, port) pairs; a "host:port" entry overrides the default port."""
    seen = set()
    with open(path) as fh:
        for line in fh:
            # Accept plain lists and Tranco-style "rank,domain" CSVs.
            entry = line.strip().split(",")[-1].strip().lower()
            if not entry or entry.startswith("#"):
                continue
            entry = entry.removeprefix("https://").removeprefix("http://").split("/")[0]
            host, _, port = entry.partition(":")
            if entry not in seen:
                seen.add(entry)
                yield host, int(port) if port else default_port


FILTERS = {
    "no-tls13": lambda row: row["status"] == "no_tls13",
    "weak-cert": lambda row: row["cert_breakable"] == "now",
    "quantum-cert": lambda row: row["cert_breakable"] in ("now", "quantum"),
}
FILTERS["any"] = lambda row: FILTERS["no-tls13"](row) or FILTERS["weak-cert"](row)


def log(message):
    print(message, file=sys.stderr)


def main():
    parser = argparse.ArgumentParser(description=__doc__.split("\n")[0])
    parser.add_argument("domains", help="file with one domain per line")
    parser.add_argument("-o", "--output", default="results.csv")
    parser.add_argument("-w", "--workers", type=int, default=50)
    parser.add_argument("-t", "--timeout", type=float, default=8.0)
    parser.add_argument("-p", "--port", type=int, default=443)
    parser.add_argument("--legacy", action="store_true",
                        help="also test whether TLS 1.0 and 1.1 are accepted")
    parser.add_argument("--subdomains", action="store_true",
                        help="also scan subdomains found in Certificate Transparency logs")
    parser.add_argument("--max-subdomains", type=int, default=100,
                        help="cap on subdomains per input domain (default 100)")
    parser.add_argument("--filter", choices=FILTERS, default="no-tls13",
                        help="which hosts to write: no TLS 1.3 (default), a certificate "
                             "breakable today (weak-cert), a certificate breakable today or "
                             "by a quantum computer (quantum-cert), or no-tls13 plus "
                             "weak-cert (any)")
    parser.add_argument("--all", action="store_true",
                        help="write every host, ignoring --filter")
    parser.add_argument("--no-registrant", action="store_true",
                        help="skip the registration record (RDAP) lookup of each owner")
    args = parser.parse_args()

    # Each target is (host, port, parent domain it was found under).
    inputs = list(read_domains(args.domains, args.port))
    targets = [(host, port, host) for host, port in inputs]

    if args.subdomains:
        known = {host for host, _ in inputs}
        parents = sorted(known)
        # crt.sh is a free shared service, so query it gently.
        with ThreadPoolExecutor(max_workers=2) as pool:
            for parent, names in zip(parents, pool.map(
                    lambda p: fetch_subdomains(p, args.max_subdomains), parents)):
                if names is None:
                    log(f"crt.sh lookup failed for {parent}; scanning the domain itself only")
                    continue
                for name in names:
                    if name not in known:
                        known.add(name)
                        targets.append((name, args.port, parent))
        log(f"{len(targets) - len(inputs)} subdomains found for {len(inputs)} domains")

    rows, ports = [], {}
    with ThreadPoolExecutor(max_workers=args.workers) as pool:
        futures = {pool.submit(resolve, host): (host, port, parent)
                   for host, port, parent in targets}
        for future in as_completed(futures):
            host, port, parent = futures[future]
            row = dict.fromkeys(FIELDS, "")
            row["domain"] = host if port == args.port else f"{host}:{port}"
            row["parent"] = parent
            ports[row["domain"]] = port
            try:
                row["ip"], cnames = future.result()
                row["cname"] = cnames[0] if cnames else ""
                row["_cnames"] = cnames
            except OSError as exc:
                row["status"] = "error"
                row["error"] = f"DNS: {exc}"
            rows.append(row)

    ips = {r["ip"] for r in rows if r["ip"]}
    asns = lookup_asns(ips)
    if len(asns) < len(ips):
        log(f"No network owner found for {len(ips) - len(asns)} of {len(ips)} IPs; "
            "those hosts are classified from CNAMEs only")
    for row in rows:
        if row["ip"]:
            row["asn"], row["as_name"] = asns.get(row["ip"], ("", ""))
            row["hosting"] = classify_hosting(row.pop("_cnames"), row["as_name"])

    counts = {"ok": 0, "no_tls13": 0, "error": 0}
    matches = FILTERS[args.filter]
    matched = leads = 0
    registrants, registrants_lock = {}, threading.Lock()
    with open(args.output, "w", newline="") as out, \
            ThreadPoolExecutor(max_workers=args.workers) as pool, \
            ThreadPoolExecutor(max_workers=4) as rdap_pool:
        writer = csv.DictWriter(out, fieldnames=FIELDS)
        writer.writeheader()

        def scan_host(row):
            scan(row, ports[row["domain"]], args.timeout, args.legacy)
            wanted = row["status"] != "error" and (args.all or matches(row))
            if wanted and not args.no_registrant:
                # One lookup per parent domain, a few at a time: RDAP servers
                # rate-limit, and every subdomain shares its parent's record.
                with registrants_lock:
                    if row["parent"] not in registrants:
                        registrants[row["parent"]] = rdap_pool.submit(
                            lookup_registrant, row["parent"])
                row.update(registrants[row["parent"]].result())
            return row

        def record(row):
            nonlocal matched, leads
            counts[row["status"]] += 1
            if matches(row):
                matched += 1
                if row["hosting"] in NOT_CUSTOMER_CONTROLLED:
                    row["lead"] = "no"
                else:
                    row["lead"] = "check" if row["hosting"] == "unknown" else "yes"
                leads += row["lead"] == "yes"
            if args.all or matches(row):
                writer.writerow(row)
                out.flush()

        futures = []
        for row in rows:
            if row["status"] == "error":
                record(row)
            else:
                futures.append(pool.submit(scan_host, row))
        for done, future in enumerate(as_completed(futures), 1):
            record(future.result())
            if done % 100 == 0 or done == len(futures):
                log(f"{done}/{len(futures)} scanned, {matched} match --filter {args.filter} "
                    f"({leads} leads), {counts['no_tls13']} without TLS 1.3, "
                    f"{counts['error']} errors")

    log(f"Wrote {args.output}")


if __name__ == "__main__":
    main()