src/services/security.py

from datetime import UTC, datetime, timedelta
from secrets import token_urlsafe
from typing import Any

import bcrypt
import httpx
import jwt
from google.auth.transport import requests as google_requests
from google.oauth2 import id_token as google_id_token

from src.config import Settings
from src.models.auth import TokenType


class SecurityError(ValueError):
    pass


def require_jwt_secret(settings: Settings) -> str:
    if settings.jwt_secret_key:
        return settings.jwt_secret_key
    if settings.app_env == "test":
        return "test-jwt-secret-do-not-use"
    raise SecurityError("JWT_SECRET_KEY is required for authentication.")


def hash_password(password: str) -> str:
    password_bytes = password.encode("utf-8")
    if len(password_bytes) > 72:
        raise SecurityError("Password must be 72 bytes or fewer.")
    return bcrypt.hashpw(password_bytes, bcrypt.gensalt()).decode("utf-8")


def verify_password(password: str, password_hash: str | None) -> bool:
    if not password_hash:
        return False
    try:
        return bcrypt.checkpw(password.encode("utf-8"), password_hash.encode("utf-8"))
    except ValueError:
        return False


def hash_token(token: str) -> str:
    import hashlib

    return hashlib.sha256(token.encode("utf-8")).hexdigest()


def create_access_token(
    *,
    user_id: str,
    email: str,
    settings: Settings,
    now: datetime | None = None,
) -> tuple[str, datetime]:
    issued_at = now or datetime.now(UTC)
    expires_at = issued_at + timedelta(minutes=settings.access_token_expire_minutes)
    payload = {
        "sub": user_id,
        "email": email,
        "type": TokenType.ACCESS.value,
        "exp": expires_at,
        "iat": issued_at,
    }
    token = jwt.encode(payload, require_jwt_secret(settings), algorithm=settings.jwt_algorithm)
    return token, expires_at


def create_refresh_token_value() -> str:
    return token_urlsafe(48)


def create_oauth_state(settings: Settings, now: datetime | None = None) -> str:
    issued_at = now or datetime.now(UTC)
    expires_at = issued_at + timedelta(minutes=10)
    payload = {
        "nonce": token_urlsafe(24),
        "type": TokenType.OAUTH_STATE.value,
        "exp": expires_at,
        "iat": issued_at,
    }
    return jwt.encode(payload, require_jwt_secret(settings), algorithm=settings.jwt_algorithm)


def decode_token(token: str, settings: Settings) -> dict[str, Any]:
    try:
        payload = jwt.decode(
            token,
            require_jwt_secret(settings),
            algorithms=[settings.jwt_algorithm],
        )
    except jwt.PyJWTError as exc:
        raise SecurityError("Invalid token.") from exc
    if not isinstance(payload, dict):
        raise SecurityError("Invalid token.")
    return payload


def verify_oauth_state(state: str, settings: Settings) -> None:
    payload = decode_token(state, settings)
    if payload.get("type") != TokenType.OAUTH_STATE.value:
        raise SecurityError("Invalid OAuth state.")


def build_google_authorization_url(settings: Settings, state: str) -> str:
    params = {
        "client_id": settings.google_client_id,
        "redirect_uri": settings.google_redirect_uri,
        "response_type": "code",
        "scope": "openid email profile",
        "state": state,
        "access_type": "offline",
        "prompt": "select_account",
    }
    return str(httpx.URL("https://accounts.google.com/o/oauth2/v2/auth", params=params))


async def exchange_google_code(code: str, settings: Settings) -> dict[str, Any]:
    async with httpx.AsyncClient(timeout=10) as client:
        response = await client.post(
            "https://oauth2.googleapis.com/token",
            data={
                "code": code,
                "client_id": settings.google_client_id,
                "client_secret": settings.google_client_secret,
                "redirect_uri": settings.google_redirect_uri,
                "grant_type": "authorization_code",
            },
        )
    response.raise_for_status()
    return response.json()


def verify_google_id_token(id_token: str, settings: Settings) -> dict[str, Any]:
    return google_id_token.verify_oauth2_token(
        id_token,
        google_requests.Request(),
        settings.google_client_id,
        clock_skew_in_seconds=30,
    )