Spaces:
Running
Running
| """Public, unauthenticated REST v1 interface alongside the Gradio API.""" | |
| from concurrent.futures import ThreadPoolExecutor | |
| from contextlib import contextmanager | |
| from enum import Enum | |
| import threading | |
| from typing import Any, Literal | |
| from fastapi import FastAPI, HTTPException, Path, Query, Response | |
| from fastapi.exceptions import RequestValidationError | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from fastapi.responses import JSONResponse | |
| from pydantic import BaseModel, ConfigDict, Field | |
| from starlette.exceptions import HTTPException as StarletteHTTPException | |
| from collector import CACHE_SECONDS, FAILURE_SECONDS, PLATFORMS, collector, utcnow | |
| VERSION = "1.0.0" | |
| _work_slots = threading.BoundedSemaphore(2) | |
| class Platform(str, Enum): | |
| hackerone = "hackerone" | |
| bugcrowd = "bugcrowd" | |
| intigriti = "intigriti" | |
| yeswehack = "yeswehack" | |
| class ProgramRef(BaseModel): | |
| model_config = ConfigDict(extra="forbid") | |
| platform: Platform | |
| handle: str = Field(min_length=1, max_length=200, pattern=r"^[A-Za-z0-9_.-]+$", | |
| description="Use the exact handle returned by the programs endpoint.") | |
| class BatchRequest(BaseModel): | |
| model_config = ConfigDict(extra="forbid") | |
| programs: list[ProgramRef] = Field(min_length=1, max_length=20) | |
| class Payload(BaseModel): | |
| """Shared envelope; platform-native policy and reward structures remain JSON.""" | |
| model_config = ConfigDict(extra="allow") | |
| ok: bool | |
| partial: bool = False | |
| api_version: str = VERSION | |
| class ErrorInfo(BaseModel): | |
| code: str | |
| message: str | |
| details: list[dict[str, Any]] | None = None | |
| class ErrorPayload(BaseModel): | |
| ok: Literal[False] = False | |
| api_version: str = VERSION | |
| error: ErrorInfo | |
| class ProgramResult(BaseModel): | |
| model_config = ConfigDict(extra="allow") | |
| id: str | |
| platform: Platform | |
| handle: str | |
| name: str | |
| url: str | |
| offers_bounties: bool | None = None | |
| min_bounty: float | None = None | |
| max_bounty: float | None = None | |
| currency: str | None = None | |
| targets: dict[str, Any] | None = None | |
| rules: dict[str, Any] | None = None | |
| reward_table: Any = None | |
| details_source: dict[str, Any] | None = None | |
| field_status: dict[str, str] | None = None | |
| class ProgramPage(Payload): | |
| total: int | |
| limit: int | |
| offset: int | |
| next_offset: int | None | |
| include_details: bool | |
| sources: dict[str, dict[str, Any]] | |
| programs: list[ProgramResult] | |
| filters: dict[str, Any] | |
| class ProgramDetails(Payload): | |
| platform: str | |
| handle: str | |
| targets: dict[str, Any] | None = None | |
| rules: dict[str, Any] | None = None | |
| reward_table: Any = None | |
| min_bounty: float | None = None | |
| max_bounty: float | None = None | |
| currency: str | None = None | |
| field_status: dict[str, str] | None = None | |
| mode: str | None = None | |
| source_url: str | None = None | |
| fetched_at: str | None = None | |
| source_updated_at: str | None = None | |
| warning: str | None = None | |
| cached: bool = False | |
| class BatchResult(Payload): | |
| count: int | |
| succeeded: int | |
| programs: list[ProgramDetails] | |
| class Asset(BaseModel): | |
| identifier: str | |
| type: str | |
| native_type: str | int | None | |
| instructions: str | None | |
| eligible_for_bounty: bool | None | |
| raw: Any | |
| class AssetPage(Payload): | |
| platform: str | |
| handle: str | |
| total: int | |
| limit: int | |
| offset: int | |
| next_offset: int | None | |
| in_scope: list[Asset] | |
| out_of_scope: list[Asset] | |
| rules: dict[str, Any] | None = None | |
| source: dict[str, Any] | |
| note: str | |
| ERRORS = {code: {"model": ErrorPayload, "description": label} for code, label in ( | |
| (400, "Invalid filter combination"), (422, "Invalid query or request body"), | |
| (429, "Collection capacity busy; retry after the Retry-After header"), | |
| (503, "Required upstream data is unavailable"))} | |
| def envelope(result): | |
| return {"api_version": VERSION, **result} | |
| def bad_request(message): | |
| raise HTTPException(400, {"code": "invalid_filters", "message": message}) | |
| def parse_platforms(value): | |
| selected = list(dict.fromkeys(p.strip().lower() for p in value.split(",") if p.strip())) | |
| if not selected or any(p not in PLATFORMS for p in selected): | |
| bad_request("platforms must contain comma-separated values from: " + ", ".join(PLATFORMS)) | |
| return selected | |
| def work_slot(): | |
| if not _work_slots.acquire(blocking=False): | |
| raise HTTPException(429, {"code": "busy", "message": "Collection capacity is busy; retry in a few seconds."}, | |
| headers={"Retry-After": "5"}) | |
| try: | |
| yield | |
| finally: | |
| _work_slots.release() | |
| def require_data(result): | |
| if not result.get("ok"): | |
| raise HTTPException(503, {"code": "source_unavailable", | |
| "message": "Public program data was not found or is temporarily unavailable."}) | |
| return envelope(result) | |
| def normalize_asset(asset): | |
| """Common identifiers/types, retaining the complete native asset record.""" | |
| if isinstance(asset, str): | |
| return {"identifier": asset, "type": "other", "native_type": None, | |
| "instructions": None, "eligible_for_bounty": None, "raw": asset} | |
| identifier = next((asset[k] for k in ("asset_identifier", "scope", "endpoint", "target", "uri", "name", "ipAddress") | |
| if asset.get(k)), "") | |
| native = asset.get("asset_type", asset.get("scope_type", asset.get("category", asset.get("type", asset.get("typeId"))))) | |
| types = {1: "url", 2: "android", 3: "ios", 4: "ip_range", 5: "device", 6: "other", 7: "wildcard", | |
| "website": "url", "web-application": "url", "url": "url", "wildcard": "wildcard", | |
| "api": "api", "cidr": "ip_range", "iprange": "ip_range", "ip_range": "ip_range", | |
| "ip_address": "ip", "ip": "ip", "android": "android", "ios": "ios", | |
| "apple_store_app_id": "ios", "google_play_app_id": "android", "domain": "domain"} | |
| key = native.lower() if isinstance(native, str) else native | |
| kind = types.get(key, "other") | |
| return {"identifier": str(identifier), "type": kind, "native_type": native, | |
| "instructions": asset.get("instruction", asset.get("description")), | |
| "eligible_for_bounty": asset.get("eligible_for_bounty"), "raw": asset} | |
| def create_api(): | |
| app = FastAPI(title="bbconnectapi", version=VERSION, | |
| description="Public bug bounty program metadata, rewards, scopes and policies. " | |
| "No API key, login or Authorization header required. " | |
| "Directory summaries are fast; live details can take longer. " | |
| "Check partial/source status before using results.", | |
| docs_url="/api/docs", redoc_url="/api/redoc", openapi_url="/api/openapi.json", | |
| servers=[{"url": "/", "description": "Current host"}], | |
| openapi_tags=[{"name": "Programs", "description": "Search public programs and retrieve details."}, | |
| {"name": "Assets", "description": "Normalized assets with native records and exclusions."}, | |
| {"name": "Service", "description": "Readiness, supported sources and directory statistics."}]) | |
| app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["GET", "POST", "OPTIONS"], | |
| allow_headers=["Content-Type"], allow_credentials=False, | |
| expose_headers=["Retry-After", "X-API-Version"]) | |
| async def version_header(request, call_next): | |
| response = await call_next(request) | |
| if request.url.path.startswith("/api/"): | |
| response.headers["X-API-Version"] = VERSION | |
| response.headers["Cache-Control"] = "no-store" | |
| return response | |
| async def http_error(request, exc): | |
| detail = exc.detail if isinstance(exc.detail, dict) else {"code": "http_error", "message": str(exc.detail)} | |
| return JSONResponse(status_code=exc.status_code, headers=exc.headers, | |
| content={"ok": False, "api_version": VERSION, "error": detail}) | |
| async def validation_error(request, exc): | |
| details = [{"location": list(e["loc"]), "message": e["msg"], "type": e["type"]} for e in exc.errors()] | |
| return JSONResponse(status_code=422, content={"ok": False, "api_version": VERSION, | |
| "error": {"code": "validation_error", "message": "Check the request parameters.", "details": details}}) | |
| def health(): | |
| return {"ok": True, "api_version": VERSION, "authentication_required": False, | |
| "generated_at": utcnow(), "upstream_checked": False, | |
| "docs_url": "/api/docs", "openapi_url": "/api/openapi.json"} | |
| def sources(): | |
| return envelope({"ok": True, "authentication_required": False, | |
| "cache_seconds": CACHE_SECONDS, "failure_cache_seconds": FAILURE_SECONDS, | |
| "platforms": [{"platform": p, "directory": "live_with_dataset_fallback" if p == "hackerone" else "public_dataset", | |
| "details": "live_with_dataset_fallback", "rules": "live_only"} for p in PLATFORMS], | |
| "limits": {"full_detail_page": 100, "summary_page": 1000, "batch_programs": 20}, | |
| "availability": "Public programs only. Fields may be unpublished; fallbacks can be partial."}) | |
| def programs( | |
| platforms: str = Query("hackerone", description="Comma-separated platform names; use /sources for supported values."), | |
| search: str = Query("", max_length=300, description="Case-insensitive substring in name, handle, URL or website."), | |
| bounty_only: bool = False, | |
| limit: int = Query(25, ge=1, le=1000), offset: int = Query(0, ge=0, le=100000), | |
| include_details: bool = Query(False, description="Live rewards, targets and rules for this page; maximum limit 100."), | |
| currency: str | None = Query(None, pattern=r"^[A-Za-z]{3}$", description="ISO currency code; excludes unknown currencies."), | |
| reward_at_least: float | None = Query(None, ge=0, allow_inf_nan=False, | |
| description="Require advertised max_bounty >= this amount. Requires currency; applied to directory metadata."), | |
| sort_by: Literal["name", "max_bounty", "min_bounty"] = "name", | |
| descending: bool = False, | |
| ): | |
| selected = parse_platforms(platforms) | |
| if (reward_at_least is not None or sort_by != "name") and currency is None: | |
| bad_request("Set currency when filtering or sorting by bounty amounts; currencies are never mixed.") | |
| if include_details and limit > 100: | |
| bad_request("include_details=true supports limit <= 100; use summary pages for larger limits.") | |
| with work_slot(): | |
| # Fetch the complete small directory before filtering/sorting/pagination. | |
| result = collector.collect(selected, search, bounty_only, 1000, 0, False) | |
| require_data(result) | |
| if result.get("next_offset") is not None: | |
| # A platform may grow beyond one summary page. | |
| items = list(result["programs"]) | |
| cursor = result["next_offset"] | |
| while cursor is not None: | |
| extra = collector.collect(selected, search, bounty_only, 1000, cursor, False) | |
| require_data(extra) | |
| items.extend(extra["programs"]) | |
| cursor = extra["next_offset"] | |
| result["programs"] = items | |
| rows = result["programs"] | |
| if currency: | |
| rows = [p for p in rows if (p.get("currency") or "").upper() == currency.upper()] | |
| if reward_at_least is not None: | |
| rows = [p for p in rows if isinstance(p.get("max_bounty"), (int, float)) and p["max_bounty"] >= reward_at_least] | |
| if sort_by == "name": | |
| rows.sort(key=lambda p: (p["name"].casefold(), p["id"]), reverse=descending) | |
| else: | |
| known = [p for p in rows if isinstance(p.get(sort_by), (int, float))] | |
| unknown = [p for p in rows if not isinstance(p.get(sort_by), (int, float))] | |
| known.sort(key=lambda p: (p[sort_by], p["id"]), reverse=descending) | |
| rows = known + sorted(unknown, key=lambda p: p["id"]) | |
| total = len(rows) | |
| page = rows[offset:offset + limit] | |
| if include_details and page: | |
| def enrich(p): | |
| detail = collector.details(p["platform"], p["handle"]) | |
| fields = ("targets", "rules", "reward_table", "field_status", "currency", "min_bounty", "max_bounty") | |
| return {**p, **{k: detail[k] for k in fields if k in detail}, | |
| "details_source": {k: v for k, v in detail.items() if k not in fields}} | |
| with ThreadPoolExecutor(max_workers=4) as pool: | |
| page = list(pool.map(enrich, page)) | |
| partial = result.get("partial", False) or any( | |
| p.get("details_source", {}).get("partial") or p.get("details_source", {}).get("ok") is False for p in page) | |
| return envelope({**result, "programs": page, "total": total, "limit": limit, "offset": offset, | |
| "next_offset": offset + limit if offset + limit < total else None, | |
| "include_details": include_details, "partial": partial, | |
| "filters": {"currency": currency.upper() if currency else None, "reward_at_least": reward_at_least, | |
| "reward_filter_source": "directory_metadata", "sort_by": sort_by, "descending": descending}}) | |
| def batch(body: BatchRequest, response: Response): | |
| with work_slot(): | |
| def lookup(ref): | |
| return collector.details(ref.platform.value, ref.handle) | |
| with ThreadPoolExecutor(max_workers=4) as pool: | |
| results = list(pool.map(lookup, body.programs)) | |
| successes = sum(bool(r.get("ok")) for r in results) | |
| if not successes: | |
| response.status_code = 503 | |
| return envelope({"ok": successes > 0, "partial": any(not r.get("ok") or r.get("partial") for r in results), | |
| "count": len(results), "succeeded": successes, "programs": results}) | |
| def program(platform: Platform, handle: str = Path(pattern=r"^[A-Za-z0-9_.-]{1,200}$")): | |
| with work_slot(): | |
| return require_data(collector.details(platform.value, handle)) | |
| def assets(platform: Platform, handle: str = Path(pattern=r"^[A-Za-z0-9_.-]{1,200}$"), | |
| asset_type: Literal["url", "api", "domain", "wildcard", "ip", "ip_range", "android", "ios", "device", "other"] | None = None, | |
| search: str = Query("", max_length=300), limit: int = Query(100, ge=1, le=1000), | |
| offset: int = Query(0, ge=0, le=100000)): | |
| with work_slot(): | |
| detail = require_data(collector.details(platform.value, handle)) | |
| targets = detail.get("targets") | |
| if not isinstance(targets, dict): | |
| raise HTTPException(503, {"code": "scope_unavailable", "message": "No scope data is available for this program."}) | |
| rows = [normalize_asset(a) for a in targets.get("in_scope", [])] | |
| rows = [a for a in rows if (asset_type is None or a["type"] == asset_type) and search.casefold() in a["identifier"].casefold()] | |
| exclusions = [normalize_asset(a) for a in targets.get("out_of_scope", [])] | |
| return envelope({"ok": True, "partial": detail.get("partial", False), "platform": platform.value, | |
| "handle": handle, "total": len(rows), "limit": limit, "offset": offset, | |
| "next_offset": offset + limit if offset + limit < len(rows) else None, | |
| "in_scope": rows[offset:offset + limit], "out_of_scope": exclusions, | |
| "rules": detail.get("rules"), "field_status": detail.get("field_status"), | |
| "source": {k: detail.get(k) for k in ("mode", "source_url", "fetched_at", "source_updated_at", "cached", "warning")}, | |
| "note": "Filters/pagination apply only to in_scope. All exclusions and program rules are retained."}) | |
| def stats(platforms: str = "hackerone,bugcrowd,intigriti,yeswehack"): | |
| selected = parse_platforms(platforms) | |
| sources, counts = {}, {} | |
| with work_slot(): | |
| for p in selected: | |
| result = collector.programs(p) | |
| rows = list({r["id"]: r for r in result.get("programs", [])}.values()) | |
| counts[p] = {"total": len(rows), | |
| "offers_bounties": sum(r.get("offers_bounties") is True for r in rows), | |
| "unknown_bounty_status": sum(r.get("offers_bounties") is None for r in rows)} | |
| sources[p] = {k: v for k, v in result.items() if k != "programs"} | |
| ok = any(s.get("ok") for s in sources.values()) | |
| result = {"ok": ok, "partial": any(not s.get("ok") for s in sources.values()), "generated_at": utcnow(), | |
| "total": sum(c["total"] for c in counts.values()), "platforms": counts, "sources": sources} | |
| return require_data(result) | |
| return app | |