#!/usr/bin/env python3 """Developer platform DNS and HTTPS survey (September 2026) for and.guide. For each apex domain in DOMAINS this script records: DNS (Cloudflare DNS-over-HTTPS JSON API, https://cloudflare-dns.com/dns-query) A, AAAA, CAA (issue / issuewild / iodef), NS (+ provider derived from NS hostnames), DS and the AD flag, HTTPS (type 65), MX (+ null MX), SPF (apex TXT starting with v=spf1) and DMARC (_dmarc. TXT, p= / sp= / pct=). Every query sets do=1. RFC 6840 section 5.8 says a validating resolver SHOULD set AD only when the query carried DO or AD; in a pilot run without do=1, Cloudflare's JSON API returned AD=false for some signed zones whose answers validated (AD=true) as soon as do=1 was added. HTTP (urllib, redirects NOT followed automatically) First response of https:/// and https://www./ (status and Location), then up to MAX_HOPS redirects followed manually; final URL and host; the Strict-Transport-Security header on the apex's first HTTPS response (and on the final response of the apex chain). Failures on the www host are published only as one coarse class (WWW_ERROR). Some specific failure modes, such as a name that does not resolve or a certificate issued for an unrelated name, can resemble a dangling DNS record, and we do not publish per-domain hints of that kind for third-party domains. Apex failures keep their detail. TLS (ssl module, SNI = the apex domain) Issuer organization and common name, notAfter, negotiated TLS version. Everything runs sequentially with a pause before every request, a timeout, and one retry on network errors. No subdomains other than the conventional www host and the _dmarc policy name are queried, and nothing is probed beyond one ordinary GET per hop and one TLS handshake per apex. Standard library only (Python 3.9+). Usage (from the repository root): python3 scripts/research/dev-domain-survey-2026-09.py --with-preload-list Full run. Writes public/research/developer-domains-2026-09.csv and src/data/research/developer-domains-2026-09.json. --with-preload-list adds one request for Chromium's HSTS preload list source file and reports how many surveyed domains it covers. python3 scripts/research/dev-domain-survey-2026-09.py --limit 3 \\ --csv /tmp/test.csv --json /tmp/test.json Quick test run on the first three domains (--only a.com,b.com also works). python3 scripts/research/dev-domain-survey-2026-09.py --summarize-only Rebuild the JSON summary from an existing CSV without contacting any surveyed domain (add --with-preload-list to refresh the list counts). Published data for September 2026: one full run on 26 September 2026 UTC (14:59:55 to 15:11:59). A pilot run earlier that day (not published) led to the do=1 change, the coarse www errors, the status-aware canonical-host labels, the caa_policy and https_rr_mode columns, and three nameserver patterns (dnsimple-edge.io, deSEC, PCH). After the run, the summary was built with --summarize-only --with-preload-list (list fetched 15:14:19 UTC); later --summarize-only passes only added summary fields. The CSV was not edited after the run. Results depend on the vantage point (CDNs answer differently by location), on the moment of measurement, and on the resolver. Treat them as a dated snapshot, not a ranking. """ import argparse import base64 import collections import csv import datetime as dt import hashlib import json import re import socket import ssl import sys import time import urllib.error import urllib.parse import urllib.request from pathlib import Path SURVEY_ID = "developer-domains-2026-09" REPO_ROOT = Path(__file__).resolve().parents[2] DEFAULT_CSV = REPO_ROOT / "public" / "research" / f"{SURVEY_ID}.csv" DEFAULT_JSON = REPO_ROOT / "src" / "data" / "research" / f"{SURVEY_ID}.json" DOH_ENDPOINT = "https://cloudflare-dns.com/dns-query" USER_AGENT = ( "and.guide-research/2026.09 " "(+https://and.guide/guides/developer-domains-dns-https-survey-2026/)" ) TIMEOUT = 15 # seconds, per request PAUSE = 0.4 # seconds, before every request (DoH, HTTP, TLS) MAX_HOPS = 5 # redirects followed manually after the first response VANTAGE_POINT = "South Korea (one macOS workstation; system resolver for HTTP/TLS connections)" WWW_ERROR = "error: www request failed (detail not published)" # --------------------------------------------------------------------------- # Sample: widely used developer-facing platforms and projects. # Rule: the platform or project uses this apex (or its www host) as its own # primary website. Products that live under a parent company's domain # (for example cloud.google.com or aws.amazon.com) are not included. # --------------------------------------------------------------------------- CATEGORIES = { "Cloud & app hosting": [ "vercel.com", "netlify.com", "heroku.com", "render.com", "fly.io", "railway.com", "digitalocean.com", "linode.com", "vultr.com", "hetzner.com", "ovhcloud.com", "scaleway.com", "backblaze.com", ], "DNS, CDN & domains": [ "cloudflare.com", "fastly.com", "akamai.com", "bunny.net", "gcore.com", "dnsimple.com", "namecheap.com", "porkbun.com", "gandi.net", "godaddy.com", "quad9.net", "nextdns.io", ], "Code hosting & CI/CD": [ "github.com", "gitlab.com", "bitbucket.org", "codeberg.org", "sr.ht", "gitea.com", "sourcegraph.com", "circleci.com", "travis-ci.com", "buildkite.com", "jenkins.io", "codecov.io", ], "Package registries": [ "npmjs.com", "pypi.org", "crates.io", "rubygems.org", "packagist.org", "nuget.org", "maven.org", "hex.pm", "pub.dev", "cocoapods.org", "jsr.io", "quay.io", ], "Developer tools & APIs": [ "jetbrains.com", "postman.com", "atlassian.com", "linear.app", "stripe.com", "twilio.com", "algolia.com", "docker.com", "hashicorp.com", "tailscale.com", "ngrok.com", "replit.com", "huggingface.co", ], "Databases & backend services": [ "supabase.com", "planetscale.com", "neon.com", "mongodb.com", "redis.io", "cockroachlabs.com", "clickhouse.com", "upstash.com", "turso.tech", "convex.dev", "prisma.io", "postgresql.org", "sqlite.org", ], "Observability & security": [ "datadoghq.com", "sentry.io", "newrelic.com", "grafana.com", "honeycomb.io", "pagerduty.com", "elastic.co", "posthog.com", "letsencrypt.org", "snyk.io", "1password.com", "auth0.com", ], "Languages, frameworks & docs": [ "python.org", "rust-lang.org", "go.dev", "nodejs.org", "deno.com", "bun.sh", "typescriptlang.org", "php.net", "react.dev", "vuejs.org", "djangoproject.com", "kubernetes.io", "stackoverflow.com", ], } DOMAINS = [(d, cat) for cat, ds in CATEGORIES.items() for d in ds] # DNS provider derived from NS hostnames. Only unambiguous, well-known # patterns are mapped; everything else is "other/unknown". Nameservers named # under the domain itself are also "other/unknown": the operator behind a # vanity or self-hosted name cannot be identified from the name alone. NS_PROVIDER_PATTERNS = [ (r"\.ns\.cloudflare\.com$", "Cloudflare"), (r"\.foundationdns\.(com|net|org)$", "Cloudflare"), (r"^ns-\d+\.awsdns-\d+\.(com|net|org|co\.uk)$", "Amazon Route 53"), (r"^ns-cloud-[a-z]\d+\.googledomains\.com$", "Google Cloud DNS"), (r"\.azure-dns\.(com|net|org|info)$", "Azure DNS"), (r"\.nsone\.net$", "NS1"), (r"\.akam\.net$", "Akamai Edge DNS"), (r"\.dynect\.net$", "Oracle Dyn"), (r"\.dns\.oraclecloud\.net$", "Oracle Cloud DNS"), (r"\.ultradns\.(com|net|org|biz|info|co\.uk)$", "UltraDNS"), (r"\.dnsimple\.com$", "DNSimple"), (r"\.dnsimple-edge\.(com|net|org|io)$", "DNSimple"), (r"\.desec\.(io|org|ch|cz|li)$", "deSEC"), (r"\.pch\.net$", "PCH (Packet Clearing House)"), (r"\.digitalocean\.com$", "DigitalOcean"), (r"\.vercel-dns\.com$", "Vercel"), (r"\.linode\.com$", "Linode (Akamai)"), (r"\.ns\.hetzner\.(com|de)$", "Hetzner"), (r"\.ovh\.net$", "OVHcloud"), (r"\.anycast\.me$", "OVHcloud"), (r"\.registrar-servers\.com$", "Namecheap"), (r"\.domaincontrol\.com$", "GoDaddy"), (r"\.ns\.porkbun\.com$", "Porkbun"), (r"\.gandi\.net$", "Gandi"), (r"\.dnsmadeeasy\.com$", "DNS Made Easy"), (r"\.bunny\.net$", "bunny.net"), (r"\.gcorelabs\.net$", "Gcore"), (r"\.gcdn\.services$", "Gcore"), (r"\.dom\.scw\.cloud$", "Scaleway"), (r"\.vultr\.com$", "Vultr"), (r"\.he\.net$", "Hurricane Electric"), (r"\.cloudns\.net$", "ClouDNS"), (r"\.constellix\.(com|net)$", "Constellix"), ] TYPE_CODES = {"A": 1, "NS": 2, "CNAME": 5, "MX": 15, "TXT": 16, "AAAA": 28, "DS": 43, "HTTPS": 65, "CAA": 257} RCODES = {0: "NOERROR", 1: "FORMERR", 2: "SERVFAIL", 3: "NXDOMAIN", 4: "NOTIMP", 5: "REFUSED"} CSV_COLUMNS = [ "domain", "category", "measured_at_utc", "a_records", "a_count", "aaaa_records", "aaaa_count", "ipv6", "ns_hosts", "dns_provider", "ds_present", "ds_algorithms", "ad_flag", "caa_present", "caa_policy", "caa_issue", "caa_issuewild", "caa_iodef", "caa_tags", "https_rr_present", "https_rr_mode", "https_rr_alpn", "https_rr", "mx_present", "mx_null", "spf_present", "spf_record_count", "spf_all", "dmarc_present", "dmarc_p", "dmarc_sp", "dmarc_pct", "apex_first_status", "apex_first_location", "apex_redirect_chain", "apex_final_url", "apex_final_host", "apex_final_status", "www_first_status", "www_first_location", "www_final_host", "www_final_status", "canonical_host", "hsts_header", "hsts_max_age", "hsts_include_subdomains", "hsts_preload", "hsts_final_header", "tls_issuer_org", "tls_issuer_cn", "tls_not_after", "tls_version", "tls_peer_ip", "errors", ] def log(msg): print(msg, file=sys.stderr, flush=True) def now_utc(): return dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def pause(): time.sleep(PAUSE) def describe_error(exc): """Short, explicit error string for the CSV.""" reason = exc if isinstance(exc, urllib.error.URLError) and not isinstance(exc, urllib.error.HTTPError): reason = exc.reason if isinstance(reason, (socket.timeout, TimeoutError)): return "error: timeout" if isinstance(reason, ssl.SSLCertVerificationError): return f"error: certificate verify failed ({reason.verify_message})" if isinstance(reason, ssl.SSLError): return f"error: TLS ({reason.reason or reason})" if isinstance(reason, socket.gaierror): return "error: name resolution failed" if isinstance(reason, ConnectionRefusedError): return "error: connection refused" if isinstance(reason, ConnectionResetError): return "error: connection reset" text = f"{type(reason).__name__}: {reason}".replace("\n", " ") return "error: " + text[:140] # --------------------------------------------------------------------------- # DNS over HTTPS (Cloudflare JSON API) # --------------------------------------------------------------------------- def doh_query(name, rtype): """Return (response_json, error_string). One retry on failure or SERVFAIL.""" # do=1: without the DO bit the AD flag is not reliable (RFC 6840 5.8). url = f"{DOH_ENDPOINT}?name={urllib.parse.quote(name)}&type={rtype}&do=1" req = urllib.request.Request(url, headers={ "accept": "application/dns-json", "user-agent": USER_AGENT}) last_err = None for attempt in (1, 2): pause() try: with urllib.request.urlopen(req, timeout=TIMEOUT) as resp: data = json.loads(resp.read().decode("utf-8")) except Exception as exc: # network, HTTP or JSON error last_err = describe_error(exc) continue status = data.get("Status") if status in (0, 3): # NOERROR, NXDOMAIN return data, None last_err = f"error: {RCODES.get(status, 'RCODE ' + str(status))}" return None, last_err def answers(data, rtype): code = TYPE_CODES[rtype] return [a.get("data", "") for a in (data or {}).get("Answer", []) if a.get("type") == code] def parse_character_strings(text): """Parse presentation-format character-strings ("a" "b" or bare tokens) and concatenate them, as SPF (RFC 7208 3.3) and DMARC require.""" out, buf, i, n = [], None, 0, len(text) while i < n: c = text[i] if buf is None: if c == '"': buf = [] elif not c.isspace(): j = i while j < n and not text[j].isspace(): j += 1 out.append(text[i:j]) i = j continue else: if c == "\\" and i + 1 < n: digits = text[i + 1:i + 4] if len(digits) == 3 and digits.isdigit(): buf.append(chr(int(digits))) i += 4 continue buf.append(text[i + 1]) i += 2 continue if c == '"': out.append("".join(buf)) buf = None else: buf.append(c) i += 1 if buf is not None: out.append("".join(buf)) return "".join(out) def parse_caa(rdata): """Return (flags, tag, value) from presentation or RFC 3597 generic format.""" rdata = rdata.strip() if rdata.startswith("\\#"): parts = rdata.split() raw = bytes.fromhex("".join(parts[2:])) taglen = raw[1] return raw[0], raw[2:2 + taglen].decode("ascii", "replace").lower(), \ raw[2 + taglen:].decode("utf-8", "replace") m = re.match(r"^(\d+)\s+(\S+)\s*(.*)$", rdata) if not m: return None, None, rdata rest = m.group(3).strip() value = parse_character_strings(rest) if rest.startswith('"') else rest return int(m.group(1)), m.group(2).lower(), value def caa_issuer(value): """Issuer domain of an issue/issuewild value; ';' means 'no CA allowed'.""" name = value.split(";", 1)[0].strip().lower() return name if name else ";" def ns_provider(hosts): providers = set() for host in hosts: h = host.lower().rstrip(".") for pattern, label in NS_PROVIDER_PATTERNS: if re.search(pattern, h): providers.add(label) break else: providers.add("other/unknown") return " + ".join(sorted(providers)) if providers else "" def svcb_alpn(rdata): m = re.search(r'alpn="?([^"\s]+)"?', rdata) return m.group(1) if m else "" def spf_all_qualifier(record): terms = record.split()[1:] for term in terms: m = re.fullmatch(r"([+\-~?]?)all", term.lower()) if m: return (m.group(1) or "+") + "all" if any(t.lower().startswith("redirect=") for t in terms): return "redirect" return "no all" def parse_dmarc(record): tags = {} for part in record.split(";"): if "=" in part: k, v = part.split("=", 1) tags[k.strip().lower()] = v.strip() return tags def measure_dns(domain, row, errors): # A (the AD flag of this response is recorded as the DNSSEC validation signal) data, err = doh_query(domain, "A") if err: row.update(a_records=err, a_count=err, ad_flag=err) errors.append(f"A {err}") else: a = answers(data, "A") row.update(a_records=" ".join(a), a_count=len(a), ad_flag="yes" if data.get("AD") else "no") data, err = doh_query(domain, "AAAA") if err: row.update(aaaa_records=err, aaaa_count=err, ipv6=err) errors.append(f"AAAA {err}") else: aaaa = answers(data, "AAAA") row.update(aaaa_records=" ".join(aaaa), aaaa_count=len(aaaa), ipv6="yes" if aaaa else "no") data, err = doh_query(domain, "NS") if err: row.update(ns_hosts=err, dns_provider=err) errors.append(f"NS {err}") else: hosts = sorted(h.lower().rstrip(".") for h in answers(data, "NS")) row.update(ns_hosts=" ".join(hosts), dns_provider=ns_provider(hosts) or "no NS answer") data, err = doh_query(domain, "DS") if err: row.update(ds_present=err, ds_algorithms=err) errors.append(f"DS {err}") else: ds = answers(data, "DS") algs = sorted({r.split()[1] for r in ds if len(r.split()) >= 2}, key=int) row.update(ds_present="yes" if ds else "no", ds_algorithms=" ".join(algs)) data, err = doh_query(domain, "CAA") if err: row.update(caa_present=err, caa_policy=err, caa_issue=err, caa_issuewild=err, caa_iodef=err, caa_tags=err) errors.append(f"CAA {err}") else: issue, issuewild, tags, iodef = set(), set(), set(), False for rdata in answers(data, "CAA"): _flags, tag, value = parse_caa(rdata) if tag is None: continue tags.add(tag) if tag == "issue": issue.add(caa_issuer(value)) elif tag == "issuewild": issuewild.add(caa_issuer(value)) elif tag == "iodef": iodef = True # RFC 8659 section 3: an RRset without issue/issuewild (for example only # iodef or contactemail) does not restrict issuance. if not tags: policy = "none" elif issue: policy = "issue" + (" + issuewild" if issuewild else "") elif issuewild: policy = "issuewild only" else: policy = "no issuer tags" row.update(caa_present="yes" if tags else "no", caa_policy=policy, caa_issue=" ".join(sorted(issue)), caa_issuewild=" ".join(sorted(issuewild)), caa_iodef="yes" if iodef else "no", caa_tags=" ".join(sorted(tags))) data, err = doh_query(domain, "HTTPS") if err: row.update(https_rr_present=err, https_rr_mode=err, https_rr_alpn=err, https_rr=err) errors.append(f"HTTPS {err}") else: rrs = answers(data, "HTTPS") # SvcPriority 0 = AliasMode, anything else = ServiceMode (RFC 9460 2.4) modes = sorted({"alias" if r.split()[0] == "0" else "service" for r in rrs if r and not r.startswith("\\#")}) row.update(https_rr_present="yes" if rrs else "no", https_rr_mode=" + ".join(modes), https_rr_alpn=" | ".join(svcb_alpn(r) for r in rrs), https_rr=" | ".join(rrs)) data, err = doh_query(domain, "MX") if err: row.update(mx_present=err, mx_null=err) errors.append(f"MX {err}") else: mx = answers(data, "MX") null_mx = len(mx) == 1 and mx[0].split()[-1] == "." row.update(mx_present="yes" if mx else "no", mx_null="yes" if null_mx else "no") data, err = doh_query(domain, "TXT") if err: row.update(spf_present=err, spf_record_count=err, spf_all=err) errors.append(f"TXT {err}") else: txts = [parse_character_strings(t) for t in answers(data, "TXT")] spf = [t for t in txts if re.match(r"^v=spf1(\s|$)", t, re.I)] row.update(spf_present="yes" if spf else "no", spf_record_count=len(spf), spf_all=spf_all_qualifier(spf[0]) if len(spf) == 1 else ("multiple records" if spf else "")) data, err = doh_query(f"_dmarc.{domain}", "TXT") if err: row.update(dmarc_present=err, dmarc_p=err, dmarc_sp=err, dmarc_pct=err) errors.append(f"DMARC {err}") else: txts = [parse_character_strings(t) for t in answers(data, "TXT")] dmarc = [t for t in txts if re.match(r"^\s*v\s*=\s*DMARC1\s*(;|$)", t, re.I)] if len(dmarc) == 1: tags = parse_dmarc(dmarc[0]) row.update(dmarc_present="yes", dmarc_p=tags.get("p", "").lower() or "missing p", dmarc_sp=tags.get("sp", "").lower(), dmarc_pct=tags.get("pct", "")) elif len(dmarc) > 1: row.update(dmarc_present="multiple records", dmarc_p="", dmarc_sp="", dmarc_pct="") else: row.update(dmarc_present="no", dmarc_p="", dmarc_sp="", dmarc_pct="") # --------------------------------------------------------------------------- # HTTP (no automatic redirects) # --------------------------------------------------------------------------- class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None # urllib then raises HTTPError carrying the 3xx response OPENER = urllib.request.build_opener( NoRedirect, urllib.request.HTTPSHandler(context=ssl.create_default_context())) def http_get(url): """One GET without following redirects. Returns (status, headers, error).""" req = urllib.request.Request(url, method="GET", headers={ "User-Agent": USER_AGENT, "Accept": "text/html,application/xhtml+xml,*/*;q=0.8", }) last_err = None for attempt in (1, 2): pause() try: resp = OPENER.open(req, timeout=TIMEOUT) status, headers = resp.status, resp.headers resp.close() return status, headers, None except urllib.error.HTTPError as exc: # 3xx (because of NoRedirect), 4xx, 5xx status, headers = exc.code, exc.headers exc.close() return status, headers, None except Exception as exc: last_err = describe_error(exc) if isinstance(exc, urllib.error.URLError) and isinstance( exc.reason, ssl.SSLCertVerificationError): break # a certificate failure will not change on retry return None, None, last_err def follow_chain(start_url): """First response plus up to MAX_HOPS manually followed redirects.""" chain, url = [], start_url for hop in range(MAX_HOPS + 1): status, headers, err = http_get(url) if err: chain.append({"url": url, "error": err}) return chain location = headers.get("Location") hsts = headers.get_all("Strict-Transport-Security") or [] step = {"url": url, "status": status, "location": location, "hsts": hsts} chain.append(step) if status in (301, 302, 303, 307, 308) and location: nxt = urllib.parse.urljoin(url, location) if urllib.parse.urlsplit(nxt).scheme not in ("http", "https"): return chain if hop == MAX_HOPS: step["error"] = f"error: more than {MAX_HOPS} redirects" return chain url = nxt continue return chain return chain def chain_summary(chain): parts = [] for step in chain: if "status" not in step: parts.append(step["error"]) continue if step.get("location"): parts.append(f"{step['status']} -> {step['location']}") else: parts.append(str(step["status"])) if step.get("error"): parts.append(step["error"]) return " ; ".join(parts) def chain_final(chain): """Return (final_url, final_host, final_status) or error strings.""" last = chain[-1] if last.get("error"): return last["error"], last["error"], last["error"] host = urllib.parse.urlsplit(last["url"]).hostname or "" return last["url"], host.lower(), last["status"] def parse_hsts(value): max_age, include_sub, preload = "", "no", "no" for part in value.split(";"): part = part.strip() if not part: continue if "=" in part: key, val = part.split("=", 1) if key.strip().lower() == "max-age": val = val.strip().strip('"') max_age = val if val.isdigit() else f"invalid ({val})" elif part.lower() == "includesubdomains": include_sub = "yes" elif part.lower() == "preload": preload = "yes" return max_age, include_sub, preload def is_ok(status): return str(status).isdigit() and 200 <= int(status) < 300 def is_blocked(status): return str(status).isdigit() and int(status) >= 400 def classify_canonical(domain, apex_first_status, apex_host, www_host, www_status): """Where https:/// ends up, qualified by what https://www./ does. apex apex stays on the apex; www redirects to the apex www apex redirects to www both-serve apex and www both return 2xx without redirecting to each other apex (www error) apex stays on the apex; the www request failed apex (www undetermined) apex stays on the apex; www answered 4xx/5xx without redirecting apex (www elsewhere) apex stays on the apex; www ends on some other host other-subdomain apex redirects to a different host under the same domain other-domain apex redirects to a different registrable domain undetermined the apex's first response was 4xx/5xx (for example bot protection), so no redirect behavior was observed error the apex request chain failed A redirect observed before a 4xx/5xx still counts: the redirect itself is the signal. """ www = f"www.{domain}" if apex_host.startswith("error"): return "error" if is_blocked(apex_first_status): return "undetermined" if apex_host == www: return "www" if apex_host == domain: if www_host == domain: return "apex" if www_host.startswith("error"): return "apex (www error)" if www_host == www: return "both-serve" if is_ok(www_status) else "apex (www undetermined)" return "apex (www elsewhere)" if apex_host.endswith("." + domain): return "other-subdomain" return "other-domain" def measure_http(domain, row, errors): apex_chain = follow_chain(f"https://{domain}/") first = apex_chain[0] if "status" in first: row["apex_first_status"] = first["status"] row["apex_first_location"] = first.get("location") or "" hsts_values = first["hsts"] row["hsts_header"] = " || ".join(hsts_values) if hsts_values: # RFC 6797 8.1: only the first header field is processed max_age, inc, pre = parse_hsts(hsts_values[0]) row.update(hsts_max_age=max_age, hsts_include_subdomains=inc, hsts_preload=pre) else: row.update(hsts_max_age="", hsts_include_subdomains="", hsts_preload="") else: err = first["error"] row.update(apex_first_status=err, apex_first_location="", hsts_header=err, hsts_max_age=err, hsts_include_subdomains=err, hsts_preload=err) errors.append(f"apex HTTPS {err}") row["apex_redirect_chain"] = chain_summary(apex_chain) final_url, final_host, final_status = chain_final(apex_chain) row.update(apex_final_url=final_url, apex_final_host=final_host, apex_final_status=final_status) last = apex_chain[-1] if "status" in last and not last.get("error"): row["hsts_final_header"] = " || ".join(last["hsts"]) else: row["hsts_final_header"] = last.get("error", "") if last.get("error") and "status" in first: errors.append(f"apex chain {last['error']}") www_chain = follow_chain(f"https://www.{domain}/") for step in www_chain: # publish only a coarse failure class for www if step.get("error"): step["error"] = WWW_ERROR wfirst = www_chain[0] if "status" in wfirst: row["www_first_status"] = wfirst["status"] row["www_first_location"] = wfirst.get("location") or "" else: row["www_first_status"] = wfirst["error"] row["www_first_location"] = "" if www_chain[-1].get("error"): errors.append(f"www {WWW_ERROR}") _url, www_host, www_status = chain_final(www_chain) row.update(www_final_host=www_host, www_final_status=www_status) row["canonical_host"] = classify_canonical( domain, row["apex_first_status"], str(final_host), str(www_host), www_status) # --------------------------------------------------------------------------- # TLS certificate for the apex # --------------------------------------------------------------------------- def measure_tls(domain, row, errors): ctx = ssl.create_default_context() last_err = None for attempt in (1, 2): pause() try: with socket.create_connection((domain, 443), timeout=TIMEOUT) as sock: with ctx.wrap_socket(sock, server_hostname=domain) as tls: cert = tls.getpeercert() version = tls.version() peer = tls.getpeername()[0] issuer = {} for rdn in cert.get("issuer", ()): for key, value in rdn: issuer.setdefault(key, value) row.update(tls_issuer_org=issuer.get("organizationName", ""), tls_issuer_cn=issuer.get("commonName", ""), tls_not_after=cert.get("notAfter", ""), tls_version=version or "", tls_peer_ip=peer) return except ssl.SSLCertVerificationError as exc: last_err = describe_error(exc) break except Exception as exc: last_err = describe_error(exc) row.update(tls_issuer_org=last_err, tls_issuer_cn=last_err, tls_not_after=last_err, tls_version=last_err, tls_peer_ip="") errors.append(f"TLS {last_err}") def measure(domain, category): row = {"domain": domain, "category": category, "measured_at_utc": now_utc()} errors = [] measure_dns(domain, row, errors) measure_http(domain, row, errors) measure_tls(domain, row, errors) row["errors"] = " | ".join(errors) return row # --------------------------------------------------------------------------- # Summary (computed only from the CSV rows, so it can be rebuilt offline) # --------------------------------------------------------------------------- def pct(n, d): return round(100.0 * n / d, 1) if d else 0.0 def count_values(rows, key, split=False): counter = collections.Counter() for r in rows: v = r[key] if split: for item in v.split(): counter[item] += 1 else: counter[v] += 1 return dict(counter.most_common()) def is_err(v): return str(v).startswith("error") def summarize(rows, meta): total = len(rows) def yes(key): return sum(1 for r in rows if r[key] == "yes") def errs(key): return sum(1 for r in rows if is_err(r[key])) def metric(key): n = yes(key) return {"count": n, "of": total, "percent": pct(n, total), "errors": errs(key)} caa_rows = [r for r in rows if r["caa_present"] == "yes"] caa_restricting = [r for r in caa_rows if r["caa_issue"] or r["caa_issuewild"]] caa_issuers = collections.Counter() for r in caa_restricting: for issuer in set(r["caa_issue"].split()) | set(r["caa_issuewild"].split()): caa_issuers[issuer] += 1 issue_counts = collections.Counter( len(r["caa_issue"].split()) for r in caa_restricting if r["caa_issue"]) blocked_first = [r for r in rows if is_blocked(r["apex_first_status"])] hsts_rows = [r for r in rows if r["hsts_header"] and not is_err(r["hsts_header"])] buckets = collections.Counter() for r in hsts_rows: v = r["hsts_max_age"] if not v.isdigit(): buckets["invalid or missing max-age"] += 1 continue age = int(v) if age == 0: buckets["0 (expire policy)"] += 1 elif age < 86400: buckets["under 1 day"] += 1 elif age < 31536000: buckets["1 day to under 1 year"] += 1 elif age < 63072000: buckets["1 year to under 2 years"] += 1 else: buckets["2 years or more"] += 1 ages = collections.Counter(r["hsts_max_age"] for r in hsts_rows) preload_ready = [ r["domain"] for r in hsts_rows if r["hsts_max_age"].isdigit() and int(r["hsts_max_age"]) >= 31536000 and r["hsts_include_subdomains"] == "yes" and r["hsts_preload"] == "yes"] final_hsts = sum(1 for r in rows if r["hsts_final_header"] and not is_err(r["hsts_final_header"])) redirecting = [r for r in rows if str(r["apex_first_status"]) in ("301", "302", "303", "307", "308")] redirect_with_hsts = sum(1 for r in redirecting if r["hsts_header"]) dmarc = collections.Counter() for r in rows: if is_err(r["dmarc_present"]): dmarc["error"] += 1 elif r["dmarc_present"] == "yes": dmarc[r["dmarc_p"]] += 1 elif r["dmarc_present"] == "no": dmarc["no record"] += 1 else: dmarc[r["dmarc_present"]] += 1 by_category = {} for cat in CATEGORIES: cat_rows = [r for r in rows if r["category"] == cat] by_category[cat] = { "domains": len(cat_rows), "caa_restricting": sum(1 for r in cat_rows if r["caa_issue"] or r["caa_issuewild"]), "dnssec_ds": sum(1 for r in cat_rows if r["ds_present"] == "yes"), "ipv6": sum(1 for r in cat_rows if r["ipv6"] == "yes"), "https_rr": sum(1 for r in cat_rows if r["https_rr_present"] == "yes"), "hsts_apex": sum(1 for r in cat_rows if r["hsts_header"] and not is_err(r["hsts_header"])), "dmarc_reject": sum(1 for r in cat_rows if r["dmarc_p"] == "reject"), } https_rr_rows = [r for r in rows if r["https_rr_present"] == "yes"] ds_rows = [r for r in rows if r["ds_present"] == "yes"] cf_rows = [r for r in rows if r["dns_provider"] == "Cloudflare"] def has_hsts(r): return bool(r["hsts_header"]) and not is_err(r["hsts_header"]) combos = collections.Counter() for r in rows: if not has_hsts(r): combos["no header"] += 1 continue label = "max-age" if r["hsts_include_subdomains"] == "yes": label += " + includeSubDomains" if r["hsts_preload"] == "yes": label += " + preload" combos[label] += 1 preload_rows = [r for r in hsts_rows if r["hsts_preload"] == "yes"] hsts_by_canonical = {} for canon in count_values(rows, "canonical_host"): group = [r for r in rows if r["canonical_host"] == canon] hsts_by_canonical[canon] = { "domains": len(group), "header_on_apex_first_response": sum(1 for r in group if has_hsts(r)), "include_subdomains_on_apex_first_response": sum( 1 for r in group if r["hsts_include_subdomains"] == "yes"), "header_on_final_response": sum(1 for r in group if r["hsts_final_header"] and not is_err(r["hsts_final_header"])), } provider_any = collections.Counter() for r in rows: for label in r["dns_provider"].split(" + "): provider_any[label] += 1 sorted_issue = sorted(len(r["caa_issue"].split()) for r in caa_restricting if r["caa_issue"]) median_issue = sorted_issue[len(sorted_issue) // 2] if sorted_issue else 0 if sorted_issue and len(sorted_issue) % 2 == 0: median_issue = (sorted_issue[len(sorted_issue) // 2 - 1] + sorted_issue[len(sorted_issue) // 2]) / 2 dmarc_rows = [r for r in rows if r["dmarc_present"] == "yes"] return { "survey": SURVEY_ID, "script": "scripts/research/dev-domain-survey-2026-09.py", "raw_data": f"/research/{SURVEY_ID}.csv", **meta, "domains_total": total, "domains_by_category": {c: len(v) for c, v in CATEGORIES.items()}, "domains_with_any_error": sum(1 for r in rows if r["errors"]), "ipv6_aaaa": metric("ipv6"), "caa": { "rrset_present": metric("caa_present"), "restricts_issuance": { "count": len(caa_restricting), "of": total, "percent": pct(len(caa_restricting), total), "note": "RRset contains at least one issue or issuewild property"}, "policy": count_values(rows, "caa_policy"), "issuewild_used": sum(1 for r in caa_restricting if r["caa_issuewild"]), "issuewild_differs_from_issue": sum( 1 for r in caa_restricting if r["caa_issue"] and r["caa_issuewild"] and set(r["caa_issue"].split()) != set(r["caa_issuewild"].split())), "iodef_used": sum(1 for r in caa_rows if r["caa_iodef"] == "yes"), "domains_naming_each_issuer": dict(caa_issuers.most_common()), "issue_identifiers_per_domain": {str(k): v for k, v in sorted(issue_counts.items())}, "median_issue_identifiers": median_issue, }, "dnssec": { "ds_present": metric("ds_present"), "ad_flag_on_a_answer": metric("ad_flag"), "ds_present_and_ad": sum(1 for r in ds_rows if r["ad_flag"] == "yes"), "ds_algorithms": count_values(ds_rows, "ds_algorithms", split=True), "ds_by_dns_provider": count_values(ds_rows, "dns_provider"), }, "https_rr": { **metric("https_rr_present"), "mode": count_values(https_rr_rows, "https_rr_mode"), "alpn_includes_h3": sum(1 for r in https_rr_rows if "h3" in r["https_rr_alpn"]), "alpn_values": count_values(https_rr_rows, "https_rr_alpn"), "by_dns_provider": count_values(https_rr_rows, "dns_provider"), "among_cloudflare_ns": { "domains": len(cf_rows), "with_https_rr": sum(1 for r in cf_rows if r["https_rr_present"] == "yes")}, "among_other_ns": { "domains": total - len(cf_rows), "with_https_rr": sum(1 for r in rows if r["dns_provider"] != "Cloudflare" and r["https_rr_present"] == "yes")}, }, "hsts": { "apex_first_response_has_header": { "count": len(hsts_rows), "of": total, "percent": pct(len(hsts_rows), total), "errors": errs("hsts_header")}, "apex_first_response_4xx_5xx": { "count": len(blocked_first), "domains": [r["domain"] for r in blocked_first], "note": "header on a blocked response may not match what browsers receive"}, "include_subdomains": sum(1 for r in hsts_rows if r["hsts_include_subdomains"] == "yes"), "preload_directive": len(preload_rows), "directive_combinations": dict(combos.most_common()), "preload_ready_header": {"count": len(preload_ready), "domains": preload_ready, "rule": "max-age >= 31536000 + includeSubDomains + preload"}, "preload_directive_but_max_age_below_1_year": sum( 1 for r in preload_rows if not (r["hsts_max_age"].isdigit() and int(r["hsts_max_age"]) >= 31536000)), "preload_directive_without_include_subdomains": sum( 1 for r in preload_rows if r["hsts_include_subdomains"] != "yes"), "no_header_on_apex_first_response_but_on_final": sum( 1 for r in rows if not has_hsts(r) and r["hsts_final_header"] and not is_err(r["hsts_final_header"])), "by_canonical_host": hsts_by_canonical, "max_age_buckets": dict(buckets.most_common()), "max_age_values": dict(ages.most_common()), "final_response_of_apex_chain_has_header": final_hsts, "apex_first_response_is_redirect": len(redirecting), "apex_redirects_carrying_hsts": redirect_with_hsts, }, "canonical_host": count_values(rows, "canonical_host"), "apex_first_status": count_values(rows, "apex_first_status"), "apex_first_status_by_canonical_host": { canon: count_values([r for r in rows if r["canonical_host"] == canon], "apex_first_status") for canon in count_values(rows, "canonical_host")}, "apex_final_status": count_values(rows, "apex_final_status"), "apex_redirects_followed": dict(collections.Counter( str(r["apex_redirect_chain"].count("->")) for r in rows).most_common()), "tls_issuer_org": count_values(rows, "tls_issuer_org"), "tls_issuer_cn": count_values(rows, "tls_issuer_cn"), "tls_version": count_values(rows, "tls_version"), "dns_provider": count_values(rows, "dns_provider"), "dns_provider_any": dict(provider_any.most_common()), "dns_multi_provider": sum(1 for r in rows if " + " in r["dns_provider"]), "mx": {"present": metric("mx_present"), "null_mx": yes("mx_null")}, "spf": {**metric("spf_present"), "all_qualifier": count_values( [r for r in rows if r["spf_present"] == "yes"], "spf_all")}, "dmarc_policy": dict(dmarc.most_common()), "dmarc_details": { "sp_tag": count_values(dmarc_rows, "dmarc_sp"), "pct_tag": count_values(dmarc_rows, "dmarc_pct"), }, "by_category": by_category, } PRELOAD_LIST_URL = ("https://chromium.googlesource.com/chromium/src/+/main/" "net/http/transport_security_state_static.json?format=TEXT") def fetch_preload_list(): """Chromium's HSTS preload list source file (one request; base64 via ?format=TEXT).""" req = urllib.request.Request(PRELOAD_LIST_URL, headers={"User-Agent": USER_AGENT}) with urllib.request.urlopen(req, timeout=120) as resp: raw = base64.b64decode(resp.read()) fetched = now_utc() text = raw.decode("utf-8") body = "\n".join(line for line in text.splitlines() if not line.lstrip().startswith("//")) return json.loads(body)["entries"], fetched, hashlib.sha256(raw).hexdigest() def preload_coverage(rows, entries, fetched, sha256): """Which survey domains a browser using this list would force to HTTPS: an exact entry, or the nearest parent entry with include_subdomains.""" by_name = {e["name"]: e for e in entries} def lookup(domain): if domain in by_name: return domain, by_name[domain] labels = domain.split(".") for i in range(1, len(labels)): parent = ".".join(labels[i:]) entry = by_name.get(parent) if entry and entry.get("include_subdomains"): return parent, entry return None, None covered, exact, via_parent = [], collections.Counter(), [] for r in rows: name, entry = lookup(r["domain"]) if not entry: continue covered.append(r) if name == r["domain"]: exact[entry.get("policy", "")] += 1 else: via_parent.append(f"{r['domain']} (via {name})") covered_names = {r["domain"] for r in covered} sends_preload = [r for r in rows if r["hsts_preload"] == "yes"] return { "source": PRELOAD_LIST_URL.split("?")[0], "fetched_utc": fetched, "sha256_of_file": sha256, "entries_in_list": len(entries), "public_suffix_entries": sum(1 for e in entries if e.get("policy") == "public-suffix"), "survey_domains_covered": len(covered), "covered_by_exact_entry": sum(exact.values()), "exact_entry_policies": dict(exact.most_common()), "covered_via_parent_suffix": via_parent, "sending_preload_directive": len(sends_preload), "sending_preload_directive_and_covered": sum( 1 for r in sends_preload if r["domain"] in covered_names), "covered_but_not_sending_preload_directive": sum( 1 for r in covered if r["hsts_preload"] != "yes"), "matching_rule": "exact name, or nearest parent entry with include_subdomains", } def write_csv(rows, path): path.parent.mkdir(parents=True, exist_ok=True) with open(path, "w", newline="", encoding="utf-8") as fh: writer = csv.DictWriter(fh, fieldnames=CSV_COLUMNS) writer.writeheader() for r in rows: writer.writerow({k: r.get(k, "") for k in CSV_COLUMNS}) def read_csv(path): with open(path, newline="", encoding="utf-8") as fh: return list(csv.DictReader(fh)) def write_json(summary, path): path.parent.mkdir(parents=True, exist_ok=True) with open(path, "w", encoding="utf-8") as fh: json.dump(summary, fh, indent=2, ensure_ascii=False) fh.write("\n") def main(): ap = argparse.ArgumentParser(description=__doc__.split("\n")[0]) ap.add_argument("--csv", type=Path, default=DEFAULT_CSV, help="raw per-domain output (CSV)") ap.add_argument("--json", type=Path, default=DEFAULT_JSON, help="aggregate summary (JSON)") ap.add_argument("--limit", type=int, default=0, help="measure only the first N domains") ap.add_argument("--only", default="", help="comma-separated domains from the list to measure") ap.add_argument("--summarize-only", action="store_true", help="rebuild the JSON summary from the existing CSV (no survey requests)") ap.add_argument("--with-preload-list", action="store_true", help="also fetch Chromium's HSTS preload list (one request to " "chromium.googlesource.com) and add coverage counts to the JSON") args = ap.parse_args() def finish(rows, meta, previous=None): summary = summarize(rows, meta) if args.with_preload_list: entries, fetched, sha256 = fetch_preload_list() summary["hsts_preload_list"] = preload_coverage(rows, entries, fetched, sha256) elif previous and "hsts_preload_list" in previous: # same CSV, same list fetch: keep the earlier coverage counts summary["hsts_preload_list"] = previous["hsts_preload_list"] write_json(summary, args.json) if args.summarize_only: rows = read_csv(args.csv) times = sorted(r["measured_at_utc"] for r in rows) meta = {"first_measurement_utc": times[0], "last_measurement_utc": times[-1]} previous = {} if args.json.exists(): # keep run metadata that the CSV cannot carry with open(args.json, encoding="utf-8") as fh: previous = json.load(fh) for key in ("run_started_utc", "run_finished_utc"): if key in previous: meta[key] = previous[key] meta.update(vantage_point=VANTAGE_POINT, dns_source=DOH_ENDPOINT + " (JSON API)") for key in ("python", "openssl"): if key in previous: meta[key] = previous[key] finish(rows, meta, previous) log(f"summary written to {args.json}") return targets = DOMAINS[: args.limit] if args.limit else DOMAINS if args.only: wanted = {d.strip().lower() for d in args.only.split(",") if d.strip()} targets = [(d, c) for d, c in targets if d in wanted] started = now_utc() log(f"survey start {started}: {len(targets)} domains, pause {PAUSE}s, timeout {TIMEOUT}s") rows = [] for i, (domain, category) in enumerate(targets, 1): t0 = time.time() row = measure(domain, category) rows.append(row) log(f"[{i:3d}/{len(targets)}] {domain:20s} {time.time() - t0:5.1f}s " f"final={row['apex_final_host']} errors={row['errors'] or '-'}") write_csv(rows, args.csv) # checkpoint after every domain finished = now_utc() meta = {"first_measurement_utc": rows[0]["measured_at_utc"] if rows else started, "last_measurement_utc": rows[-1]["measured_at_utc"] if rows else finished, "run_started_utc": started, "run_finished_utc": finished, "vantage_point": VANTAGE_POINT, "dns_source": DOH_ENDPOINT + " (JSON API)", "python": sys.version.split()[0], "openssl": ssl.OPENSSL_VERSION} # Summarize from the CSV as written, so --summarize-only reproduces it exactly. finish(read_csv(args.csv), meta) log(f"survey done {finished}; CSV {args.csv}; JSON {args.json}") if __name__ == "__main__": main()