"""Public program metadata collection. No target scanning or private programs.""" from __future__ import annotations import copy import re import threading import time from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone from html.parser import HTMLParser from urllib.parse import urlsplit import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry PLATFORMS = ("hackerone", "bugcrowd", "intigriti", "yeswehack") DIRECTORY = "https://hackerone.com/opportunities/all" DATA_ROOT = "https://raw.githubusercontent.com/arkadiyt/bounty-targets-data/main/data" CACHE_SECONDS = 900 FAILURE_SECONDS = 60 # Only public, open programs, excluding sandbox and soft-launched entries. PROGRAM_QUERY = """ query PublicPrograms($after: String) { teams(first: 100, after: $after, where: { submission_state: {_eq: open}, state: {_nin: [sandboxed, soft_launched]}, _not: {external_program: {}} }) { nodes { handle name url website offers_bounties submission_state currency min_bounty: minimum_bounty_table_value max_bounty: maximum_bounty_table_value } pageInfo { endCursor hasNextPage } } } """ def utcnow(): return datetime.now(timezone.utc).isoformat() class CollectionError(Exception): pass class CSRFParser(HTMLParser): token = None def handle_starttag(self, tag, attrs): attrs = dict(attrs) if tag == "meta" and attrs.get("name") == "csrf-token": self.token = attrs.get("content") def session(): client = requests.Session() client.headers.update({"User-Agent": "bbcollect/1.0 (public program metadata)"}) # Retry idempotent reads only; do not aggressively retry GraphQL requests. client.mount("https://", HTTPAdapter(max_retries=Retry( total=2, backoff_factor=1, status_forcelist=[429, 502, 503, 504], allowed_methods=["GET"], respect_retry_after_header=False))) return client def get_json(response): response.raise_for_status() try: return response.json() except ValueError as exc: raise CollectionError("Upstream returned a non-JSON response") from exc def hackerone_pages(query, field, handle=None): with session() as client: response = client.get(DIRECTORY, timeout=(10, 30)) response.raise_for_status() parser = CSRFParser() parser.feed(response.text) if not parser.token: raise CollectionError("HackerOne did not provide a public session token") client.headers.update({"X-CSRF-Token": parser.token, "Referer": DIRECTORY}) cursor, seen, nodes = None, set(), [] for _ in range(100): variables = {"after": cursor} if handle is not None: variables["handle"] = handle payload = get_json(client.post("https://hackerone.com/graphql", json={"query": query, "variables": variables}, timeout=(10, 30))) if payload.get("errors"): raise CollectionError("HackerOne rejected the public GraphQL query") connection = payload.get("data", {}).get(field) if handle is not None: connection = connection.get("structured_scopes") if connection else None if not isinstance(connection, dict) or not isinstance(connection.get("nodes"), list): raise CollectionError("HackerOne program not found or response schema changed") nodes.extend(connection["nodes"]) page = connection.get("pageInfo", {}) if not page.get("hasNextPage"): return nodes cursor = page.get("endCursor") if not cursor or cursor in seen: raise CollectionError("HackerOne returned an invalid pagination cursor") seen.add(cursor) time.sleep(0.15) raise CollectionError("HackerOne exceeded the pagination safety limit") def bounty_value(value): if isinstance(value, dict): return value.get("value") return value def normalize(platform, row): url = row.get("url") or "" handle = str(row.get("handle") or row.get("id") or urlsplit(url).path.rstrip("/").split("/")[-1]) # Intigriti handles are unique within a company, not across the platform. if platform == "intigriti": handle = str(row.get("id") or f"{row.get('company_handle', '')}--{handle}") if platform == "yeswehack" and not url: url = f"https://yeswehack.com/programs/{handle}" maximum = bounty_value(row.get("max_bounty", row.get("max_payout"))) paid = row.get("offers_bounties") if paid is None and isinstance(maximum, (int, float)): paid = maximum > 0 return {"id": f"{platform}:{handle}", "platform": platform, "handle": handle, "name": row.get("name", handle), "url": url, "website": row.get("website"), "offers_bounties": paid, "min_bounty": bounty_value(row.get("min_bounty")), "max_bounty": maximum, "currency": row.get("currency") or (row.get("max_bounty", {}).get("currency") if isinstance(row.get("max_bounty"), dict) else None), "submission_state": row.get("submission_state", row.get("status")), "targets": row.get("targets")} class Collector: def __init__(self): self._cache = {} self._fill_locks = [threading.RLock() for _ in range(32)] def _cached(self, key, loader): # Coalesce identical requests while allowing independent program fills. with self._fill_locks[hash(key) % len(self._fill_locks)]: entry = self._cache.get(key) if entry and entry[0] > time.monotonic(): result = copy.deepcopy(entry[1]) result["cached"] = True return result try: result = loader() except (requests.RequestException, CollectionError, ValueError, KeyError, TypeError) as exc: # Do not include session headers, cookies, or raw upstream bodies. result = {"ok": False, "error": type(exc).__name__, "message": "Source unavailable or response schema changed; retry later.", "fetched_at": utcnow()} result["cached"] = False ttl = CACHE_SECONDS if result["ok"] else FAILURE_SECONDS self._cache[key] = (time.monotonic() + ttl, copy.deepcopy(result)) return result def dataset(self, platform): def load(): url = f"{DATA_ROOT}/{platform}_data.json" with session() as client: response = client.get(url, timeout=(10, 45)) data = get_json(response) if not isinstance(data, list) or not data or not all( isinstance(row, dict) and row.get("name") for row in data ): raise CollectionError("Invalid program dataset") return {"ok": True, "mode": "public_dataset", "source_url": url, "fetched_at": utcnow(), "source_updated_at": response.headers.get("Last-Modified"), "programs": [normalize(platform, row) for row in data]} return self._cached(("dataset", platform), load) def programs(self, platform): if platform != "hackerone": return self.dataset(platform) def load(): try: data = hackerone_pages(PROGRAM_QUERY, "teams") if not data: raise CollectionError("HackerOne returned an empty directory") return {"ok": True, "mode": "live", "source_url": DIRECTORY, "fetched_at": utcnow(), "source_updated_at": None, "programs": [normalize(platform, row) for row in data]} except (requests.RequestException, CollectionError, ValueError, KeyError, TypeError): fallback = self.dataset(platform) fallback["warning"] = "Live HackerOne collection failed; using a public dataset snapshot." return fallback return self._cached(("programs", platform), load) def collect(self, platforms=None, search="", bounty_only=False, limit=100, offset=0, include_details=True): platforms = ["hackerone"] if platforms is None else platforms if not isinstance(platforms, list) or any(p not in PLATFORMS for p in platforms): raise ValueError("Choose platforms from: " + ", ".join(PLATFORMS)) if not platforms: raise ValueError("Select at least one platform") limit, offset = int(limit), int(offset) if not 1 <= limit <= 1000 or not 0 <= offset <= 100000: raise ValueError("limit must be 1–1000; offset must be 0–100000") if include_details and limit > 100: raise ValueError("Full details support at most 100 programs per page; disable include_details for larger summary pages") query = search.strip().casefold() programs, sources = {}, {} for platform in dict.fromkeys(platforms): result = self.programs(platform) sources[platform] = {k: v for k, v in result.items() if k != "programs"} for program in result.get("programs", []): if bounty_only and program["offers_bounties"] is not True: continue if query and query not in " ".join(str(program.get(k) or "") for k in ("name", "handle", "url", "website")).casefold(): continue programs[program["id"]] = {k: v for k, v in program.items() if k != "targets"} ordered = sorted(programs.values(), key=lambda p: (p["platform"], p["name"].casefold(), p["id"])) ok = any(s["ok"] for s in sources.values()) page = ordered[offset:offset + limit] if include_details and page: def enrich(program): detail = self.details(program["platform"], program["handle"]) enriched = dict(program) for key in ("targets", "rules", "reward_table", "field_status", "currency", "min_bounty", "max_bounty"): if key in detail: enriched[key] = detail[key] enriched["details_source"] = {k: v for k, v in detail.items() if k not in ( "targets", "rules", "reward_table", "min_bounty", "max_bounty", "currency", "field_status")} return enriched with ThreadPoolExecutor(max_workers=4) as pool: page = list(pool.map(enrich, page)) details_partial = any(p.get("details_source", {}).get("partial") or p.get("details_source", {}).get("ok") is False for p in page) return {"ok": ok, "partial": ok and (any(not s["ok"] for s in sources.values()) or details_partial), "generated_at": utcnow(), "total": len(ordered), "limit": limit, "offset": offset, "next_offset": offset + limit if offset + limit < len(ordered) else None, "include_details": include_details, "sources": sources, "programs": page} def details(self, platform="hackerone", handle="security"): if platform not in PLATFORMS or not re.fullmatch(r"[A-Za-z0-9_.-]{1,200}", handle): raise ValueError("Provide a supported platform and a handle returned by /collect") program = None dataset = None if platform == "intigriti": dataset = self.dataset(platform) program = next((p for p in dataset.get("programs", []) if p["handle"] == handle), None) def load(): from details import live_details if platform == "intigriti" and program is None: raise CollectionError("Program not found in public directory") return live_details(platform, handle, program) result = self._cached(("details", platform, handle), load) if result["ok"]: return result dataset = dataset or self.dataset(platform) program = next((p for p in dataset.get("programs", []) if p["handle"] == handle), None) if not program: return {**result, "platform": platform, "handle": handle, "partial": True, "targets": None, "rules": None, "reward_table": None, "field_status": {k: "unavailable" for k in ("targets", "rules", "min_bounty", "max_bounty")}} return {**{k: v for k, v in dataset.items() if k != "programs"}, "ok": True, "partial": True, "platform": platform, "handle": handle, "targets": program.get("targets"), "rules": None, "reward_table": None, "min_bounty": program.get("min_bounty"), "max_bounty": program.get("max_bounty"), "currency": program.get("currency"), "warning": "Live program details unavailable; dataset scopes/rewards retained. Full rules are unavailable.", "field_status": {"targets": "dataset_snapshot", "rules": "unavailable", "min_bounty": "dataset_snapshot" if program.get("min_bounty") is not None else "unavailable", "max_bounty": "dataset_snapshot" if program.get("max_bounty") is not None else "unavailable"}} def scope(self, platform="hackerone", handle="security"): return self.details(platform, handle) collector = Collector()