OAuth 2.1 para Python

from __future__ import annotations

import base64
import hashlib
import json
import secrets
from urllib.parse import urlencode
from urllib.request import Request, urlopen
from urllib.error import HTTPError


class PietroCalOAuth:
    def __init__(
        self,
        client_id: str,
        redirect_uri: str,
        client_secret: str | None = None,
        base_url: str = "https://pietrocal.com",
    ) -> None:
        if not client_id.startswith("pc_oauth_"):
            raise ValueError("El Client ID debe comenzar con pc_oauth_.")
        self.client_id = client_id
        self.redirect_uri = redirect_uri
        self.client_secret = client_secret
        self.base_url = base_url.rstrip("/")

    def create_pkce(self) -> dict[str, str]:
        verifier = self._base64url(secrets.token_bytes(48))
        challenge = self._base64url(hashlib.sha256(verifier.encode()).digest())
        return {"codeVerifier": verifier, "codeChallenge": challenge}

    def authorization_url(
        self,
        scopes: list[str],
        state: str,
        code_challenge: str,
    ) -> str:
        return self.base_url + "/oauth/authorize.php?" + urlencode({
            "response_type": "code",
            "client_id": self.client_id,
            "redirect_uri": self.redirect_uri,
            "scope": " ".join(dict.fromkeys(scopes)),
            "state": state,
            "code_challenge": code_challenge,
            "code_challenge_method": "S256",
        })

    def exchange_code(self, code: str, code_verifier: str) -> dict:
        return self._post_form("/oauth/token.php", {
            "grant_type": "authorization_code",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "code": code,
            "redirect_uri": self.redirect_uri,
            "code_verifier": code_verifier,
        })

    def refresh(self, refresh_token: str) -> dict:
        return self._post_form("/oauth/token.php", {
            "grant_type": "refresh_token",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "refresh_token": refresh_token,
        })

    def revoke(self, token: str, token_type_hint: str | None = None) -> None:
        self._post_form("/oauth/revoke.php", {
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "token": token,
            "token_type_hint": token_type_hint,
        }, allow_empty=True)

    def _post_form(self, path: str, fields: dict, allow_empty: bool = False) -> dict:
        body = urlencode({
            key: value
            for key, value in fields.items()
            if value is not None and value != ""
        }).encode()

        request = Request(
            self.base_url + path,
            data=body,
            headers={
                "Accept": "application/json",
                "Content-Type": "application/x-www-form-urlencoded",
            },
            method="POST",
        )

        try:
            with urlopen(request, timeout=30) as response:
                raw = response.read().decode()
        except HTTPError as error:
            raw = error.read().decode(errors="replace")
            try:
                data = json.loads(raw)
            except json.JSONDecodeError:
                data = {}
            raise RuntimeError(
                data.get("error_description", "La operación OAuth falló.")
            ) from error

        if allow_empty and not raw.strip():
            return {}
        return json.loads(raw) if raw.strip() else {}

    @staticmethod
    def _base64url(value: bytes) -> str:
        return base64.urlsafe_b64encode(value).decode().rstrip("=")