Add MMDB coverage scan script with anti-poisoning guards

find_unmapped_as_domains.py turns the manual "Checking ASN-domain coverage of the MMDB" recipe into a maintainer script: walk every IPv4 record in the bundled IPinfo Lite MMDB, aggregate routed footprint per as_domain, subtract mapped/known-unknown keys, apply PSL folding and the full-IP privacy filter, and emit domain,ipv4_count,as_name sorted by footprint for the collector -> classifier pipeline. Because ASN registration data is self-declared to the RIRs and as_domain derives from registrant-controlled WHOIS, bulk-categorizing the MMDB needs poisoning defenses: - An IPv4-footprint floor (--min-ips, default 4096, a /20) keeps tiny self-described ASNs out of the auto-classification queue; dropped counts are always printed. - A brand-collision guard in classify_unknown_domains.py loads the existing map (--map) and demotes any single-category candidate whose proposed display name matches an existing map name without a lexical relationship to that operator's keys into the ambiguous bucket (marked name-collision-with-existing-map-entry) for human review. HAND overrides bypass the guard. The guard protects the PTR-side flow as well as the MMDB-coverage flow. Verified: scan yields 132 candidates at the default floor (1512 dropped); collector accepts the output directly; a fixture titled as Comcast under an unrelated domain lands in ambiguous while a comcast-rooted sibling auto-promotes. Also fix the maps README links that still pointed at the root AGENTS.md for the classification workflow after its extraction to maps/AGENTS.md, and correct the classify_tsv docstring's return signature. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-02 13:42:19 +00:00 · 2026-07-16 18:06:58 -04:00
parent cd2029f5c7
commit 387764a716
5 changed files with 357 additions and 38 deletions
@@ -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
@@ -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

@@ -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 `<title>`/`<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

@@ -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,
@@ -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()