bbdiscover.py

#!/usr/bin/env python3
"""
bbdiscover - self-hosted bug bounty / VDP program discovery

Finds vulnerability disclosure and bug bounty programs that companies run
themselves, rather than through HackerOne / Bugcrowd / Intigriti / YesWeHack.

Sources:
  security_txt  crawl a domain list for /.well-known/security.txt (RFC 9116)
  diodb         disclose.io open program database (JSON)
  urlscan       urlscan.io search API, date sorted
  gcse          Google Programmable Search JSON API (optional, needs key + cx)

State lives in SQLite so every run only reports what is genuinely new.
Output goes to Google Sheets, CSV, and XLSX.

Author: Aay Kay
"""

import argparse
import asyncio
import csv
import json
import os
import re
import sqlite3
import sys
import time
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlparse

import httpx
import yaml

# --------------------------------------------------------------------------
# Configuration defaults
# --------------------------------------------------------------------------

BASE = Path(os.environ.get("BBD_HOME", Path.home() / "bbdiscover"))
DB_PATH = BASE / "state.db"
OUT_DIR = BASE / "out"
LIST_DIR = BASE / "lists"

UA = "bbdiscover/1.0 (security research; contact: imsyedkarim@gmail.com)"

TRANCO_URL = "https://tranco-list.eu/top-1m.csv.zip"
DIODB_URL = "https://raw.githubusercontent.com/disclose/diodb/master/program-list.json"
URLSCAN_SEARCH = "https://urlscan.io/api/v1/search/"

# Anything whose policy or contact points here is platform hosted, not self hosted.
PLATFORM_HOSTS = re.compile(
    r"(hackerone\.com|bugcrowd\.com|intigriti\.com|yeswehack\.com|synack\.com|"
    r"openbugbounty\.org|hackenproof\.com|immunefi\.com|zerocopter\.com|"
    r"cobalt\.io|federacy\.com|huntr\.dev|huntr\.com|safehats\.com|"
    r"bugbase\.ai|antihack\.me|hackrate\.io)",
    re.I,
)

# Signals that money (not just kudos) is on the table.
REWARD_MONETARY = re.compile(
    r"\b(bounty|bounties|monetary|cash|payout|paid|we pay|will pay|"
    r"compensat\w+|remunerat\w+|honorarium|reward(s|ed)?\s+(of|up to|range|"
    r"start|from|between)|\$\s?\d|usd\s?\d|eur\s?\d)\b",
    re.I,
)
REWARD_NONMONETARY = re.compile(
    r"\b(hall of fame|acknowledg\w+|swag|t-?shirt|sticker|thank you|"
    r"recognition|credit(ed)?)\b",
    re.I,
)
SAFE_HARBOR = re.compile(
    r"\b(safe harbou?r|good faith|will not (pursue|initiate|recommend|take|"
    r"bring)\s+(legal|civil|criminal)|authorized under|not (be )?considered "
    r"a violation|no legal action|exempt(ion)? from)\b",
    re.I,
)
SCOPE_HINT = re.compile(r"\b(in[- ]scope|out of scope|scope|eligible (domains|assets|targets))\b", re.I)

PROGRAM_URL_HINT = re.compile(
    r"(bug[-_ ]?bounty|responsible[-_ ]?disclosure|vulnerability[-_ ]?disclosure|"
    r"vuln[-_ ]?report|security\.txt|report[-_ ]a[-_ ]vulnerability|psirt|/vdp)",
    re.I,
)


# --------------------------------------------------------------------------
# Record model
# --------------------------------------------------------------------------

@dataclass
class Program:
    domain: str
    org: str = ""
    source: str = ""
    policy_url: str = ""
    contact: str = ""
    ack_url: str = ""
    hiring_url: str = ""
    expires: str = ""
    langs: str = ""
    self_hosted: str = "unknown"     # yes / no / unknown
    program_type: str = "unknown"    # BBP / VDP / unknown
    safe_harbor: str = "unknown"     # full / partial / none / unclear
    reward_signal: str = ""
    launch_date: str = ""
    disclosure: str = ""             # coordinated / nda / discretionary / no
    policy_status: str = ""
    score: int = 0
    first_seen: str = ""
    last_seen: str = ""
    notes: str = ""

    def key(self) -> str:
        return self.domain.lower().strip()


