Source code for kelvin.sdk.services.auth.pkce

"""PKCE (Proof Key for Code Exchange) utilities for OAuth 2.0."""

from __future__ import annotations

import base64
import hashlib
import secrets
from urllib.parse import urlencode




def generate_code_verifier() -> str:
    """Generate a cryptographically random code verifier for PKCE.

    Returns:
        A URL-safe base64 encoded random string (43 characters).
    """
    code_verifier = base64.urlsafe_b64encode(secrets.token_bytes(32)).decode("utf-8")
    return code_verifier.rstrip("=")





def generate_code_challenge(code_verifier: str) -> str:
    """Generate a code challenge from a code verifier for PKCE.

    Args:
        code_verifier: The code verifier string.

    Returns:
        URL-safe base64 encoded SHA256 hash of the verifier.
    """
    digest = hashlib.sha256(code_verifier.encode("utf-8")).digest()
    code_challenge = base64.urlsafe_b64encode(digest).decode("utf-8")
    return code_challenge.rstrip("=")





def generate_state() -> str:
    """Generate a random state parameter for CSRF protection.

    Returns:
        A random URL-safe string.
    """
    return secrets.token_urlsafe(32)





def build_authorization_url(
    authorization_endpoint: str,
    client_id: str,
    redirect_uri: str,
    code_challenge: str,
    state: str,
    scope: str = "openid profile email",
) -> str:
    """Build the OAuth authorization URL with PKCE parameters.

    Args:
        authorization_endpoint: The Keycloak authorization endpoint.
        client_id: The OAuth client ID.
        redirect_uri: The redirect URI (callback URL).
        code_challenge: The PKCE code challenge.
        state: The state parameter for CSRF protection.
        scope: The OAuth scopes to request.

    Returns:
        The complete authorization URL.
    """
    params = {
        "response_type": "code",
        "client_id": client_id,
        "redirect_uri": redirect_uri,
        "code_challenge": code_challenge,
        "code_challenge_method": "S256",
        "state": state,
        "scope": scope,
    }
    return f"{authorization_endpoint}?{urlencode(params)}"