liumangmang
123 lines
3.6 KiB
Python
123 lines
3.6 KiB
Python
"""Auth capture API for real-browser credential imports."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.services.browser_import_service import (
|
|
ImportSessionError,
|
|
browser_imports,
|
|
build_import_result,
|
|
)
|
|
from app.utils.auth import get_current_user
|
|
|
|
router = APIRouter(prefix="/api/auth-capture", tags=["auth-capture"])
|
|
|
|
SENSITIVE_CANDIDATE_FIELDS = frozenset({"value", "cookie_value"})
|
|
|
|
|
|
class CaptureExtractResponse(BaseModel):
|
|
cookies: list[dict] = []
|
|
storage: dict[str, str] = {}
|
|
session_storage: dict[str, str] = {}
|
|
auth_headers: list[dict] = []
|
|
candidates: list[dict] = []
|
|
|
|
|
|
class ImportSessionCreate(BaseModel):
|
|
target_url: str = Field(..., description="Target page URL opened in the user's real browser")
|
|
|
|
|
|
class ImportSessionCreateResponse(BaseModel):
|
|
session_id: str
|
|
secret: str
|
|
expires_in_seconds: int
|
|
|
|
|
|
class ImportSessionStatusResponse(BaseModel):
|
|
session_id: str
|
|
ready: bool = False
|
|
expires_at: float
|
|
result: Optional[CaptureExtractResponse] = None
|
|
|
|
|
|
class ImportSessionSubmit(BaseModel):
|
|
secret: str
|
|
page_url: str = ""
|
|
cookies: list[dict] = []
|
|
local_storage: dict[str, Any] = {}
|
|
session_storage: dict[str, Any] = {}
|
|
auth_headers: list[dict] = []
|
|
|
|
|
|
def _sanitize_candidate(candidate: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
key: value
|
|
for key, value in candidate.items()
|
|
if key not in SENSITIVE_CANDIDATE_FIELDS
|
|
}
|
|
|
|
|
|
@router.post("/import-sessions", response_model=ImportSessionCreateResponse, status_code=201)
|
|
async def create_import_session(
|
|
body: ImportSessionCreate,
|
|
user=Depends(get_current_user),
|
|
):
|
|
"""Create a one-time real-browser import session.
|
|
|
|
The returned secret is intended for the local browser extension only.
|
|
"""
|
|
target_url = body.target_url.strip()
|
|
if not target_url.startswith(("http://", "https://")):
|
|
raise HTTPException(400, "Only http/https URLs are allowed")
|
|
session, secret = browser_imports.create(target_url=target_url, created_by=user.email)
|
|
return ImportSessionCreateResponse(
|
|
session_id=session.id,
|
|
secret=secret,
|
|
expires_in_seconds=600,
|
|
)
|
|
|
|
|
|
@router.get("/import-sessions/{session_id}", response_model=ImportSessionStatusResponse)
|
|
async def get_import_session(
|
|
session_id: str,
|
|
include_raw: bool = Query(default=False),
|
|
user=Depends(get_current_user),
|
|
):
|
|
try:
|
|
session = browser_imports.get(session_id, created_by=user.email)
|
|
except ImportSessionError as exc:
|
|
raise HTTPException(404, str(exc))
|
|
|
|
result = None
|
|
if session.payload is not None:
|
|
full_result = build_import_result(session.payload)
|
|
if not include_raw:
|
|
candidates = [_sanitize_candidate(candidate) for candidate in full_result.get("candidates", [])]
|
|
result = CaptureExtractResponse(candidates=candidates)
|
|
else:
|
|
result = CaptureExtractResponse(**full_result)
|
|
return ImportSessionStatusResponse(
|
|
session_id=session.id,
|
|
ready=session.payload is not None,
|
|
expires_at=session.expires_at,
|
|
result=result,
|
|
)
|
|
|
|
|
|
@router.post("/import-sessions/{session_id}/submit", status_code=204)
|
|
async def submit_import_session(
|
|
session_id: str,
|
|
body: ImportSessionSubmit,
|
|
):
|
|
try:
|
|
browser_imports.submit(
|
|
session_id=session_id,
|
|
secret=body.secret,
|
|
payload=body.model_dump(exclude={"secret"}),
|
|
)
|
|
except ImportSessionError as exc:
|
|
raise HTTPException(400, str(exc))
|