# --------------------------------------------------------------------------
# security.txt parsing (RFC 9116)
# --------------------------------------------------------------------------

def truthy(v):
    """diodb mixes real bools with the strings yes/no/''. Returns True/False/None."""
    if isinstance(v, bool):
        return v
    s = str(v or "").strip().lower()
    if s in {"yes", "true", "1"}:
        return True
    if s in {"no", "false", "0"}:
        return False
    return None


# Subdomains that are program infrastructure rather than the org's real domain.
_STRIP_PREFIX = re.compile(r"^(www|security|bounty|bugbounty|bug-bounty|vdp|psirt|disclosure)\.", re.I)
# Helpdesk hosts: the netloc is the vendor, not the customer.
_HELPDESK = re.compile(r"(zendesk\.com|freshdesk\.com|helpscoutdocs\.com|intercom\.help|"
                       r"notion\.site|gitbook\.io|atlassian\.net|hubspot\.com)$", re.I)


def normalize_domain(netloc: str) -> str:
    d = (netloc or "").split(":")[0].strip().lower()
    if not d or "." not in d:
        return ""
    prev = None
    while prev != d:
        prev = d
        d = _STRIP_PREFIX.sub("", d)
    return d


def parse_security_txt(body: str) -> dict:
    """Parse an RFC 9116 file into a dict of lowercase field -> list of values."""
    fields: dict = {}
    for raw in body.splitlines():
        line = raw.strip()
        if not line or line.startswith("#"):
            continue
        if ":" not in line:
            continue
        name, _, value = line.partition(":")
        name = name.strip().lower()
        value = value.strip()
        if not name or not value:
            continue
        # Ignore the PGP signature block wrapper
        if name in {"hash", "version", "-----begin pgp signed message-----"}:
            continue
        fields.setdefault(name, []).append(value)
    return fields


def is_expired(expires_value: str) -> bool:
    if not expires_value:
        return False
    for fmt in ("%Y-%m-%dT%H:%M:%S.%fZ", "%Y-%m-%dT%H:%M:%SZ", "%Y-%m-%dT%H:%M:%S%z", "%Y-%m-%d"):
        try:
            dt = datetime.strptime(expires_value.strip(), fmt)
            if dt.tzinfo is None:
                dt = dt.replace(tzinfo=timezone.utc)
            return dt < datetime.now(timezone.utc)
        except ValueError:
            continue
    return False


# --------------------------------------------------------------------------
# Collector: security.txt sweep
# --------------------------------------------------------------------------

async def fetch_security_txt(client: httpx.AsyncClient, domain: str, sem: asyncio.Semaphore):
    """Try the RFC path first, then the legacy root path, then www."""
    candidates = [
        f"https://{domain}/.well-known/security.txt",
        f"https://www.{domain}/.well-known/security.txt",
        f"https://{domain}/security.txt",
    ]
    async with sem:
        for url in candidates:
            try:
                r = await client.get(url, timeout=8.0, follow_redirects=True)
            except Exception:
                continue
            if r.status_code != 200:
                continue
            ctype = r.headers.get("content-type", "")
            if "html" in ctype.lower():
                continue
            body = r.text[:20000]
            if "contact:" not in body.lower():
                continue
            return domain, body
    return domain, None


def program_from_security_txt(domain: str, body: str) -> Program:
    f = parse_security_txt(body)
    contact = "; ".join(f.get("contact", []))[:500]
    policy = (f.get("policy") or [""])[0]
    ack = (f.get("acknowledgments") or f.get("acknowledgements") or [""])[0]
    hiring = (f.get("hiring") or [""])[0]
    expires = (f.get("expires") or [""])[0]
    langs = "; ".join(f.get("preferred-languages", []))

    blob = f"{contact} {policy} {ack}"
    self_hosted = "no" if PLATFORM_HOSTS.search(blob) else "yes"

    p = Program(
        domain=normalize_domain(domain) or domain,
        source="security_txt",
        policy_url=policy,
        contact=contact,
        ack_url=ack,
        hiring_url=hiring,
        expires=expires,
        langs=langs,
        self_hosted=self_hosted,
    )
    if is_expired(expires):
        p.notes = "security.txt Expires field is in the past"
    return p


