pryscraper/routers/auth.py

"""Pry - Auth router (credential vault, SSO, CAPTCHA, session health).

Split from api.py on the api-router-split refactor. This file replaces
the inline endpoints that were tagged ["Auth"] in api.py. The behavior
is identical; the new shape is a FastAPI APIRouter that gets included
in the main app.

Endpoints (6):
  POST   /v1/auth/credentials        - store credentials in the vault
  GET    /v1/auth/credentials        - list credentials (no secrets)
  DELETE /v1/auth/credentials/{id}   - delete a credential
  POST   /v1/auth/sso                - generate SSO login script
  POST   /v1/auth/captcha            - solve a CAPTCHA
  POST   /v1/auth/session/health     - check session health

Part of Pry - https://git.rugmunch.io/RugMunchMedia/pryscraper
Licensed under MIT. See LICENSE.
"""

# SPDX-License-Identifier: MIT
# Copyright (c) 2026 Rug Munch Media LLC
#
# Part of Pry - https://git.rugmunch.io/RugMunchMedia/pryscraper
# Licensed under MIT. See LICENSE.
from __future__ import annotations

from typing import Any

from fastapi import APIRouter, Body

from errors import NotFoundError  # type: ignore[import-not-found]

router = APIRouter(tags=["Auth"])


@router.post("/v1/auth/credentials", summary="Store credentials in the vault")
async def store_credentials(
    name: str = Body(...),
    credential_type: str = Body("password"),
    credentials: dict[str, Any] = Body(...),
    target_url: str = Body(""),
) -> dict[str, Any]:
    """Store credentials in the encrypted vault for authenticated scraping.

    Credential types: password, api_key, cookie, token, sso
    """
    from auth_connector import store_credential

    result = store_credential(name, credential_type, credentials, target_url)
    return {"success": result["success"], "data": result}


@router.get("/v1/auth/credentials", summary="List stored credentials")
async def list_credentials_endpoint() -> dict[str, Any]:
    """List all stored credentials (without exposing secrets)."""
    from auth_connector import list_credentials

    creds = list_credentials()
    return {"success": True, "data": {"credentials": creds, "total": len(creds)}}


@router.delete("/v1/auth/credentials/{credential_id}", summary="Delete stored credentials")
async def delete_credentials_endpoint(credential_id: str) -> dict[str, Any]:
    """Delete stored credentials from the vault."""
    from auth_connector import delete_credential

    success = delete_credential(credential_id)
    if not success:
        raise NotFoundError(f"Credential not found: {credential_id}")
    return {"success": True, "data": {"deleted": True}}


@router.post("/v1/auth/sso", summary="Generate SSO login script")
async def generate_sso(
    provider: str = Body("okta"),
    username: str = Body(...),
    password: str = Body(...),
    target_url: str = Body(...),
    tenant: str = Body(""),
) -> dict[str, Any]:
    """Generate a browser automation script for SSO login.

    Supports: okta, azure_ad, google_workspace, onelogin
    """
    from auth_connector import generate_sso_script

    result = await generate_sso_script(provider, username, password, target_url, tenant)
    return {"success": result["success"], "data": result}


@router.post("/v1/auth/captcha", summary="Solve a CAPTCHA using multi-provider chain")
async def solve_captcha_endpoint(
    captcha_type: str = Body("recaptcha_v2"),
    site_key: str = Body(""),
    page_url: str = Body(""),
    action: str = Body("verify"),
    min_score: float = Body(0.3),
    image_base64: str = Body(""),
    api_keys: dict[str, str] = Body(default_factory=dict),
) -> dict[str, Any]:
    """Solve a CAPTCHA using the 6-provider chain in captcha_solver.

    Supported types:
    - recaptcha_v2: needs site_key + page_url
    - recaptcha_v3: needs site_key + page_url, optional action/min_score
    - hcaptcha:     needs site_key + page_url
    - image:        needs image_base64

    api_keys maps provider name -> API key. Providers attempted in priority
    order (capsolver -> 2captcha -> anti-captcha -> capmonster -> deathbycaptcha
    -> nextcaptcha). Returns the first successful token.
    """
    from captcha_solver import CaptchaSolver

    solver = CaptchaSolver(api_keys=api_keys)

    if captcha_type == "recaptcha_v2":
        if not (site_key and page_url):
            return {"success": False, "error": "site_key + page_url required for recaptcha_v2"}
        result = await solver.solve_recaptcha_v2(site_key=site_key, page_url=page_url)
    elif captcha_type == "recaptcha_v3":
        if not (site_key and page_url):
            return {"success": False, "error": "site_key + page_url required for recaptcha_v3"}
        result = await solver.solve_recaptcha_v3(
            site_key=site_key, page_url=page_url, action=action, min_score=min_score
        )
    elif captcha_type == "hcaptcha":
        if not (site_key and page_url):
            return {"success": False, "error": "site_key + page_url required for hcaptcha"}
        result = await solver.solve_hcaptcha(site_key=site_key, page_url=page_url)
    elif captcha_type == "image":
        if not image_base64:
            return {"success": False, "error": "image_base64 required for image CAPTCHA"}
        result = await solver.solve_image(image_base64=image_base64)
    else:
        return {"success": False, "error": f"Unknown captcha_type: {captcha_type}"}

    return {"success": result.get("success", False), "data": result}


@router.post("/v1/auth/session/health", summary="Check authenticated session health")
async def check_session(session_id: str = Body(...)) -> dict[str, Any]:
    """Check the health of an authenticated session.

    Reports cookie validity, session age, and whether re-auth is needed.
    """
    from auth_connector import check_session_health

    result = check_session_health(session_id)
    return {"success": True, "data": result}