Blame for api.py

"""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


@contextmanager
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"])

    @app.middleware("http")
    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

    @app.exception_handler(StarletteHTTPException)
    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})

    @app.exception_handler(RequestValidationError)
    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}})

    @app.get("/api/v1/health", tags=["Service"], response_model=Payload,
             summary="Check application readiness without contacting platforms")
    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"}

    @app.get("/api/v1/sources", tags=["Service"], response_model=Payload,
             summary="Describe data sources, capabilities and cache lifetimes")
    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."})

    @app.get("/api/v1/programs", tags=["Programs"], response_model=ProgramPage, responses=ERRORS,
             summary="Search and paginate programs; summaries by default")
    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}})

    @app.post("/api/v1/programs/batch", tags=["Programs"], response_model=BatchResult,
              responses={**ERRORS, 503: {"model": BatchResult, "description": "All requested program lookups failed; inspect each item."}},
              summary="Retrieve details for up to 20 programs; preserve input order")
    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})

    @app.get("/api/v1/programs/{platform}/{handle}", tags=["Programs"], response_model=ProgramDetails, responses=ERRORS,
             summary="Retrieve public program rewards, rules, scopes and exclusions")
    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))

    @app.get("/api/v1/programs/{platform}/{handle}/assets", tags=["Assets"], response_model=AssetPage, responses=ERRORS,
             summary="Get normalized assets; all exclusions remain visible")
    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."})

    @app.get("/api/v1/stats", tags=["Service"], response_model=Payload, responses=ERRORS,
             summary="Count directory programs without fetching live detail documents")
    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