async def collect_security_txt(domains, concurrency: int, verbose: bool):
    sem = asyncio.Semaphore(concurrency)
    limits = httpx.Limits(max_connections=concurrency, max_keepalive_connections=concurrency)
    results = []
    async with httpx.AsyncClient(headers={"User-Agent": UA}, limits=limits, verify=False) as client:
        tasks = [fetch_security_txt(client, d, sem) for d in domains]
        done = 0
        for coro in asyncio.as_completed(tasks):
            domain, body = await coro
            done += 1
            if verbose and done % 500 == 0:
                print(f"  [security_txt] {done}/{len(domains)} checked, {len(results)} hits", file=sys.stderr)
            if body:
                results.append(program_from_security_txt(domain, body))
    return results


# --------------------------------------------------------------------------
# Collector: disclose.io diodb
# --------------------------------------------------------------------------

def collect_diodb() -> list:
    out = []
    try:
        r = httpx.get(DIODB_URL, headers={"User-Agent": UA}, timeout=30.0, follow_redirects=True)
        r.raise_for_status()
        data = r.json()
    except Exception as e:
        print(f"[!] diodb fetch failed: {e}", file=sys.stderr)
        return out

    entries = data if isinstance(data, list) else data.get("programs", [])
    for e in entries:
        if not isinstance(e, dict):
            continue
        policy = (e.get("policy_url") or "").strip()
        contact = (e.get("contact_url") or e.get("contact_email") or "").strip()
        domain = normalize_domain(urlparse(policy).netloc) if policy else ""
        if not domain:
            continue

        # diodb stores booleans as the strings "yes"/"no"/"" and sometimes real bools.
        bounty = truthy(e.get("offers_bounty"))
        swag = truthy(e.get("offers_swag"))
        hof = (e.get("hall_of_fame") or "").strip()
        safe = (e.get("safe_harbor") or "").strip().lower()

        blob = f"{policy} {contact}"
        self_hosted = "no" if PLATFORM_HOSTS.search(blob) else "yes"

        if bounty is True:
            ptype = "BBP"
        elif bounty is False:
            ptype = "VDP"
        else:
            ptype = "unknown"

        signals = []
        if bounty is True:
            signals.append("bounty")
        if swag is True:
            signals.append("swag")
        if hof:
            signals.append("hall of fame")

        p = Program(
            domain=domain,
            org=(e.get("program_name") or "").strip(),
            source="diodb",
            policy_url=policy,
            contact=contact,
            ack_url=hof,
            hiring_url=(e.get("hiring") or "").strip(),
            langs=(e.get("preferred_languages") or "").strip(),
            self_hosted=self_hosted,
            program_type=ptype,
            safe_harbor=safe if safe in {"full", "partial", "none"} else "unclear",
            reward_signal=", ".join(signals),
            launch_date=(e.get("launch_date") or "").strip(),
            disclosure=(e.get("public_disclosure") or "").strip(),
            policy_status=(e.get("policy_url_status") or "").strip(),
        )
        if (e.get("securitytxt_url") or "").strip():
            p.notes = "has security.txt"
        out.append(p)
    return out


# --------------------------------------------------------------------------
# Collector: urlscan.io
# --------------------------------------------------------------------------

def collect_urlscan(api_key: str, pages: int = 5) -> list:
    out = []
    if not api_key:
        return out
    headers = {"User-Agent": UA, "API-Key": api_key}
    query = 'page.url:"/.well-known/security.txt"'
    search_after = None
    for _ in range(pages):
        params = {"q": query, "size": 100}
        if search_after:
            params["search_after"] = search_after
        try:
            r = httpx.get(URLSCAN_SEARCH, params=params, headers=headers, timeout=30.0)
            r.raise_for_status()
            data = r.json()
        except Exception as e:
            print(f"[!] urlscan query failed: {e}", file=sys.stderr)
            break
        results = data.get("results", [])
        if not results:
            break
        for item in results:
            dom = normalize_domain(item.get("page", {}).get("domain") or "")
            if not dom:
                continue
            out.append(Program(
                domain=dom,
                source="urlscan",
                policy_url=item.get("page", {}).get("url", ""),
                notes=f"urlscan seen {item.get('task', {}).get('time', '')}",
            ))
        if not data.get("has_more"):
            break
        search_after = ",".join(str(x) for x in results[-1].get("sort", []))
        time.sleep(2)
    return out


