diff --git a/.gitignore b/.gitignore index 155f3419..950781ec 100644 --- a/.gitignore +++ b/.gitignore @@ -142,6 +142,7 @@ scratch.py parsedmarc/resources/maps/base_reverse_dns.csv parsedmarc/resources/maps/unknown_base_reverse_dns.csv +parsedmarc/resources/maps/unmapped_as_domains.csv parsedmarc/resources/maps/sus_domains.csv parsedmarc/resources/maps/unknown_domains.txt *.bak diff --git a/parsedmarc/resources/maps/AGENTS.md b/parsedmarc/resources/maps/AGENTS.md index d758b76e..29801464 100644 --- a/parsedmarc/resources/maps/AGENTS.md +++ b/parsedmarc/resources/maps/AGENTS.md @@ -245,28 +245,24 @@ When someone points at a specific domain — from a DMARC report they inspected, ### Checking ASN-domain coverage of the MMDB -Separately from `base_reverse_dns.csv`, the MMDB itself is a source of keys worth mapping. To find ASN domains with high IP weight that don't yet have a map entry, walk every record in `ipinfo_lite.mmdb`, aggregate IPv4 count per `as_domain`, and subtract what's already a map key: +Separately from `base_reverse_dns.csv`, the MMDB itself is a source of keys worth mapping. `find_unmapped_as_domains.py` walks every IPv4 record in `ipinfo_lite.mmdb`, aggregates the routed IPv4 footprint per `as_domain`, and subtracts domains already covered by `base_reverse_dns_map.csv` or `known_unknown_base_reverse_dns.txt`: -```python -import csv, maxminddb -from collections import defaultdict -keys = set() -with open("base_reverse_dns_map.csv", newline="", encoding="utf-8") as f: - for row in csv.DictReader(f): - keys.add(row["base_reverse_dns"].strip().lower()) -v4 = defaultdict(int); names = {} -for net, rec in maxminddb.open_database("parsedmarc/resources/ipinfo/ipinfo_lite.mmdb"): - if net.version != 4 or not isinstance(rec, dict): continue - d = rec.get("as_domain") - if not d: continue - v4[d.lower()] += net.num_addresses - names[d.lower()] = rec.get("as_name", "") -miss = sorted(((d, v4[d], names[d]) for d in v4 if d not in keys), key=lambda x: -x[1]) -for d, c, n in miss[:50]: - print(f"{c:>12,} {d:<30} {n}") +```bash +python find_unmapped_as_domains.py ``` -Apply the same classification rules above (precedence, naming consistency, skip-if-ambiguous, privacy). Many top misses will be brands already in the map under a different rDNS-base key — the goal there is to alias the ASN domain to the same `(name, type)` so both lookup paths hit. For ASN domains with no obvious brand identity (small resellers, parked ASNs), don't map them — the attribution code falls back to the raw `as_name` from the MMDB, which is better than a guess. +This writes `unmapped_as_domains.csv` (`domain,ipv4_count,as_name`, sorted by descending footprint) — an untracked scratch file, not committed. Feed it straight into the existing collector → classifier pipeline: + +```bash +python collect_domain_info.py -i unmapped_as_domains.csv -o /tmp/domain_info.tsv +python classify_unknown_domains.py -i /tmp/domain_info.tsv --map-out /tmp/additions.csv --ku-out /tmp/ku_additions.txt --ambiguous-out /tmp/ambiguous_additions.tsv +``` + +**The `--min-ips` floor (default 4,096, a /20) is an anti-poisoning guard, not a tuning knob to casually override.** ASN registration data is self-declared to the RIRs, and `as_domain` is derived from registrant-controlled WHOIS — anyone can stand up a tiny ASN and self-declare an `as_domain` that impersonates an established brand. A large routed footprint is at least some evidence of a real, long-lived operator; a handful of IPs is cheap for an adversary to acquire. Candidates dropped by the floor are counted and printed, never silently discarded. Raise `--min-ips` for a stricter pass; lowering it below the default should be a deliberate, justified choice, not a default habit. + +**The classifier's brand-collision guard is the second anti-poisoning layer**, and it benefits the PTR-side flow too, not just the MMDB-coverage flow. `classify_unknown_domains.py` loads `base_reverse_dns_map.csv` (via `--map`, defaulting to the bundled map) into a normalized-name index. When a single-category classification proposes a display name that already exists in the map, but the candidate domain has no lexical relationship to any existing key filed under that name (see `_lexically_related`), the row is demoted from the auto-promote (`--map-out`) bucket into `--ambiguous-out` with an `alternatives` marker of `name-collision-with-existing-map-entry`, instead of being silently auto-promoted as if it were the real operator. A human reviewer then decides whether it's a legitimate additional domain for that operator (promote) or an unrelated/impersonating domain (reject to KU or a different category). HAND-dict overrides bypass the guard, since those are already human-forced. + +Apply the same classification rules as the rest of this file (precedence, naming consistency, skip-if-ambiguous, privacy) when reviewing `--map-out` and `--ambiguous-out`. Many top misses will be brands already in the map under a different rDNS-base key — the goal there is to alias the ASN domain to the same `(name, type)` so both lookup paths hit. For ASN domains with no obvious brand identity (small resellers, parked ASNs), don't map them — the attribution code falls back to the raw `as_name` from the MMDB, which is better than a guess. The two-corroborating-sources rule (see the "Workflow for classifying unknown domains" section above) still binds every promotion out of this flow — a high IPv4 footprint and a matching `as_name` alone are not two independent sources. ### Discovering overrides from the live PSL private-domains section diff --git a/parsedmarc/resources/maps/README.md b/parsedmarc/resources/maps/README.md index 6bf6e30b..f9500a06 100644 --- a/parsedmarc/resources/maps/README.md +++ b/parsedmarc/resources/maps/README.md @@ -81,8 +81,9 @@ The file currently contains over 5,000 mappings from a wide variety of email sen `base_reverse_dns_map.csv` is a curated derivative work. Many entries are derived from the bundled IPinfo Lite MMDB (`as_domain` and -`as_name` fields) by walking the database for unmapped operators and -classifying them via the workflow described in [AGENTS.md](../../../AGENTS.md). +`as_name` fields) by walking the database with `find_unmapped_as_domains.py` +for unmapped operators and classifying them via the workflow described in +[AGENTS.md](AGENTS.md). Because IPinfo Lite is licensed under [Creative Commons Attribution-ShareAlike 4.0 (CC BY-SA 4.0)](https://creativecommons.org/licenses/by-sa/4.0/), this CSV is also distributed under **CC BY-SA 4.0** with attribution to @@ -127,6 +128,16 @@ When a `source_name` is not domain-shaped (e.g. `Vodafone Group PLC`), parsedmar Scans `unknown_base_reverse_dns.csv` for full-IP-containing entries that share a common brand suffix. Any suffix repeated by N+ distinct domains (default 3, configurable via `--threshold`) is appended to `psl_overrides.txt`, and every affected entry across the unknown / known-unknown / map files is folded to that suffix's base. Any remaining full-IP entries — whether they clustered or not — are then removed for privacy. After running, the newly exposed base domains still need to be researched and classified via `collect_domain_info.py` and a classifier pass. Supports `--dry-run` to preview without writing. +## find_unmapped_as_domains.py + +Walks every IPv4 record in the bundled `ipinfo_lite.mmdb`, aggregates the routed IPv4 footprint per `as_domain`, and subtracts domains already covered by `base_reverse_dns_map.csv` or `known_unknown_base_reverse_dns.txt`, applying `psl_overrides.txt` folding and the same full-IP privacy filter as `find_unknown_base_reverse_dns.py`. Writes `unmapped_as_domains.csv` (`domain,ipv4_count,as_name`, sorted by descending footprint), which `collect_domain_info.py -i` reads directly. + +Candidates below `--min-ips` (default `4096`, a /20) are dropped as an anti-poisoning guard — ASN registration data is self-declared to the RIRs and `as_domain` comes from registrant-controlled WHOIS, so a tiny ASN is cheap to register under a brand-impersonating domain name. The dropped count is always printed, never silently discarded. + +## unmapped_as_domains.csv + +A CSV file with the fields `domain`, `ipv4_count`, and `as_name`, produced by `find_unmapped_as_domains.py`. This file is not tracked by Git. + ## collect_domain_info.py Bulk enrichment collector. For every domain in `unknown_base_reverse_dns.csv` that is not already in `base_reverse_dns_map.csv`, runs `whois` on the domain, fetches a size-capped `https://` GET, resolves A/AAAA records, and runs `whois` on the first resolved IP. Writes a TSV (`domain_info.tsv` by default) with the registrant org/country/registrar, page ``/`<meta description>`, resolved IPs, and IP-WHOIS org/netname/country — the compact metadata a classifier needs to decide each domain in one pass. Respects `psl_overrides.txt`, skips full-IP entries, and is resume-safe (re-running only fetches domains missing from the output file). @@ -162,7 +173,9 @@ Detectors cover all 44 industry types listed in [base_reverse_dns_map.csv](#base Brand-name selection prefers (in order): the MMDB `as_name` for the domain; the page title's first segment; non-redacted WHOIS registrant org; domain-derived fallback. A `clean_brand` step strips common legal-form suffixes (LLC / GmbH / Ltda / EIRELI / sp. z o.o. / etc.) and prefixes (PT, OOO). When the title has multiple segments separated by `|` / `-` / `—` etc., the segment whose simplified form contains the domain root is preferred — so e.g. accessmontana.com whose `as_name` is "MONTANA WEST, L.L.C." but whose title is "Internet, Phone & TV Bundles | Access Montana" maps to "Access Montana", not "Montana West". -The classifier is the regex baseline of step 4 of the [Workflow for classifying unknown domains](../../../AGENTS.md#workflow-for-classifying-unknown-domains) — it catches obvious cases at scale and leaves only the genuinely ambiguous to manual / LLM review. The empty `HAND` dict at the top of the script is an extension point for batch-specific overrides (e.g. acquisition aliases, brand-name corrections that don't fit any detector); each `domain → ("Brand", "Type")` entry wins over the auto-classifier. +The classifier is the regex baseline of step 4 of the [Workflow for classifying unknown domains](AGENTS.md#workflow-for-classifying-unknown-domains) — it catches obvious cases at scale and leaves only the genuinely ambiguous to manual / LLM review. The empty `HAND` dict at the top of the script is an extension point for batch-specific overrides (e.g. acquisition aliases, brand-name corrections that don't fit any detector); each `domain → ("Brand", "Type")` entry wins over the auto-classifier and bypasses the guard below. + +A brand-collision guard also loads `base_reverse_dns_map.csv` (`--map`, defaulting to the bundled map) so that a candidate whose proposed name matches an *existing* map display name, but whose domain has no lexical relationship to any existing key filed under that name, is demoted from `--map-out` to `--ambiguous-out` (marked `name-collision-with-existing-map-entry`) instead of being auto-promoted. This defends against a low-footprint or brand-impersonating candidate being silently attributed to an established operator, and applies to both the PTR-side and MMDB-coverage flows. ## detect_rebrands.py @@ -175,7 +188,7 @@ Drift sweep that re-fetches every key in `base_reverse_dns_map.csv` with the sam `external_links` is captured into the output for context but is not a default trigger — most outbound links are to partners / customers / vendors and would generate noise. Pass `--flag-external-links` to also flag on this column during a thorough sweep where missing an image-only banner that lacks a rebrand-themed slug or alt text is worse than the noise. -The output is for periodic review, not automated map mutation. Each hit is one corroborating source; promoting a flagged row into the map still requires a second source per the two-corroborating-sources rule in [AGENTS.md](../../../AGENTS.md). Resume-safe: re-running only re-fetches keys not already in the output file. Use `--limit N` to spot-check a slice and `--include-clean` to also write non-flagged rows for inspection of the no-signal majority. +The output is for periodic review, not automated map mutation. Each hit is one corroborating source; promoting a flagged row into the map still requires a second source per the two-corroborating-sources rule in [AGENTS.md](AGENTS.md). Resume-safe: re-running only re-fetches keys not already in the output file. Use `--limit N` to spot-check a slice and `--include-clean` to also write non-flagged rows for inspection of the no-signal majority. ## rebrand_drift.tsv diff --git a/parsedmarc/resources/maps/classify_unknown_domains.py b/parsedmarc/resources/maps/classify_unknown_domains.py index f97ccd77..e3c6e713 100644 --- a/parsedmarc/resources/maps/classify_unknown_domains.py +++ b/parsedmarc/resources/maps/classify_unknown_domains.py @@ -74,12 +74,14 @@ import csv import os import re import sys +from collections import defaultdict import maxminddb # Repo-relative default for the MMDB. _HERE = os.path.dirname(os.path.abspath(__file__)) DEFAULT_MMDB = os.path.normpath(os.path.join(_HERE, "..", "ipinfo", "ipinfo_lite.mmdb")) +DEFAULT_MAP = os.path.normpath(os.path.join(_HERE, "base_reverse_dns_map.csv")) # Per-batch HAND overrides go here. Each entry is: # "domain.example": ("Brand Name", "Type") @@ -9183,6 +9185,58 @@ def _domain_root(domain: str) -> str: return domain.split(".")[0].lower() +def _norm_name(name: str) -> str: + """Collapse a display name to lowercase alphanumerics for comparison.""" + return re.sub(r"[^a-z0-9]", "", name.lower().strip()) + + +def _load_map_names(map_path: str) -> dict[str, set[str]]: + """Return {normalized display name: {existing base_reverse_dns keys}}. + + Used by the brand-collision guard: a candidate whose proposed name + matches an existing map display name, but whose domain isn't lexically + related to any existing key under that name, is a possible brand + impersonation and gets demoted to the ambiguous bucket instead of + auto-promoted. Returns an empty dict (guard silently disabled) if the + map file is missing. + """ + out: dict[str, set[str]] = defaultdict(set) + if not os.path.exists(map_path): + print(f"Warning: {map_path} not found; brand-collision guard disabled") + return {} + with open(map_path, encoding="utf-8", newline="") as f: + for row in csv.DictReader(f): + name = (row.get("name") or "").strip() + domain = (row.get("base_reverse_dns") or "").strip().lower() + if not name or not domain: + continue + out[_norm_name(name)].add(domain) + return dict(out) + + +def _lexically_related(domain: str, name: str, existing_keys: set[str]) -> bool: + """True if `domain` plausibly belongs to the same operator as `name`. + + Checks whether the candidate domain's root appears in the proposed + display name (or vice versa), or whether the candidate shares its + domain root with any existing map key already filed under that name. + """ + root = _domain_root(domain) + root_simple = re.sub(r"[^a-z0-9]", "", root) + name_simple = _norm_name(name) + if root_simple and name_simple: + if root_simple in name_simple: + return True + # Only check the reverse direction (name substring of domain root) + # when the name is long enough to avoid trivial matches like "AT". + if len(name_simple) >= 4 and name_simple in root_simple: + return True + for key in existing_keys: + if _domain_root(key) == root: + return True + return False + + def pick_brand(row: dict, domain: str, as_name: str) -> str: title = fix_text(row.get("title", "").strip()) domain_root = _domain_root(domain) @@ -9735,30 +9789,44 @@ def _load_mmdb_as_names(mmdb_path: str) -> dict: return out -def classify_tsv(input_path: str, mmdb_path: str) -> tuple: +def classify_tsv( + input_path: str, + mmdb_path: str, + map_names: dict[str, set[str]] | None = None, +) -> tuple: """Classify every row of a collect_domain_info.py TSV. - Returns ``(adds, ambiguous, ku, stats)``: + Returns ``(adds, ambiguous, ku, dropped, stats)``: - ``adds`` — ``(domain, name, type)`` tuples where the classifier matched exactly one category and the row can be promoted into the map without review. - ``ambiguous`` — ``(domain, name, primary_type, alternatives, title)`` - rows where two or more distinct detector categories fired. The - classifier won't auto-promote these — the operator-typology question - is "does this domain belong to category A or category B?", and that's - a judgement call the classifier shouldn't make on its own (per - AGENTS.md). The output file is a worklist: a human picks one of the - candidates (or a different category, or rejects the row to KU). + rows where two or more distinct detector categories fired, or where a + single-category match collided with an existing map display name + under an unrelated domain (see the brand-collision guard below). The + classifier won't auto-promote these — a human must pick one of the + candidates (or a different category, or reject the row to KU). - ``ku`` — domains where no detector fired. - ``stats`` — counters. + + ``map_names`` is ``{normalized display name: {existing map keys}}`` as + returned by ``_load_map_names``. When a single-category classification + proposes a name that already exists in the map, but the candidate + domain has no lexical relationship to any existing key filed under that + name (see ``_lexically_related``), the row is demoted from ``adds`` to + ``ambiguous`` instead of being auto-promoted — this guards against a + candidate impersonating an established brand. Rows resolved via the + HAND dict bypass this guard (they're human-forced overrides). """ + if map_names is None: + map_names = {} asn = _load_mmdb_as_names(mmdb_path) adds: list = [] ambiguous: list = [] ku: list = [] dropped: list = [] - auto = hand = ambig = 0 + auto = hand = ambig = name_collision = 0 with open(input_path, encoding="utf-8", newline="") as f: reader = csv.DictReader(f, delimiter="\t") for row in reader: @@ -9793,11 +9861,33 @@ def classify_tsv(input_path: str, mmdb_path: str) -> tuple: # to remove the domain from KU if it's currently there. dropped.append(domain) elif len(r) == 2: - adds.append((domain, r[0], r[1])) - auto += 1 - if link_target and link_target != domain: - adds.append((link_target, r[0], r[1])) + name, category = r + existing_keys = map_names.get(_norm_name(name)) + if existing_keys and not _lexically_related( + domain, name, existing_keys + ): + # Proposed name collides with an established map brand + # but the domain isn't lexically related to it — treat + # as a possible impersonation and route to human review + # instead of auto-promoting. + title = (row.get("title") or "").strip() + ambiguous.append( + ( + domain, + name, + category, + ["name-collision-with-existing-map-entry"], + title, + ) + ) + ambig += 1 + name_collision += 1 + else: + adds.append((domain, name, category)) auto += 1 + if link_target and link_target != domain: + adds.append((link_target, name, category)) + auto += 1 else: # (brand, primary, alternatives) — multi-category match. title = (row.get("title") or "").strip() @@ -9819,6 +9909,7 @@ def classify_tsv(input_path: str, mmdb_path: str) -> tuple: "ambig": ambig, "ku": len(ku), "dropped": len(dropped), + "name_collision": name_collision, }, ) @@ -9868,9 +9959,20 @@ def main(): default=DEFAULT_MMDB, help="Path to ipinfo_lite.mmdb. Default: bundled MMDB", ) + p.add_argument( + "--map", + default=DEFAULT_MAP, + help=( + "Path to base_reverse_dns_map.csv, used for the brand-collision " + "guard (a proposed name matching an existing map entry under an " + "unrelated domain is routed to ambiguous instead of " + "auto-promoted). Default: bundled map" + ), + ) args = p.parse_args() - adds, ambiguous, ku, dropped, stats = classify_tsv(args.input, args.mmdb) + map_names = _load_map_names(args.map) + adds, ambiguous, ku, dropped, stats = classify_tsv(args.input, args.mmdb, map_names) with open(args.map_out, "w", encoding="utf-8", newline="") as f: w = csv.writer(f, lineterminator="\r\n") @@ -9891,7 +9993,7 @@ def main(): print( f"auto: {stats['auto']}, hand: {stats['hand']}, " - f"ambig: {stats['ambig']}, " + f"ambig: {stats['ambig']} (name_collision: {stats['name_collision']}), " f"ku: {stats['ku']} (unique: {len(set(ku))}), " f"dropped: {stats['dropped']}", file=sys.stderr, diff --git a/parsedmarc/resources/maps/find_unmapped_as_domains.py b/parsedmarc/resources/maps/find_unmapped_as_domains.py new file mode 100644 index 00000000..2b7aa067 --- /dev/null +++ b/parsedmarc/resources/maps/find_unmapped_as_domains.py @@ -0,0 +1,207 @@ +#!/usr/bin/env python +"""Find ASN domains in the bundled MMDB that aren't in base_reverse_dns_map.csv. + +Walks every IPv4 record in the bundled IPinfo Lite MMDB +(``../ipinfo/ipinfo_lite.mmdb``), aggregates the routed IPv4 footprint per +``as_domain``, and subtracts domains already present in +``base_reverse_dns_map.csv`` or ``known_unknown_base_reverse_dns.txt``. The +remaining candidates are ASN-fallback lookup keys with no map coverage yet. + +Candidates below ``--min-ips`` (default 4096, i.e. a /20) are dropped. This +floor exists because ASN registration data is self-declared to the RIRs and +``as_domain`` is derived from registrant-controlled WHOIS, so a tiny ASN is +cheap for an adversary to register under an impersonating domain — a large +routed footprint is at least some evidence of a real, long-lived operator. + +Output feeds ``collect_domain_info.py -i`` directly (its ``domain`` header is +read by ``_load_input_domains``), which in turn feeds +``classify_unknown_domains.py``. See AGENTS.md's "Checking ASN-domain +coverage of the MMDB" section for the full workflow. +""" + +import argparse +import csv +import os +import re +import sys +from collections import defaultdict + + +# Privacy filter: an as_domain containing a full IPv4 address (four dotted +# or dashed octets) reveals a specific customer IP. Such entries are dropped +# here so they never enter the map or the known-unknown list. Mirrors +# find_unknown_base_reverse_dns.py's _FULL_IP_RE/_has_full_ip. +_FULL_IP_RE = re.compile( + r"(?<![\d])(\d{1,3})[-.](\d{1,3})[-.](\d{1,3})[-.](\d{1,3})(?![\d])" +) + + +def _has_full_ip(s: str) -> bool: + for m in _FULL_IP_RE.finditer(s): + octets = [int(g) for g in m.groups()] + if all(0 <= o <= 255 for o in octets): + return True + return False + + +def _load_as_domain_footprints(mmdb_path: str) -> dict[str, tuple[int, str]]: + """Return {as_domain.lower(): (ipv4_count, as_name)}. + + Aggregates ``net.num_addresses`` per lowercased/stripped ``as_domain`` + across every IPv4 record. When an as_domain appears under more than one + as_name (uncommon), the as_name carrying the largest aggregate footprint + wins. + """ + try: + import maxminddb + except ImportError: + print( + "Error: maxminddb is required to walk the MMDB; " + "install parsedmarc's runtime dependencies (pip install maxminddb)", + file=sys.stderr, + ) + sys.exit(1) + + counts: dict[tuple[str, str], int] = defaultdict(int) + with maxminddb.open_database(mmdb_path) as reader: + for net, rec in reader: + if net.version != 4 or not isinstance(rec, dict): + continue + as_domain = rec.get("as_domain") + if not as_domain: + continue + as_domain = as_domain.lower().strip() + as_name = (rec.get("as_name") or "").strip() + counts[(as_domain, as_name)] += net.num_addresses + + totals: dict[str, int] = defaultdict(int) + for (as_domain, _as_name), count in counts.items(): + totals[as_domain] += count + + best_name: dict[str, tuple[str, int]] = {} + for (as_domain, as_name), count in counts.items(): + existing = best_name.get(as_domain) + if existing is None or count > existing[1]: + best_name[as_domain] = (as_name, count) + + return {d: (totals[d], best_name[d][0]) for d in totals} + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "Find ASN domains in the bundled MMDB with no " + "base_reverse_dns_map.csv coverage yet, ranked by routed IPv4 " + "footprint. Output feeds collect_domain_info.py -i." + ) + ) + parser.add_argument( + "--mmdb", + default="../ipinfo/ipinfo_lite.mmdb", + help="Path to ipinfo_lite.mmdb (default: %(default)s)", + ) + parser.add_argument( + "--min-ips", + type=int, + default=4096, + help=( + "Minimum aggregate IPv4 footprint (default: %(default)s, a " + "/20) below which a candidate is dropped as anti-poisoning " + "protection against self-declared ASN registration data" + ), + ) + parser.add_argument( + "-o", + "--output", + default="unmapped_as_domains.csv", + help="Path to write the output CSV to (default: %(default)s)", + ) + return parser.parse_args() + + +def _main(): + args = _parse_args() + + base_reverse_dns_map_file_path = "base_reverse_dns_map.csv" + known_unknown_list_file_path = "known_unknown_base_reverse_dns.txt" + psl_overrides_file_path = "psl_overrides.txt" + + known_domains: set[str] = set() + known_unknown_domains: set[str] = set() + psl_overrides: list[str] = [] + + def load_list(file_path, list_var): + if not os.path.exists(file_path): + print(f"Error: {file_path} does not exist") + sys.exit(1) + print(f"Loading {file_path}") + with open(file_path) as f: + for line in f.readlines(): + domain = line.lower().strip() + if domain != "": + list_var.append(domain) + + if not os.path.exists(base_reverse_dns_map_file_path): + print(f"Error: {base_reverse_dns_map_file_path} does not exist") + sys.exit(1) + print(f"Loading {base_reverse_dns_map_file_path}") + with open(base_reverse_dns_map_file_path, newline="", encoding="utf-8") as f: + for row in csv.DictReader(f): + known_domains.add(row["base_reverse_dns"].lower().strip()) + + known_unknown_list: list[str] = [] + load_list(known_unknown_list_file_path, known_unknown_list) + known_unknown_domains.update(known_unknown_list) + + load_list(psl_overrides_file_path, psl_overrides) + + if not os.path.exists(args.mmdb): + print(f"Error: {args.mmdb} does not exist") + sys.exit(1) + print(f"Loading {args.mmdb}") + footprints = _load_as_domain_footprints(args.mmdb) + print(f"Indexed {len(footprints)} as_domains from the MMDB") + + below_floor = 0 + output_rows = [] + for domain, (count, as_name) in footprints.items(): + for psl_domain in psl_overrides: + if domain.endswith(psl_domain): + domain = psl_domain.strip(".").strip("-") + break + if _has_full_ip(domain): + continue + if domain in known_domains or domain in known_unknown_domains: + continue + if count < args.min_ips: + below_floor += 1 + continue + output_rows.append((domain, count, as_name)) + + # A PSL fold can merge multiple as_domains onto the same base domain; + # keep the row with the larger footprint for each resulting key. + merged: dict[str, tuple[int, str]] = {} + for domain, count, as_name in output_rows: + existing = merged.get(domain) + if existing is None or count > existing[0]: + merged[domain] = (count, as_name) + + output_rows = sorted( + ((d, c, n) for d, (c, n) in merged.items()), + key=lambda r: -r[1], + ) + + print( + f"{len(output_rows)} candidate(s) at or above the {args.min_ips:,} " + f"IPv4 floor; {below_floor} below-floor candidate(s) dropped" + ) + print(f"Writing {args.output}") + with open(args.output, "w", newline="", encoding="utf-8") as f: + writer = csv.writer(f) + writer.writerow(["domain", "ipv4_count", "as_name"]) + for domain, count, as_name in output_rows: + writer.writerow([domain, count, as_name]) + + +if __name__ == "__main__": + _main()