gh-christianlouis-docuelevate/app/api/qr_auth.py

"""QR code login API endpoints for mobile app authentication.

Provides a secure challenge-response flow for logging into the mobile app
by scanning a QR code displayed in the web interface:

1. **Web user** calls ``POST /qr-auth/challenge`` → receives a time-limited
   challenge token (encoded in the QR code).
2. **Web UI** polls ``GET /qr-auth/challenge/{id}/status`` to detect when
   the mobile app has claimed the challenge.
3. **Mobile app** scans the QR code and calls ``POST /qr-auth/claim`` with
   the challenge token + device name → receives an API token.

Security properties:
* Challenges expire after a configurable TTL (default 2 minutes).
* Single-use: once claimed, a challenge cannot be reused (replay-safe).
* Cryptographically random 64-byte tokens.
* IP addresses are logged for audit.
"""

from __future__ import annotations

import base64
import io
import logging
from datetime import datetime
from typing import Annotated, Any

import segno
from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session

from app.auth import require_login
from app.database import get_db
from app.middleware.audit_log import get_client_ip
from app.utils.session_manager import (
    claim_qr_challenge,
    create_qr_challenge,
    get_challenge_status,
)
from app.utils.user_scope import get_current_owner_id

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/qr-auth", tags=["qr-auth"])

DbSession = Annotated[Session, Depends(get_db)]


# ---------------------------------------------------------------------------
# Auth helper
# ---------------------------------------------------------------------------


def _get_owner_id(request: Request) -> str:
    """Return the current user's owner ID, raising 401 if unauthenticated."""
    owner_id = get_current_owner_id(request)
    if not owner_id:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
    return owner_id


CurrentOwner = Annotated[str, Depends(_get_owner_id)]


# ---------------------------------------------------------------------------
# Request / Response schemas
# ---------------------------------------------------------------------------


class CreateChallengeResponse(BaseModel):
    """Response after creating a QR login challenge."""

    challenge_id: int
    challenge_token: str
    expires_at: datetime
    ttl_seconds: int = Field(description="Seconds until the challenge expires (use for client-side countdown).")
    qr_payload: str = Field(description="The string to encode in the QR code.")
    qr_code_svg: str = Field(description="Base64-encoded SVG data URI of the QR code, ready for use in an <img> src.")


class ChallengeStatusResponse(BaseModel):
    """Response for polling the status of a QR challenge."""

    id: int
    status: str  # "pending", "claimed", "expired", "cancelled"
    device_name: str | None = None
    claimed_at: datetime | None = None
    expires_at: datetime


class ClaimChallengeRequest(BaseModel):
    """Request body for claiming a QR login challenge."""

    challenge_token: str = Field(min_length=1, max_length=256)
    device_name: str = Field(
        default="Mobile App",
        min_length=1,
        max_length=120,
        description="Human-readable device name.",
    )


class ClaimChallengeResponse(BaseModel):
    """Response after successfully claiming a QR challenge."""

    token: str
    token_id: int
    name: str
    owner_id: str
    created_at: datetime


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

# QR code rendering parameters
_QR_ERROR_LEVEL = "M"  # Medium error correction (~15% recovery); sufficient for on-screen display
_QR_SCALE = 4  # Each QR module is rendered as 4×4 SVG pixels


def _generate_qr_svg(payload: str) -> str:
    """Generate a QR code for *payload* and return it as a base64 SVG data URI.

    Using ``segno`` (pure-Python, no Pillow dependency) and SVG output so the
    QR code scales crisply at any resolution without requiring a canvas or any
    client-side JavaScript library.
    """
    qr = segno.make(payload, error=_QR_ERROR_LEVEL)
    buf = io.BytesIO()
    qr.save(buf, kind="svg", scale=_QR_SCALE, xmldecl=False, svgclass=None, lineclass=None, omitsize=True)
    svg_bytes = buf.getvalue()
    return "data:image/svg+xml;base64," + base64.b64encode(svg_bytes).decode("ascii")


# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------


@router.post("/challenge", status_code=status.HTTP_201_CREATED, response_model=CreateChallengeResponse)
@require_login
async def create_challenge(
    request: Request,
    owner_id: CurrentOwner,
    db: DbSession,
) -> dict[str, Any]:
    """Create a new QR login challenge.

    The returned ``qr_payload`` should be encoded into a QR code and
    displayed to the user.  The mobile app scans this QR code and
    calls the ``/claim`` endpoint.
    """
    ip = get_client_ip(request)
    challenge = create_qr_challenge(db, owner_id, ip_address=ip)

    # The QR payload is a JSON-like string with enough info for the mobile
    # app to know the server URL and challenge token.
    base_url = str(request.base_url).rstrip("/")
    qr_payload = f"docuelevate://qr-login?token={challenge.challenge_token}&server={base_url}"

    # Compute the TTL in seconds so the client can run a countdown timer
    # without comparing absolute timestamps (which breaks when client and
    # server clocks are out of sync).
    ttl_seconds = max(0, int((challenge.expires_at - challenge.created_at).total_seconds()))

    return {
        "challenge_id": challenge.id,
        "challenge_token": challenge.challenge_token,
        "expires_at": challenge.expires_at,
        "ttl_seconds": ttl_seconds,
        "qr_payload": qr_payload,
        "qr_code_svg": _generate_qr_svg(qr_payload),
    }


@router.get("/challenge/{challenge_id}/status", response_model=ChallengeStatusResponse)
@require_login
async def poll_challenge_status(
    request: Request,
    challenge_id: int,
    owner_id: CurrentOwner,
    db: DbSession,
) -> dict[str, Any]:
    """Poll the status of a QR login challenge.

    The web UI calls this endpoint every few seconds to check if the
    mobile app has scanned the QR code and claimed the challenge.
    """
    result = get_challenge_status(db, challenge_id, owner_id)
    if not result:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Challenge not found")
    return result


@router.post("/claim", response_model=ClaimChallengeResponse)
async def claim_challenge(
    request: Request,
    body: ClaimChallengeRequest,
    db: DbSession,
) -> dict[str, Any]:
    """Claim a QR login challenge and receive an API token.

    This endpoint is called by the mobile app after scanning a QR code.
    It does **not** require authentication — the challenge token itself
    serves as proof that the user authorized this login from their web
    session.
    """
    ip = get_client_ip(request)
    result = claim_qr_challenge(db, body.challenge_token, device_name=body.device_name, ip_address=ip)

    if not result:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Invalid, expired, or already claimed challenge.",
        )

    try:
        from app.utils.audit_service import record_event

        record_event(
            db,
            action="qr_login_claimed",
            user=result["owner_id"],
            resource_type="session",
            ip_address=ip,
            details={"device_name": body.device_name, "token_id": result["token_id"]},
            severity="info",
        )
    except Exception:
        logger.debug("Failed to write QR login audit event", exc_info=True)

    return result