# --------------------------------------------------------------------------
# Collector: Google Programmable Search
# --------------------------------------------------------------------------

GCSE_DORKS = [
    'inurl:bug-bounty intext:"scope"',
    'inurl:responsible-disclosure intext:"safe harbor"',
    'inurl:vulnerability-disclosure-policy',
    'intitle:"security acknowledgements" OR intitle:"hall of fame" intext:researchers',
    'intext:"bug bounty program" intext:"in scope" intext:"reward"',
    'inurl:"/security/report-a-vulnerability"',
]


def collect_gcse(api_key: str, cx: str, date_restrict: str = "m1") -> list:
    out = []
    if not (api_key and cx):
        return out
    for dork in GCSE_DORKS:
        for start in (1, 11, 21, 31, 41):
            params = {
                "key": api_key, "cx": cx, "q": dork,
                "dateRestrict": date_restrict, "sort": "date",
                "start": start, "num": 10,
            }
            try:
                r = httpx.get("https://www.googleapis.com/customsearch/v1",
                              params=params, timeout=20.0)
                if r.status_code != 200:
                    break
                items = r.json().get("items", [])
            except Exception:
                break
            if not items:
                break
            for it in items:
                link = it.get("link", "")
                dom = normalize_domain(urlparse(link).netloc)
                if not dom or PLATFORM_HOSTS.search(link):
                    continue
                out.append(Program(
                    domain=dom, source="gcse", policy_url=link,
                    org=it.get("title", "")[:120],
                    notes=f"dork: {dork[:60]}",
                ))
            time.sleep(0.5)
    return out


# --------------------------------------------------------------------------
# Enrichment: fetch the policy page and classify it
# --------------------------------------------------------------------------

async def enrich_one(client, p: Program, sem):
    if not p.policy_url or not p.policy_url.startswith("http"):
        # security.txt with a Contact but no Policy: still a lead, just weaker.
        if p.safe_harbor == "unknown":
            p.safe_harbor = "unclear"
        if not p.notes:
            p.notes = "no Policy field, contact-only"
        return p
    async with sem:
        try:
            r = await client.get(p.policy_url, timeout=12.0, follow_redirects=True)
            p.policy_status = str(r.status_code)
            if r.status_code != 200:
                return p
            text = re.sub(r"<script.*?</script>|<style.*?</style>", " ",
                          r.text[:200000], flags=re.S | re.I)
            text = re.sub(r"<[^>]+>", " ", text)
            text = re.sub(r"\s+", " ", text)
        except Exception as e:
            p.policy_status = f"err:{type(e).__name__}"
            return p

    monetary = bool(REWARD_MONETARY.search(text))
    nonmonetary = bool(REWARD_NONMONETARY.search(text))

    if p.program_type == "unknown":
        p.program_type = "BBP" if monetary else ("VDP" if nonmonetary or SCOPE_HINT.search(text) else "unknown")

    signals = []
    if monetary:
        signals.append("monetary")
    if nonmonetary:
        signals.append("recognition/swag")
    if signals and not p.reward_signal:
        p.reward_signal = ", ".join(signals)

    if p.safe_harbor == "unknown":
        p.safe_harbor = "yes" if SAFE_HARBOR.search(text) else "unclear"

    if p.self_hosted == "unknown":
        p.self_hosted = "no" if PLATFORM_HOSTS.search(text[:5000]) else "yes"

    return p


async def enrich_all(programs, concurrency: int):
    sem = asyncio.Semaphore(concurrency)
    limits = httpx.Limits(max_connections=concurrency)
    async with httpx.AsyncClient(headers={"User-Agent": UA}, limits=limits, verify=False) as client:
        return await asyncio.gather(*[enrich_one(client, p, sem) for p in programs])


# --------------------------------------------------------------------------
# Scoring
# --------------------------------------------------------------------------

