# SDK Python: Guía OAuth ## Resumen Desde la versión 0.5.0, el SDK Python de UTILIA soporta autenticación OAuth 2.1 con PKCE como alternativa a las API Keys. OAuth es la opción recomendada cuando tu aplicación actúa en nombre de un usuario o cuando necesitas permisos granulares. ## API Key vs OAuth | Aspecto | API Key | OAuth 2.1 | |---|---|---| | **Caso de uso** | Integraciones servidor-a-servidor | Aplicaciones que actúan en nombre de un usuario | | **Configuración** | Un único valor (`api_key`) | client_id, scopes, redirect_uri | | **Permisos** | Todos los permisos del tenant | Solo los scopes concedidos por el usuario | | **Tokens** | Estático, no expira | Access token con expiración + refresh token | | **Seguridad** | Adecuada para backends | Recomendada para cualquier entorno | ## Configuración OAuth ```python from utilia_sdk import UtiliaSDK sdk = UtiliaSDK( base_url="https://os.utilia.ai/api", oauth={ "client_id": "client_xxxxxxxxxx", "client_secret": "secret_xxxxxxxxxx", # Solo para apps confidenciales "redirect_uri": "http://localhost:8000/callback", "scopes": ["openid", "profile", "email", "dev:read"], }, ) ``` ```python from utilia_sdk import UtiliaSDKSync sdk = UtiliaSDKSync( base_url="https://os.utilia.ai/api", oauth={ "client_id": "client_xxxxxxxxxx", "client_secret": "secret_xxxxxxxxxx", "redirect_uri": "http://localhost:8000/callback", "scopes": ["openid", "profile", "email", "dev:read"], }, ) ``` ### Parámetros OAuth | Parámetro | Tipo | Requerido | Descripción | |---|---|---|---| | `client_id` | `str` | Sí | Client ID de tu aplicación OAuth | | `client_secret` | `str` | No | Client Secret (solo apps confidenciales / servidor) | | `redirect_uri` | `str` | Sí | URI de redirección registrada en el panel | | `scopes` | `list[str]` | No | Scopes a solicitar. Opcional; por defecto `['openid', 'profile', 'email']` | | `token_storage` | `TokenStorage` | No | Estrategia de almacenamiento de tokens | ## PKCE (Proof Key for Code Exchange) PKCE es un mecanismo de seguridad obligatorio en OAuth 2.1 que protege el flujo de autorización contra ataques de interceptación. El SDK lo gestiona automáticamente: 1. Genera un `code_verifier` aleatorio. 2. Calcula el `code_challenge` como hash SHA-256 codificado en base64url. 3. Envía el challenge en la petición de autorización y el verifier al intercambiar el código. No necesitas implementar nada manualmente. ## TokenStorage ### MemoryTokenStorage (por defecto) Almacena los tokens en memoria. Se pierden al finalizar el proceso. ```python from utilia_sdk import UtiliaSDK, MemoryTokenStorage sdk = UtiliaSDK( base_url="https://os.utilia.ai/api", oauth={ "client_id": "client_xxxxxxxxxx", "redirect_uri": "http://localhost:8000/callback", "scopes": ["openid", "profile", "email"], "token_storage": MemoryTokenStorage(), }, ) ``` ### FileTokenStorage Almacena los tokens en un archivo JSON. Ideal para scripts y CLIs. ```python from utilia_sdk import UtiliaSDK, FileTokenStorage sdk = UtiliaSDK( base_url="https://os.utilia.ai/api", oauth={ "client_id": "client_xxxxxxxxxx", "client_secret": "secret_xxxxxxxxxx", "redirect_uri": "http://localhost:8000/callback", "scopes": ["openid", "profile", "email"], "token_storage": FileTokenStorage(".utilia-tokens.json"), }, ) ``` El archivo de tokens contiene credenciales sensibles. Añádelo a `.gitignore` y protege sus permisos (`chmod 600`). ### TokenStorage personalizado Hereda de la clase abstracta `TokenStorage` e implementa sus tres métodos abstractos, que se llaman `get_tokens`, `set_tokens` y `clear_tokens`: ```python import dataclasses import json from utilia_sdk import TokenStorage, OAuthTokens class RedisTokenStorage(TokenStorage): def __init__(self, redis_client, key: str = "utilia:tokens"): self._redis = redis_client self._key = key async def set_tokens(self, tokens: OAuthTokens) -> None: await self._redis.set(self._key, json.dumps(dataclasses.asdict(tokens))) async def get_tokens(self) -> OAuthTokens | None: data = await self._redis.get(self._key) return OAuthTokens(**json.loads(data)) if data else None async def clear_tokens(self) -> None: await self._redis.delete(self._key) ``` `TokenStorage` es una clase abstracta (`ABC`), no un `Protocol`, y sus métodos abstractos son `get_tokens`, `set_tokens` y `clear_tokens`. Una subclase que defina `save`/`load`/`clear` no cubre ninguno de los tres y Python impide instanciarla: `TypeError: Can't instantiate abstract class ... with abstract methods clear_tokens, get_tokens, set_tokens`. Además, `OAuthTokens` es un `dataclass`, no un modelo Pydantic: no tiene `model_dump_json()` ni `model_validate_json()`. Serialízalo con `dataclasses.asdict()`. Para `UtiliaSDKSync` hay que heredar de `SyncTokenStorage`, cuyos métodos son `get_tokens_sync`, `set_tokens_sync` y `clear_tokens_sync`. Ambas clases ofrecen además tres métodos opcionales con implementación por defecto (`get_pending_state`, `set_pending_state` y `clear_pending_state`), que el SDK usa para el flujo PKCE automático. Sólo hace falta redefinirlos si quieres persistir el estado PKCE. ## Renovación automática de tokens El SDK comprueba la expiración del access token antes de cada petición. Si queda menos de 30 segundos de vida, usa el refresh token para obtener uno nuevo de forma transparente. Si el refresh token también ha expirado, la petición falla con `ErrorCode.UNAUTHORIZED` (HTTP 401): ```python from utilia_sdk import UtiliaSDKError, ErrorCode try: tickets = await sdk.tickets.list("user-456") except UtiliaSDKError as error: if error.code == ErrorCode.UNAUTHORIZED: auth_url = await sdk.oauth.get_authorization_url() print(f"Sesión expirada. Autoriza de nuevo: {auth_url}") ``` ```python from utilia_sdk import UtiliaSDKError, ErrorCode try: tickets = sdk.tickets.list("user-456") except UtiliaSDKError as error: if error.code == ErrorCode.UNAUTHORIZED: auth_url = sdk.oauth.get_authorization_url() print(f"Sesión expirada. Autoriza de nuevo: {auth_url}") ``` El enum `ErrorCode` incluye el valor `AUTH_EXPIRED`, pero hoy ningún punto del SDK lo produce: una sesión OAuth caducada llega siempre como `UNAUTHORIZED`. Comparar `error.code` contra `ErrorCode.AUTH_EXPIRED` no se cumple nunca y deja al usuario sin el aviso de reautorizar. ## Ejemplo completo: script CLI (async) ```python import asyncio from utilia_sdk import UtiliaSDK, FileTokenStorage async def main(): sdk = UtiliaSDK( base_url="https://os.utilia.ai/api", oauth={ "client_id": "client_xxxxxxxxxx", "client_secret": "secret_xxxxxxxxxx", "redirect_uri": "http://localhost:8000/callback", "scopes": ["openid", "profile", "email", "dev:read"], "token_storage": FileTokenStorage(".utilia-tokens.json"), }, ) if await sdk.oauth.is_authenticated(): user = await sdk.oauth.get_user_info() print(f"Autenticado como {user.name}") async with sdk: tickets = await sdk.tickets.list("user-456") print(f"Tienes {len(tickets.data)} tickets") else: auth_url = await sdk.oauth.get_authorization_url() print(f"Abre esta URL en tu navegador:\n{auth_url}") code = input("Introduce el código de autorización: ") await sdk.oauth.handle_callback(code) user = await sdk.oauth.get_user_info() print(f"Autenticado como {user.name}") asyncio.run(main()) ``` ## Ejemplo completo: servidor FastAPI ```python from fastapi import FastAPI, Request from fastapi.responses import RedirectResponse from utilia_sdk import UtiliaSDK app = FastAPI() sdk = UtiliaSDK( base_url="https://os.utilia.ai/api", oauth={ "client_id": "client_xxxxxxxxxx", "client_secret": "secret_xxxxxxxxxx", "redirect_uri": "http://localhost:8000/callback", "scopes": ["openid", "profile", "email"], }, ) @app.get("/login") async def login(): auth_url = await sdk.oauth.get_authorization_url() return RedirectResponse(auth_url) @app.get("/callback") async def callback(code: str): tokens = await sdk.oauth.handle_callback(code) user = await sdk.oauth.get_user_info() return {"message": f"Bienvenido, {user.name}", "email": user.email} ``` ## Parámetro `theme` (desde v0.7.0) Los métodos `get_authorization_url()` y `get_login_url()` (tanto la versión asíncrona como la síncrona) aceptan un parámetro opcional `theme` que permite solicitar un tema visual concreto para la pantalla de consentimiento OAuth. ```python # Solicitar pantalla de consentimiento con tema oscuro auth_url = await sdk.oauth.get_authorization_url(theme="dark") # Solicitar pantalla de consentimiento con tema claro auth_url = await sdk.oauth.get_authorization_url(theme="light") ``` ```python # Solicitar pantalla de consentimiento con tema oscuro auth_url = sdk.oauth.get_authorization_url(theme="dark") # Solicitar pantalla de consentimiento con tema claro auth_url = sdk.oauth.get_authorization_url(theme="light") ``` | Valor | Descripción | |---|---| | `'light'` | Fuerza la pantalla de consentimiento en tema claro | | `'dark'` | Fuerza la pantalla de consentimiento en tema oscuro | | `None` | El tema se determina automáticamente según la preferencia del navegador del usuario | Usa este parámetro cuando tu aplicación tenga un tema fijo y quieras que la pantalla de consentimiento sea coherente visualmente con tu interfaz. ## Referencia de tipos OAuth ```python from dataclasses import dataclass from abc import ABC, abstractmethod @dataclass class OAuthConfig: client_id: str redirect_uri: str client_secret: str | None = None scopes: list[str] = field( default_factory=lambda: ["openid", "profile", "email"] ) token_storage: Any | None = None @dataclass class OAuthTokens: access_token: str token_type: str # normalmente "Bearer" expires_in: int # duración en segundos del access token expires_at: float # time.time() + expires_in refresh_token: str | None = None scope: str | None = None # cadena, NO una lista `scopes` id_token: str | None = None @dataclass class OAuthUserInfo: sub: str name: str | None = None email: str | None = None email_verified: bool | None = None picture: str | None = None given_name: str | None = None family_name: str | None = None class TokenStorage(ABC): @abstractmethod async def get_tokens(self) -> OAuthTokens | None: ... @abstractmethod async def set_tokens(self, tokens: OAuthTokens) -> None: ... @abstractmethod async def clear_tokens(self) -> None: ... # Opcionales, con implementación por defecto (flujo PKCE automático) async def get_pending_state(self) -> PendingOAuthState | None: ... async def set_pending_state(self, state: PendingOAuthState) -> None: ... async def clear_pending_state(self) -> None: ... class SyncTokenStorage(ABC): @abstractmethod def get_tokens_sync(self) -> OAuthTokens | None: ... @abstractmethod def set_tokens_sync(self, tokens: OAuthTokens) -> None: ... @abstractmethod def clear_tokens_sync(self) -> None: ... ``` Los scopes concedidos llegan en `scope`, una cadena única y opcional, no en una lista `scopes`. Leer `tokens.scopes` lanza `AttributeError`. Y si construyes un `OAuthTokens` a mano (por ejemplo en un `TokenStorage` propio), `token_type` y `expires_in` son obligatorios: omitirlos lanza `TypeError`.