def score(p: Program) -> int:
    """Higher = more worth your time. Self-hosted + pays + safe harbor + reachable."""
    s = 0
    if p.self_hosted == "yes":
        s += 3
    if p.program_type == "BBP":
        s += 4
    elif p.program_type == "VDP":
        s += 1
    if re.search(r"bounty|monetary", p.reward_signal, re.I):
        s += 2
    if p.safe_harbor == "full":
        s += 3
    elif p.safe_harbor == "partial":
        s += 1
    elif p.safe_harbor == "none":
        s -= 2
    if p.ack_url:
        s += 1
    if p.hiring_url:
        s += 1                      # staffed security team, triage tends to be faster
    if p.policy_status in {"200", "alive"}:
        s += 1
    elif p.policy_status in {"404", "dead"}:
        s -= 4
    if p.expires and is_expired(p.expires):
        s -= 1
    if _HELPDESK.search(p.domain):
        s -= 2                      # netloc is the helpdesk vendor, not the org
    if p.disclosure.lower() == "nda":
        s -= 1
    return s


# --------------------------------------------------------------------------
# State (SQLite)
# --------------------------------------------------------------------------

SCHEMA = """
CREATE TABLE IF NOT EXISTS programs (
    domain TEXT PRIMARY KEY, org TEXT, source TEXT, policy_url TEXT,
    contact TEXT, ack_url TEXT, hiring_url TEXT, expires TEXT, langs TEXT,
    self_hosted TEXT, program_type TEXT, safe_harbor TEXT, reward_signal TEXT,
    launch_date TEXT, disclosure TEXT, policy_status TEXT, score INTEGER,
    first_seen TEXT, last_seen TEXT, notes TEXT
);
CREATE INDEX IF NOT EXISTS idx_first_seen ON programs(first_seen);
CREATE INDEX IF NOT EXISTS idx_score ON programs(score);
"""


def db_connect():
    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    con = sqlite3.connect(DB_PATH)
    con.executescript(SCHEMA)
    return con


def upsert(con, programs):
    """Insert new programs, refresh existing ones. Returns only the new records."""
    now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
    new = []
    cur = con.cursor()
    for p in programs:
        k = p.key()
        cur.execute("SELECT first_seen FROM programs WHERE domain=?", (k,))
        row = cur.fetchone()
        p.score = score(p)
        p.last_seen = now
        if row:
            p.first_seen = row[0]
        else:
            p.first_seen = now
            new.append(p)
        d = asdict(p)
        d["domain"] = k
        cur.execute(
            """INSERT INTO programs VALUES
               (:domain,:org,:source,:policy_url,:contact,:ack_url,:hiring_url,
                :expires,:langs,:self_hosted,:program_type,:safe_harbor,
                :reward_signal,:launch_date,:disclosure,:policy_status,:score,
                :first_seen,:last_seen,:notes)
               ON CONFLICT(domain) DO UPDATE SET
                 org=COALESCE(NULLIF(excluded.org,''),org),
                 policy_url=COALESCE(NULLIF(excluded.policy_url,''),policy_url),
                 contact=COALESCE(NULLIF(excluded.contact,''),contact),
                 ack_url=COALESCE(NULLIF(excluded.ack_url,''),ack_url),
                 program_type=excluded.program_type,
                 safe_harbor=excluded.safe_harbor,
                 self_hosted=excluded.self_hosted,
                 reward_signal=excluded.reward_signal,
                 launch_date=COALESCE(NULLIF(excluded.launch_date,''),launch_date),
                 disclosure=COALESCE(NULLIF(excluded.disclosure,''),disclosure),
                 policy_status=excluded.policy_status,
                 score=excluded.score,
                 last_seen=excluded.last_seen,
                 source=source||','||excluded.source""",
            d,
        )
    con.commit()
    return new


def dedupe(programs):
    """Merge records for the same domain, preferring the richest one."""
    best = {}
    for p in programs:
        k = p.key()
        if k not in best:
            best[k] = p
            continue
        cur = best[k]
        for f in ("org", "policy_url", "contact", "ack_url", "hiring_url",
                  "expires", "langs", "reward_signal", "launch_date",
                  "disclosure", "policy_status"):
            if not getattr(cur, f) and getattr(p, f):
                setattr(cur, f, getattr(p, f))
        if cur.self_hosted == "unknown":
            cur.self_hosted = p.self_hosted
        if cur.program_type == "unknown":
            cur.program_type = p.program_type
        if p.source not in cur.source:
            cur.source = f"{cur.source},{p.source}"
    return list(best.values())


# --------------------------------------------------------------------------
# Output sinks
# --------------------------------------------------------------------------

HEADERS = ["domain", "org", "program_type", "self_hosted", "safe_harbor",
           "reward_signal", "score", "launch_date", "policy_url", "contact",
           "ack_url", "hiring_url", "disclosure", "expires", "langs",
           "policy_status", "source", "first_seen", "last_seen", "notes"]


def rows_from_db_recent(con, days=30, self_hosted_only=True):
    q = ("SELECT " + ",".join(HEADERS) + " FROM programs "
         "WHERE first_seen >= datetime('now', ?) ")
    args = [f"-{int(days)} days"]
    if self_hosted_only:
        q += " AND self_hosted = 'yes'"
    q += " ORDER BY score DESC, first_seen DESC"
    return con.execute(q, args).fetchall()


def rows_from_db(con, self_hosted_only=True, min_score=0):
    q = "SELECT " + ",".join(HEADERS) + " FROM programs WHERE score >= ?"
    args = [min_score]
    if self_hosted_only:
        q += " AND self_hosted = 'yes'"
    q += " ORDER BY first_seen DESC, score DESC"
    return con.execute(q, args).fetchall()


def write_csv(rows, path):
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w", newline="", encoding="utf-8") as fh:
        w = csv.writer(fh)
        w.writerow(HEADERS)
        w.writerows(rows)
    return path


def write_sheet(rows, sheet_id, worksheet, creds_path):
    """Full refresh of a Google Sheet tab. Requires gspread + a service account."""
    try:
        import gspread
        from google.oauth2.service_account import Credentials
    except ImportError:
        print("[!] gspread not installed, skipping Sheets. pip install gspread google-auth",
              file=sys.stderr)
        return False
    scopes = ["https://www.googleapis.com/auth/spreadsheets"]
    creds = Credentials.from_service_account_file(creds_path, scopes=scopes)
    gc = gspread.authorize(creds)
    sh = gc.open_by_key(sheet_id)
    try:
        ws = sh.worksheet(worksheet)
    except Exception:
        ws = sh.add_worksheet(title=worksheet, rows=2000, cols=len(HEADERS))
    ws.clear()
    ws.update(values=[HEADERS] + [list(r) for r in rows], range_name="A1")
    ws.freeze(rows=1)
    ws.format("A1:T1", {"textFormat": {"bold": True}})
    return True


def notify(webhook: str, new_programs):
    if not webhook or not new_programs:
        return
    top = sorted(new_programs, key=lambda p: -p.score)[:20]
    lines = [f"*{len(new_programs)} new self-hosted programs*"]
    for p in top:
        lines.append(f"[{p.score}] {p.domain} | {p.program_type} | SH:{p.safe_harbor} | {p.policy_url[:70]}")
    payload = {"text": "\n".join(lines)[:3900]}
    try:
        httpx.post(webhook, json=payload, timeout=15.0)
    except Exception as e:
        print(f"[!] notify failed: {e}", file=sys.stderr)


# --------------------------------------------------------------------------
# Domain list
# --------------------------------------------------------------------------

def load_domains(path: Path, limit=None):
    if not path.exists():
        print(f"[!] Domain list not found: {path}", file=sys.stderr)
        print("    Get one:  curl -sL https://tranco-list.eu/top-1m.csv.zip -o t.zip && "
              "unzip -p t.zip | cut -d, -f2 > lists/tranco.txt", file=sys.stderr)
        sys.exit(1)
    doms = []
    with open(path, encoding="utf-8", errors="ignore") as fh:
        for line in fh:
            d = line.strip().split(",")[-1].strip().lower()
            if d and "." in d:
                doms.append(d)
            if limit and len(doms) >= limit:
                break
    return doms


# --------------------------------------------------------------------------
# Main
# --------------------------------------------------------------------------

def main():
    ap = argparse.ArgumentParser(description="Discover self-hosted bug bounty / VDP programs")
    ap.add_argument("--sources", default="security_txt,diodb",
                    help="comma list: security_txt,diodb,urlscan,gcse")
    ap.add_argument("--domains", default=str(LIST_DIR / "tranco.txt"),
                    help="domain list file for the security_txt sweep")
    ap.add_argument("--limit", type=int, help="only check first N domains (testing)")
    ap.add_argument("--concurrency", type=int, default=120)
    ap.add_argument("--config", default=str(BASE / "config.yaml"))
    ap.add_argument("--no-enrich", action="store_true", help="skip fetching policy pages")
    ap.add_argument("--include-platform", action="store_true",
                    help="also output platform-hosted programs")
    ap.add_argument("--min-score", type=int, default=0)
    ap.add_argument("--quiet", action="store_true")
    args = ap.parse_args()

    cfg = {}
    if Path(args.config).exists():
        cfg = yaml.safe_load(open(args.config)) or {}

    verbose = not args.quiet
    sources = [s.strip() for s in args.sources.split(",") if s.strip()]
    collected = []

    if "diodb" in sources:
        if verbose:
            print("[*] Collecting diodb...", file=sys.stderr)
        got = collect_diodb()
        if verbose:
            print(f"    {len(got)} entries", file=sys.stderr)
        collected += got

    if "urlscan" in sources:
        if verbose:
            print("[*] Collecting urlscan...", file=sys.stderr)
        got = collect_urlscan(cfg.get("urlscan_api_key", ""))
        if verbose:
            print(f"    {len(got)} entries", file=sys.stderr)
        collected += got

    if "gcse" in sources:
        if verbose:
            print("[*] Collecting Google CSE...", file=sys.stderr)
        got = collect_gcse(cfg.get("gcse_api_key", ""), cfg.get("gcse_cx", ""),
                           cfg.get("gcse_date_restrict", "m1"))
        if verbose:
            print(f"    {len(got)} entries", file=sys.stderr)
        collected += got

    if "security_txt" in sources:
        doms = load_domains(Path(args.domains), args.limit)
        if verbose:
            print(f"[*] Sweeping {len(doms)} domains for security.txt...", file=sys.stderr)
        got = asyncio.run(collect_security_txt(doms, args.concurrency, verbose))
        if verbose:
            print(f"    {len(got)} security.txt files found", file=sys.stderr)
        collected += got

    programs = dedupe(collected)
    if not args.include_platform:
        programs = [p for p in programs if p.self_hosted != "no"]
    if verbose:
        print(f"[*] {len(programs)} candidate programs after dedupe/filter", file=sys.stderr)

    if not args.no_enrich:
        if verbose:
            print("[*] Enriching policy pages...", file=sys.stderr)
        programs = asyncio.run(enrich_all(programs, min(args.concurrency, 40)))
        if not args.include_platform:
            programs = [p for p in programs if p.self_hosted != "no"]

    con = db_connect()
    new = upsert(con, programs)

    rows = rows_from_db(con, self_hosted_only=not args.include_platform,
                        min_score=args.min_score)

    stamp = datetime.now(timezone.utc).strftime("%Y%m%d")
    csv_path = write_csv(rows, OUT_DIR / f"programs_{stamp}.csv")
    write_csv(rows, OUT_DIR / "programs_latest.csv")

    recent = rows_from_db_recent(con, cfg.get("new_days", 30),
                                 self_hosted_only=not args.include_platform)
    write_csv(recent, OUT_DIR / "programs_new_30d.csv")

    if cfg.get("sheet_id") and cfg.get("google_creds"):
        ws_main = cfg.get("worksheet", "all_programs")
        ok = write_sheet(rows, cfg["sheet_id"], ws_main, cfg["google_creds"])
        write_sheet(recent, cfg["sheet_id"], cfg.get("worksheet_new", "new_30d"),
                    cfg["google_creds"])
        if ok and verbose:
            print(f"[+] Google Sheet updated: {ws_main} + new_30d ({len(recent)} rows)",
                  file=sys.stderr)

    notify(cfg.get("webhook", ""), new)

    print(f"\n=== RUN COMPLETE {datetime.now(timezone.utc):%Y-%m-%d %H:%M UTC} ===")
    print(f"Total tracked : {con.execute('SELECT COUNT(*) FROM programs').fetchone()[0]}")
    print(f"In sheet      : {len(rows)}")
    print(f"NEW this run  : {len(new)}")
    if new:
        print("\nTop new finds:")
        for p in sorted(new, key=lambda x: -x.score)[:15]:
            print(f"  [{p.score:2d}] {p.domain:<38} {p.program_type:<8} "
                  f"SH:{p.safe_harbor:<8} {p.policy_url[:60]}")
    print(f"\nCSV: {csv_path}")
    con.close()


if __name__ == "__main__":
    import warnings
    warnings.filterwarnings("ignore")
    main()