diff --git a/CHANGES.md b/CHANGES.md index 548df829..737264bd 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -1,5 +1,15 @@ ## Changes in version 0.2.1 (in development) +### Enhancements + +- **Cuiman** authentication configuration now uses distinct, nested data models + for no authentication, Basic, token, proprietary login, OAuth2, and API-key + authentication. OAuth2 grant types are typed, and the configuration and CLI + use unambiguous names such as `login_url`, `token_url`, `access_token`, and + `access_token_header`. Existing flat file configurations for unambiguous + authentication types are converted when read; legacy login configurations + using `auth_url` now instruct users to rerun `cuiman configure`. (#176) + ### Fixes - Fixed `client.show_app()` forcing a redundant interactive OAuth2/PKCE login diff --git a/cuiman/src/cuiman/api/auth/__init__.py b/cuiman/src/cuiman/api/auth/__init__.py index 3e7b2129..21f00644 100644 --- a/cuiman/src/cuiman/api/auth/__init__.py +++ b/cuiman/src/cuiman/api/auth/__init__.py @@ -1,16 +1,45 @@ # Copyright (c) 2025-2026 by the Eozilla team and contributors # Permissions are hereby granted under the terms of the Apache 2.0 License: -# https://opensource.org/license/apache-2-0. +# https://opensource.org/license/apache-2.0. -from cuiman.api.auth.config import AuthConfig, AuthType -from cuiman.api.auth.login import LoginResult, login, login_for_tokens -from cuiman.api.auth.login_async import login_async +from cuiman.api.auth.config import ( + ApiKeyAuthConfig, + AuthConfig, + AuthConfigBase, + AuthType, + BasicAuthConfig, + LoginAuthConfig, + NoAuthConfig, + OAuth2AuthConfig, + OAuth2GrantType, + TokenAuthConfig, +) +from cuiman.api.auth.login import TokenResult, login, login_for_tokens +from cuiman.api.auth.login_async import login_async, login_async_for_tokens +from cuiman.api.auth.oauth2 import obtain_oauth2_tokens, renew_oauth2_tokens +from cuiman.api.auth.oauth2_async import ( + obtain_oauth2_tokens_async, + renew_oauth2_tokens_async, +) __all__ = [ + "ApiKeyAuthConfig", "AuthConfig", + "AuthConfigBase", "AuthType", - "LoginResult", + "BasicAuthConfig", + "LoginAuthConfig", + "NoAuthConfig", + "OAuth2AuthConfig", + "OAuth2GrantType", + "TokenAuthConfig", + "TokenResult", "login", "login_async", + "login_async_for_tokens", "login_for_tokens", + "obtain_oauth2_tokens", + "obtain_oauth2_tokens_async", + "renew_oauth2_tokens", + "renew_oauth2_tokens_async", ] diff --git a/cuiman/src/cuiman/api/auth/config.py b/cuiman/src/cuiman/api/auth/config.py index 217dd2bd..c1decfa6 100644 --- a/cuiman/src/cuiman/api/auth/config.py +++ b/cuiman/src/cuiman/api/auth/config.py @@ -2,160 +2,188 @@ # Permissions are hereby granted under the terms of the Apache 2.0 License: # https://opensource.org/license/apache-2-0. -from typing import Awaitable, Callable, Literal, Optional, TypeAlias, get_args +import base64 +from typing import Annotated, Awaitable, Callable, Literal, TypeAlias, get_args -from pydantic import HttpUrl, field_validator -from pydantic_settings import BaseSettings, SettingsConfigDict +from pydantic import BaseModel, ConfigDict, Field, HttpUrl, model_validator AuthType: TypeAlias = Literal[ - # No authentication required "none", - # HTTP Basic Auth. "basic", - # Static token (X-Auth-Token or Bearer) "token", - # username/password (X-Auth-Token or Bearer) "login", - # X-API-Key. + "oauth2", "api-key", ] +"""Supported authentication mechanisms.""" + +OAuth2GrantType: TypeAlias = Literal["password", "client_credentials"] +"""OAuth2 grants supported by Cuiman.""" AUTH_TYPE_NAMES: tuple[str, ...] = get_args(AuthType) +"""Names of the supported authentication mechanisms.""" -# TODO: enhance AuthConfig by adding Annotated[Optional[type], Field(info)] = None -# with info containing metadata and validation info +OAUTH2_GRANT_TYPE_NAMES: tuple[str, ...] = get_args(OAuth2GrantType) +"""Names of the supported OAuth2 grants.""" -# TODO: make all fields optional (even bool), -# do not use default values other than None, -# define default values in defaults module and use -# when required by auth_type only. +class AuthConfigBase(BaseModel): + """Base class for authentication configuration models.""" -class AuthConfig(BaseSettings): - """Authentication configuration.""" + model_config = ConfigDict(extra="forbid") - model_config = SettingsConfigDict( - extra="allow", # allow for extensions - ) + auth_type: AuthType - # Authentication type - auth_type: Optional[AuthType] = None + @property + def auth_headers(self) -> dict[str, str]: + """Return the HTTP authentication headers for this configuration.""" + return {} - # Authentication URL, usually an endpoint ending with "/auth/login". - auth_url: Optional[str] = None + def make_token_refresher(self) -> Callable[[], dict[str, str]] | None: + """Create a synchronous token refresh callback when supported.""" + return None - # For type "basic" or "login" (username/password -> token) - username: Optional[str] = None - password: Optional[str] = None + def make_async_token_refresher( + self, + ) -> Callable[[], Awaitable[dict[str, str]]] | None: + """Create an asynchronous token refresh callback when supported.""" + return None - # For type "login", initial password grant (OAuth2 Resource Owner Password Credentials) - client_id: Optional[str] = None - client_secret: Optional[str] = None - grant_type: str = "password" - # For type "login", token refresh phase — set after a successful login if the server - # returned a refresh token; presence of this field activates automatic token refresh on 401 - refresh_token: Optional[str] = None +class NoAuthConfig(AuthConfigBase): + """Configuration for APIs that require no authentication.""" - # For type "token" or "login" - token: Optional[str] = None + auth_type: Literal["none"] = "none" - # For type "token": custom header or Bearer - use_bearer: bool = True # if True → Authorization: Bearer - token_header: str = "X-Auth-Token" # noqa: S105 - # For type "api-key" - api_key: Optional[str] = None - api_key_header: str = "X-API-Key" +class BasicAuthConfig(AuthConfigBase): + """HTTP Basic authentication configuration.""" + + auth_type: Literal["basic"] = "basic" + username: str + password: str @property def auth_headers(self) -> dict[str, str]: - """ - Return the HTTP authentication headers for this auth configuration. - """ - return get_auth_headers(self) + """Return an HTTP Basic Authorization header.""" + if not self.username or not self.password: + raise ValueError("username/password required for basic authentication.") + credentials = f"{self.username}:{self.password}" + encoded = base64.b64encode(credentials.encode()).decode() + return {"Authorization": f"Basic {encoded}"} + - def _maybe_make_token_refresher(self) -> Callable[[], dict[str, str]] | None: - """Create a sync token refresh callback, or None if not applicable.""" - if self.auth_type != "login" or not self.refresh_token: - return None +class _AccessTokenAuthConfig(AuthConfigBase): + access_token: str | None = None + use_bearer: bool = True + access_token_header: str = "X-Auth-Token" # noqa: S105 + + @property + def auth_headers(self) -> dict[str, str]: + if not self.access_token: + raise ValueError("Missing access token.") + if self.use_bearer: + return {"Authorization": f"Bearer {self.access_token}"} + return {self.access_token_header: self.access_token} + + +class TokenAuthConfig(_AccessTokenAuthConfig): + """Static access-token authentication configuration.""" + + auth_type: Literal["token"] = "token" + access_token: str + + +class LoginAuthConfig(_AccessTokenAuthConfig): + """Configuration for a proprietary username/password login endpoint.""" + + auth_type: Literal["login"] = "login" + login_url: HttpUrl + username: str + password: str + + +class OAuth2AuthConfig(_AccessTokenAuthConfig): + """OAuth2 token endpoint configuration.""" + + auth_type: Literal["oauth2"] = "oauth2" + token_url: HttpUrl + grant_type: OAuth2GrantType = "password" + username: str | None = None + password: str | None = None + client_id: str | None = None + client_secret: str | None = None + refresh_token: str | None = None + + @model_validator(mode="after") + def validate_grant_credentials(self) -> "OAuth2AuthConfig": + """Validate the credentials required by the selected grant.""" + if self.grant_type == "password" and not (self.username and self.password): + raise ValueError( + "Username and password are required for the OAuth2 password grant." + ) + if self.grant_type == "client_credentials" and not ( + self.client_id and self.client_secret + ): + raise ValueError( + "Client ID and client secret are required for the OAuth2 " + "client credentials grant." + ) + return self + + def make_token_refresher(self) -> Callable[[], dict[str, str]]: + """Create a synchronous OAuth2 token renewal callback.""" def refresh() -> dict[str, str]: - from .login import refresh_login + from .oauth2 import renew_oauth2_tokens - result = refresh_login(self) - self.token = result.access_token - if result.refresh_token: + result = renew_oauth2_tokens(self) + self.access_token = result.access_token + if self.grant_type == "password" and result.refresh_token: self.refresh_token = result.refresh_token return self.auth_headers return refresh - def _make_async_token_refresher( + def make_async_token_refresher( self, - ) -> Callable[[], Awaitable[dict[str, str]]] | None: - """Create an async token refresh callback, or None if not applicable.""" - if self.auth_type != "login" or not self.refresh_token: - return None + ) -> Callable[[], Awaitable[dict[str, str]]]: + """Create an asynchronous OAuth2 token renewal callback.""" async def refresh() -> dict[str, str]: - from .login_async import refresh_login_async + from .oauth2_async import renew_oauth2_tokens_async - result = await refresh_login_async(self) - self.token = result.access_token - if result.refresh_token: + result = await renew_oauth2_tokens_async(self) + self.access_token = result.access_token + if self.grant_type == "password" and result.refresh_token: self.refresh_token = result.refresh_token return self.auth_headers return refresh - # noinspection PyMethodParameters - @field_validator("auth_url") - def validate_auth_url(cls, v: str | None) -> str | None: - return None if v is None or v == "" else str(HttpUrl(v)) - - -def get_auth_headers(config: AuthConfig) -> dict[str, str]: - """ - Returns the HTTP authentication headers for given auth type. - """ - - auth_type = config.auth_type - # Static API token - if auth_type == "token": - if not config.token: - raise ValueError("Missing API token.") +class ApiKeyAuthConfig(AuthConfigBase): + """API-key authentication configuration.""" - if config.use_bearer: - return {"Authorization": f"Bearer {config.token}"} - else: - return {config.token_header: config.token} - - # Username/password login (token acquired earlier) - if auth_type == "login": - if not config.token: - raise ValueError("Token is missing. Run CLI 'configure' first.") - if config.use_bearer: - return {"Authorization": f"Bearer {config.token}"} - return {config.token_header: config.token} + auth_type: Literal["api-key"] = "api-key" + api_key: str + api_key_header: str = "X-API-Key" - # API Key header - if auth_type == "api-key": - if not config.api_key: + @property + def auth_headers(self) -> dict[str, str]: + """Return the configured API-key header.""" + if not self.api_key: raise ValueError("api_key must be set for authentication type 'api-key'.") - return {config.api_key_header: config.api_key} - - # Basic Auth (username/password) - if auth_type == "basic": - if not (config.username and config.password): - raise ValueError("username/password required for basic authentication.") + return {self.api_key_header: self.api_key} - import base64 - creds = f"{config.username}:{config.password}" - encoded = base64.b64encode(creds.encode()).decode() - return {"Authorization": f"Basic {encoded}"} - - # Here, auth_type is either None or "none" - return {} +AuthConfig: TypeAlias = Annotated[ + NoAuthConfig + | BasicAuthConfig + | TokenAuthConfig + | LoginAuthConfig + | OAuth2AuthConfig + | ApiKeyAuthConfig, + Field(discriminator="auth_type"), +] +"""Discriminated union of authentication configuration models.""" diff --git a/cuiman/src/cuiman/api/auth/login.py b/cuiman/src/cuiman/api/auth/login.py index f238131f..62d14895 100644 --- a/cuiman/src/cuiman/api/auth/login.py +++ b/cuiman/src/cuiman/api/auth/login.py @@ -8,119 +8,53 @@ import httpx from pydantic import BaseModel -from .config import AuthConfig +from .config import LoginAuthConfig -class LoginResult(BaseModel): - """Result of a login or token refresh operation.""" +class TokenResult(BaseModel): + """Access and optional refresh tokens returned by an authentication service.""" access_token: str refresh_token: str | None = None -def login(auth_config: AuthConfig) -> Any: - """ - Performs a synchronous login (username+password → token) - and returns a token. - - Args: - auth_config: authentication configuration. - - Returns: - An access token either as JSON or plain text. - """ +def login(auth_config: LoginAuthConfig) -> str: + """Log in through a proprietary endpoint and return its access token.""" return login_for_tokens(auth_config).access_token -def login_for_tokens(auth_config: AuthConfig) -> LoginResult: - """ - Performs a synchronous login and returns both - access token and refresh token (if available). - - Args: - auth_config: authentication configuration. - - Returns: - A LoginResult with access_token and optional refresh_token. - """ +def login_for_tokens(auth_config: LoginAuthConfig) -> TokenResult: + """Log in through a proprietary endpoint and parse its token response.""" url, data = prepare_login(auth_config) with httpx.Client() as client: response = client.post(url, data=data) return process_login_response_for_tokens(response) -def refresh_login(auth_config: AuthConfig) -> LoginResult: - """ - Performs a synchronous token refresh using a refresh token. - - Args: - auth_config: authentication configuration (must have refresh_token set). - - Returns: - A LoginResult with the new access_token and optional new refresh_token. - """ - url, data = prepare_refresh(auth_config) - with httpx.Client() as client: - response = client.post(url, data=data) - return process_login_response_for_tokens(response) - - -def prepare_login(config: AuthConfig) -> tuple[str, dict[str, str | None]]: - if not config.auth_url: - raise ValueError("Authentication URL must be set.") +def prepare_login(config: LoginAuthConfig) -> tuple[str, dict[str, str]]: + """Build a proprietary username/password login request.""" if not config.username or not config.password: raise ValueError( "Username and password must be set for authentication type 'login'." ) - data = _add_client_credentials( - config, - { - "grant_type": config.grant_type, - "username": config.username, - "password": config.password, - }, - ) - return config.auth_url, data - - -def _add_client_credentials( - config: AuthConfig, data: dict[str, str | None] -) -> dict[str, str | None]: - if config.client_id: - data["client_id"] = config.client_id - if config.client_secret: - data["client_secret"] = config.client_secret - return data - - -def prepare_refresh(config: AuthConfig) -> tuple[str, dict[str, str | None]]: - if not config.auth_url: - raise ValueError("Authentication URL must be set.") - if not config.refresh_token: - raise ValueError("Refresh token must be set.") - data = _add_client_credentials( - config, - { - "grant_type": "refresh_token", - "refresh_token": config.refresh_token, - }, - ) - return config.auth_url, data - - -def process_login_response(response: httpx.Response) -> Any: + return str(config.login_url), { + "username": config.username, + "password": config.password, + } + + +def process_login_response(response: httpx.Response) -> str: + """Parse an access token from a proprietary login response.""" response.raise_for_status() - # noinspection PyBroadException try: - # Accept JSON ... token_data = response.json() - except Exception: - # ... or plain-text tokens + except Exception: # noqa: BLE001 - proprietary endpoints may return plain text token_data = response.text.strip() return parse_token(token_data) -def process_login_response_for_tokens(response: httpx.Response) -> LoginResult: +def process_login_response_for_tokens(response: httpx.Response) -> TokenResult: + """Parse access and optional refresh tokens from a login response.""" response.raise_for_status() try: token_data = response.json() @@ -130,10 +64,11 @@ def process_login_response_for_tokens(response: httpx.Response) -> LoginResult: refresh_token = None if isinstance(token_data, dict): refresh_token = token_data.get("refresh_token") - return LoginResult(access_token=access_token, refresh_token=refresh_token) + return TokenResult(access_token=access_token, refresh_token=refresh_token) def parse_token(token_data: Any) -> str: + """Extract a token string from common proprietary response shapes.""" token: Any = None if isinstance(token_data, str): token = token_data @@ -154,12 +89,7 @@ def parse_token(token_data: Any) -> str: def _find_token(token_data: dict) -> Any: - # TODO: This is a more or less generic hack. - # Either we make the path to the token configurable or we - # allow clients to pass a token-obtaining function to their - # client API configuration. - - for k in ( + for key in ( "token", "authToken", "auth_token", @@ -168,13 +98,12 @@ def _find_token(token_data: dict) -> Any: "apiToken", "api_token", ): - if k in token_data: - return token_data[k] + if key in token_data: + return token_data[key] - for v in token_data.values(): - if isinstance(v, dict): - token = _find_token(v) + for value in token_data.values(): + if isinstance(value, dict): + token = _find_token(value) if token is not None: return token - return None diff --git a/cuiman/src/cuiman/api/auth/login_async.py b/cuiman/src/cuiman/api/auth/login_async.py index 7df7a6ca..09272a69 100644 --- a/cuiman/src/cuiman/api/auth/login_async.py +++ b/cuiman/src/cuiman/api/auth/login_async.py @@ -2,61 +2,20 @@ # Permissions are hereby granted under the terms of the Apache 2.0 License: # https://opensource.org/license/apache-2-0. -from typing import Any - import httpx -from .config import AuthConfig -from .login import ( - LoginResult, - prepare_login, - prepare_refresh, - process_login_response_for_tokens, -) - - -async def login_async(auth_config: AuthConfig) -> Any: - """ - Performs an asynchronous login (username+password → token) - and returns a token. +from .config import LoginAuthConfig +from .login import TokenResult, prepare_login, process_login_response_for_tokens - Args: - auth_config: authentication configuration. - Returns: - An access token either as JSON or plain text. - """ +async def login_async(auth_config: LoginAuthConfig) -> str: + """Asynchronously log in and return a proprietary access token.""" return (await login_async_for_tokens(auth_config)).access_token -async def login_async_for_tokens(auth_config: AuthConfig) -> LoginResult: - """ - Performs an asynchronous login and returns both - access token and refresh token (if available). - - Args: - auth_config: authentication configuration. - - Returns: - A LoginResult with access_token and optional refresh_token. - """ +async def login_async_for_tokens(auth_config: LoginAuthConfig) -> TokenResult: + """Asynchronously log in and parse the token response.""" url, data = prepare_login(auth_config) async with httpx.AsyncClient() as client: response = await client.post(url, data=data) return process_login_response_for_tokens(response) - - -async def refresh_login_async(auth_config: AuthConfig) -> LoginResult: - """ - Performs an asynchronous token refresh using a refresh token. - - Args: - auth_config: authentication configuration (must have refresh_token set). - - Returns: - A LoginResult with the new access_token and optional new refresh_token. - """ - url, data = prepare_refresh(auth_config) - async with httpx.AsyncClient() as client: - response = await client.post(url, data=data) - return process_login_response_for_tokens(response) diff --git a/cuiman/src/cuiman/api/auth/oauth2.py b/cuiman/src/cuiman/api/auth/oauth2.py new file mode 100644 index 00000000..c73d5605 --- /dev/null +++ b/cuiman/src/cuiman/api/auth/oauth2.py @@ -0,0 +1,77 @@ +# Copyright (c) 2026 by the Eozilla team and contributors +# Permissions are hereby granted under the terms of the Apache 2.0 License: +# https://opensource.org/license/apache-2-0. + +from typing import Any + +import httpx + +from .config import OAuth2AuthConfig +from .login import TokenResult + + +def obtain_oauth2_tokens(auth_config: OAuth2AuthConfig) -> TokenResult: + """Obtain OAuth2 tokens using the configured grant.""" + url, data = prepare_oauth2_token_request(auth_config) + with httpx.Client() as client: + response = client.post(url, data=data) + return process_oauth2_token_response(response) + + +def renew_oauth2_tokens(auth_config: OAuth2AuthConfig) -> TokenResult: + """Refresh or reacquire OAuth2 tokens according to the configured grant.""" + url, data = prepare_oauth2_renewal_request(auth_config) + with httpx.Client() as client: + response = client.post(url, data=data) + return process_oauth2_token_response(response) + + +def prepare_oauth2_token_request( + config: OAuth2AuthConfig, +) -> tuple[str, dict[str, str]]: + """Build an OAuth2 token request for the configured grant.""" + data: dict[str, str] = {"grant_type": config.grant_type} + if config.grant_type == "password": + assert config.username is not None + assert config.password is not None + data.update(username=config.username, password=config.password) + _add_client_credentials(config, data) + return str(config.token_url), data + + +def prepare_oauth2_renewal_request( + config: OAuth2AuthConfig, +) -> tuple[str, dict[str, str]]: + """Build a grant-aware OAuth2 token renewal request.""" + if config.grant_type == "password" and config.refresh_token: + data = { + "grant_type": "refresh_token", + "refresh_token": config.refresh_token, + } + _add_client_credentials(config, data) + return str(config.token_url), data + return prepare_oauth2_token_request(config) + + +def _add_client_credentials(config: OAuth2AuthConfig, data: dict[str, str]) -> None: + if config.client_id: + data["client_id"] = config.client_id + if config.client_secret: + data["client_secret"] = config.client_secret + + +def process_oauth2_token_response(response: httpx.Response) -> TokenResult: + """Parse a standards-based OAuth2 token response.""" + response.raise_for_status() + token_data: Any = response.json() + if not isinstance(token_data, dict): + raise RuntimeError("OAuth2 token response must be a JSON object.") + access_token = token_data.get("access_token") + refresh_token = token_data.get("refresh_token") + if not isinstance(access_token, str) or not access_token: + raise RuntimeError( + "OAuth2 token response must contain a non-empty string access_token." + ) + if refresh_token is not None and not isinstance(refresh_token, str): + raise RuntimeError("OAuth2 refresh_token must be a string when present.") + return TokenResult(access_token=access_token, refresh_token=refresh_token) diff --git a/cuiman/src/cuiman/api/auth/oauth2_async.py b/cuiman/src/cuiman/api/auth/oauth2_async.py new file mode 100644 index 00000000..c898d565 --- /dev/null +++ b/cuiman/src/cuiman/api/auth/oauth2_async.py @@ -0,0 +1,33 @@ +# Copyright (c) 2026 by the Eozilla team and contributors +# Permissions are hereby granted under the terms of the Apache 2.0 License: +# https://opensource.org/license/apache-2-0. + +import httpx + +from .config import OAuth2AuthConfig +from .login import TokenResult +from .oauth2 import ( + prepare_oauth2_renewal_request, + prepare_oauth2_token_request, + process_oauth2_token_response, +) + + +async def obtain_oauth2_tokens_async( + auth_config: OAuth2AuthConfig, +) -> TokenResult: + """Asynchronously obtain OAuth2 tokens using the configured grant.""" + url, data = prepare_oauth2_token_request(auth_config) + async with httpx.AsyncClient() as client: + response = await client.post(url, data=data) + return process_oauth2_token_response(response) + + +async def renew_oauth2_tokens_async( + auth_config: OAuth2AuthConfig, +) -> TokenResult: + """Asynchronously refresh or reacquire OAuth2 tokens.""" + url, data = prepare_oauth2_renewal_request(auth_config) + async with httpx.AsyncClient() as client: + response = await client.post(url, data=data) + return process_oauth2_token_response(response) diff --git a/cuiman/src/cuiman/api/config.py b/cuiman/src/cuiman/api/config.py index 24cb9514..1a105452 100644 --- a/cuiman/src/cuiman/api/config.py +++ b/cuiman/src/cuiman/api/config.py @@ -7,6 +7,7 @@ from typing import ( Annotated, Any, + Awaitable, Callable, ClassVar, Optional, @@ -15,16 +16,16 @@ import yaml from pydantic import Field, HttpUrl, field_validator -from pydantic_settings import BaseSettings, SettingsConfigDict +from pydantic_settings import BaseSettings, EnvSettingsSource, SettingsConfigDict from gavicore.models import InputDescription, ProcessDescription, ProcessSummary -from .auth import AuthConfig +from .auth import AuthConfig, NoAuthConfig from .defaults import DEFAULT_API_URL from .opener import JobResultOpener, JobResultOpenerRegistry -class ClientConfig(AuthConfig, BaseSettings): +class ClientConfig(BaseSettings): """Client configuration. Args: @@ -34,6 +35,7 @@ class ClientConfig(AuthConfig, BaseSettings): model_config = SettingsConfigDict( env_prefix="EOZILLA_", + env_nested_delimiter="__", extra="forbid", ) @@ -67,6 +69,26 @@ class ClientConfig(AuthConfig, BaseSettings): OGC API - Processes, Part 1 - Core. """ + auth: AuthConfig = Field(default_factory=NoAuthConfig) + """Authentication configuration selected by its ``auth_type`` field.""" + + @property + def auth_headers(self) -> dict[str, str]: + """Return the HTTP authentication headers for this client.""" + return self.auth.auth_headers + + def _maybe_make_token_refresher( + self, + ) -> Callable[[], dict[str, str]] | None: + """Create a synchronous token renewal callback when supported.""" + return self.auth.make_token_refresher() + + def _make_async_token_refresher( + self, + ) -> Callable[[], Awaitable[dict[str, str]]] | None: + """Create an asynchronous token renewal callback when supported.""" + return self.auth.make_async_token_refresher() + def _repr_json_(self): return self.model_dump(mode="json", by_alias=True), dict( root="Client configuration:" @@ -88,9 +110,11 @@ def create( if file_config is not None: _update_if_not_none(config_dict, file_config.to_dict()) - # 2. from env - env_config = cls() - _update_if_not_none(config_dict, env_config.to_dict()) + # 2. from env. Read raw settings so a partial nested auth override can + # be merged with the auth model loaded from defaults or a file before + # the discriminated union is validated. + env_config = EnvSettingsSource(cls)() + _update_if_not_none(config_dict, env_config) # 3. from config if config is not None: @@ -111,6 +135,8 @@ def from_file( with config_path_.open("rt") as stream: # Note, we may switch TOML config_dict = yaml.safe_load(stream) + if isinstance(config_dict, dict): + config_dict = _convert_legacy_file_config(config_dict) return cls.new_instance(**config_dict) def write(self, config_path: Optional[str | Path] = None) -> Path: @@ -140,13 +166,16 @@ def new_instance( return config_cls(**kwargs) def to_dict(self): - return self.model_dump( + config_dict = self.model_dump( mode="json", by_alias=True, exclude_none=True, exclude_defaults=True, exclude_unset=True, ) + if "auth" in config_dict: + config_dict["auth"]["auth_type"] = self.auth.auth_type + return config_dict # noinspection PyMethodParameters @field_validator("api_url") @@ -201,4 +230,71 @@ def get_job_result_opener_registry(cls) -> JobResultOpenerRegistry: def _update_if_not_none(target: dict[str, Any], updates: dict[str, Any]): - target.update({k: v for k, v in updates.items() if v is not None}) + for key, value in updates.items(): + if value is None: + continue + if isinstance(value, dict) and isinstance(target.get(key), dict): + _update_if_not_none(target[key], value) + else: + target[key] = value + + +_LEGACY_AUTH_KEYS = { + "api_key", + "api_key_header", + "auth_type", + "auth_url", + "client_id", + "client_secret", + "grant_type", + "password", + "refresh_token", + "token", + "token_header", + "use_bearer", + "username", +} + + +def _convert_legacy_file_config(config: dict[str, Any]) -> dict[str, Any]: + """Convert the former flat CLI auth configuration to nested auth data.""" + if "auth" in config or "auth_type" not in config: + return config + + converted = { + key: value for key, value in config.items() if key not in _LEGACY_AUTH_KEYS + } + auth_type = config.get("auth_type") or "none" + auth: dict[str, Any] = {"auth_type": auth_type} + + if auth_type == "basic": + _copy_legacy_values(auth, config, "username", "password") + elif auth_type == "token": + _copy_legacy_values( + auth, + config, + ("token", "access_token"), + "use_bearer", + ("token_header", "access_token_header"), + ) + elif auth_type == "login" and "auth_url" in config: + raise ValueError( + "Legacy configuration format detected, please run 'cuiman configure'" + ) + elif auth_type == "api-key": + _copy_legacy_values(auth, config, "api_key", "api_key_header") + + converted["auth"] = auth + return converted + + +def _copy_legacy_values( + target: dict[str, Any], + source: dict[str, Any], + *keys: str | tuple[str, str], +) -> None: + for key in keys: + source_key, target_key = key if isinstance(key, tuple) else (key, key) + value = source.get(source_key) + if value is not None: + target[target_key] = value diff --git a/cuiman/src/cuiman/app/service.py b/cuiman/src/cuiman/app/service.py index 348a20f4..65afa28f 100644 --- a/cuiman/src/cuiman/app/service.py +++ b/cuiman/src/cuiman/app/service.py @@ -6,7 +6,13 @@ from pydantic import BaseModel -from cuiman.api.config import AuthConfig, ClientConfig +from cuiman.api.auth import ( + AuthConfigBase, + LoginAuthConfig, + OAuth2AuthConfig, + TokenAuthConfig, +) +from cuiman.api.config import ClientConfig ServiceProviderType = Literal["test", "dev", "custom", "system"] ServiceProviderOption = bool | int | float | str | None @@ -40,75 +46,34 @@ def create_app_service_provider(client_config: ClientConfig) -> ServiceProvider: ) -_AUTH_TYPE_TO_APPLICABLE_KEYS: dict[str, tuple[str, ...]] = { - "none": ("auth_type",), - "basic": ("auth_type", "auth_url", "username", "password"), - "login": ( - "auth_type", - "auth_url", - "username", - "password", - "token", - "use_bearer", - "token_header", - "refresh_token", - "client_id", - "client_secret", - "grant_type", - ), - "token": ( - "auth_type", - "auth_url", - "token", - "use_bearer", - "token_header", - "refresh_token", - ), - "api-key": ("auth_type", "auth_url", "api_key", "api_key_header"), -} - - -def _effective_auth_type(client_config: ClientConfig) -> str: - """ - Resolve the auth type to forward to the app. - - ``"login"`` means the *Python* client performs a username/password login - to obtain a token. By the time this is called (e.g. from ``show_app()``), - that login has already happened and ``client_config.token`` holds a - resolved access token — the app doesn't need to (and can't) repeat that - login. Forwarding ``auth_type="login"`` verbatim makes the app treat it - like an interactive OAuth2/PKCE login, discarding the already-valid - token and forcing a redundant sign-in. Once a token is resolved, forward - it as ``"token"`` instead, so the app connects immediately. - """ - auth_type = client_config.auth_type or "none" - if auth_type == "login" and client_config.token: - return "token" - return auth_type +def _effective_app_auth(auth: AuthConfigBase) -> AuthConfigBase: + """Convert an already-resolved login or OAuth2 config to token auth.""" + if isinstance(auth, (LoginAuthConfig, OAuth2AuthConfig)) and auth.access_token: + return TokenAuthConfig( + access_token=auth.access_token, + use_bearer=auth.use_bearer, + access_token_header=auth.access_token_header, + ) + return auth def _config_to_service_options(client_config: ClientConfig) -> dict[str, Any]: """ - Convert a ClientConfig object to a JSON-serializable dict, which includes - only keywords applicable to the given ``auth_config.auth_type``. + Convert a ClientConfig object to a flat, JSON-serializable app config. """ - auth_keys = set(AuthConfig.model_fields.keys()) - effective_auth_type = _effective_auth_type(client_config) - applicable_auth_keys = _AUTH_TYPE_TO_APPLICABLE_KEYS[effective_auth_type] - auth_config_dict = { - k: v - for k, v in client_config.model_dump( - mode="json", - exclude_none=True, - ).items() - if k not in auth_keys or k in applicable_auth_keys - } - if "auth_type" in auth_config_dict: - auth_config_dict["auth_type"] = effective_auth_type - # Additional cleanup - if "token_header" in auth_config_dict and client_config.use_bearer: - del auth_config_dict["token_header"] - return {_snake_to_camel(k): v for k, v in auth_config_dict.items()} + config_dict = client_config.model_dump( + mode="json", + exclude_none=True, + exclude={"auth"}, + ) + auth_dict = _effective_app_auth(client_config.auth).model_dump( + mode="json", + exclude_none=True, + ) + if auth_dict.get("use_bearer"): + auth_dict.pop("access_token_header", None) + config_dict.update(auth_dict) + return {_snake_to_camel(k): v for k, v in config_dict.items()} def _snake_to_camel(s: str) -> str: diff --git a/cuiman/src/cuiman/cli/cli.py b/cuiman/src/cuiman/cli/cli.py index 27a97c8d..2860b87a 100644 --- a/cuiman/src/cuiman/cli/cli.py +++ b/cuiman/src/cuiman/cli/cli.py @@ -6,7 +6,7 @@ import typer.core -from cuiman.api.auth.config import AUTH_TYPE_NAMES +from cuiman.api.auth.config import AUTH_TYPE_NAMES, OAUTH2_GRANT_TYPE_NAMES from cuiman.cli.output import OutputFormat from gavicore.util.cli.group import AliasedGroup from gavicore.util.cli.parameters import ( @@ -181,11 +181,25 @@ def configure( f"({'|'.join(AUTH_TYPE_NAMES)}).", ), ] = None, - auth_url: Annotated[ + login_url: Annotated[ str | None, typer.Option( - "--auth-url", - help="The URL of the authorisation service for the API ", + "--login-url", + help="The proprietary login endpoint URL.", + ), + ] = None, + token_url: Annotated[ + str | None, + typer.Option( + "--token-url", + help="The OAuth2 token endpoint URL.", + ), + ] = None, + grant_type: Annotated[ + str | None, + typer.Option( + "--grant-type", + help=f"The OAuth2 grant type ({'|'.join(OAUTH2_GRANT_TYPE_NAMES)}).", ), ] = None, username: Annotated[ @@ -208,20 +222,20 @@ def configure( str | None, typer.Option( "--client-id", - help="OAuth2 client ID for login authentication.", + help="OAuth2 client ID.", ), ] = None, client_secret: Annotated[ str | None, typer.Option( "--client-secret", - help="OAuth2 client secret for login authentication.", + help="OAuth2 client secret.", ), ] = None, - token: Annotated[ + access_token: Annotated[ str | None, typer.Option( - "--token", + "--access-token", "-t", help="Access token.", ), @@ -233,10 +247,10 @@ def configure( help="Use bearer token?", ), ] = None, - token_header: Annotated[ + access_token_header: Annotated[ str | None, typer.Option( - "--token-header", + "--access-token-header", help="Access token header", ), ] = None, @@ -254,14 +268,16 @@ def configure( config_path=config_file, api_url=api_url, auth_type=auth_type, # type: ignore[arg-type] - auth_url=auth_url, + login_url=login_url, + token_url=token_url, + grant_type=grant_type, client_id=client_id, client_secret=client_secret, username=username, password=password, - token=token, + access_token=access_token, use_bearer=use_bearer, - token_header=token_header, + access_token_header=access_token_header, ) except ValueError as exc: typer.echo(str(exc), err=True) diff --git a/cuiman/src/cuiman/cli/config.py b/cuiman/src/cuiman/cli/config.py index aade82a2..4a81347f 100644 --- a/cuiman/src/cuiman/cli/config.py +++ b/cuiman/src/cuiman/cli/config.py @@ -4,13 +4,22 @@ import os from pathlib import Path -from typing import Any +from typing import Any, cast import typer from pydantic import BaseModel -from cuiman.api.auth import login_for_tokens -from cuiman.api.auth.config import AUTH_TYPE_NAMES, AuthConfig +from cuiman.api.auth import ( + LoginAuthConfig, + OAuth2AuthConfig, + login_for_tokens, + obtain_oauth2_tokens, +) +from cuiman.api.auth.config import ( + AUTH_TYPE_NAMES, + OAUTH2_GRANT_TYPE_NAMES, + OAuth2GrantType, +) from cuiman.api.config import ClientConfig from cuiman.api.defaults import DEFAULT_API_URL, DEFAULT_AUTH_TYPE @@ -40,11 +49,13 @@ def configure_client_with_prompt( config_path: Path | str | None = None, **cli_params: str | bool | None, ) -> Path: + previous = ClientConfig.create(config_path=config_path).to_dict() + previous_auth = previous.pop("auth", {}) ctx = _Context( cli_params=cli_params, # ClientConfig.create() merges file config with env vars, so env var values # surface as prompt defaults rather than silently bypassing prompts. - prev_params=ClientConfig.create(config_path=config_path).to_dict(), + prev_params={**previous, **previous_auth}, curr_params={}, ) @@ -54,7 +65,13 @@ def configure_client_with_prompt( if auth_type and auth_type != "none": _configure_auth_with_prompt(ctx, auth_type) - config = ClientConfig.new_instance(**ctx.curr_params) + auth_params = { + key: value for key, value in ctx.curr_params.items() if key != "api_url" + } + config = ClientConfig.new_instance( + api_url=ctx.curr_params["api_url"], + auth=auth_params, + ) return config.write(config_path=config_path) @@ -63,6 +80,8 @@ def _configure_auth_with_prompt(ctx: _Context, auth_type: str) -> None: _configure_basic_auth_with_prompt(ctx) elif auth_type == "login": _configure_login_auth_with_prompt(ctx) + elif auth_type == "oauth2": + _configure_oauth2_auth_with_prompt(ctx) elif auth_type == "token": _configure_token_auth_with_prompt(ctx) elif auth_type == "api-key": @@ -74,21 +93,31 @@ def _configure_basic_auth_with_prompt(ctx: _Context) -> None: def _configure_login_auth_with_prompt(ctx: _Context) -> None: - # TODO: add URL validator - _prompt_for_str(ctx, "auth_url", "Authentication URL", "") - _prompt_for_str(ctx, "client_id", "client ID", "") - _prompt_for_str(ctx, "client_secret", "client secret", "") + _prompt_for_str(ctx, "login_url", "Login URL", "") _configure_username_password_with_prompt(ctx) - auth_config = AuthConfig(**ctx.curr_params) + auth_config = LoginAuthConfig(**_current_auth_params(ctx)) result = login_for_tokens(auth_config) - ctx.curr_params["token"] = result.access_token + ctx.curr_params["access_token"] = result.access_token + _configure_token_type_with_prompt(ctx) + + +def _configure_oauth2_auth_with_prompt(ctx: _Context) -> None: + _prompt_for_str(ctx, "token_url", "OAuth2 token URL", "") + grant_type = _prompt_for_oauth2_grant_type(ctx) + if grant_type == "password": + _configure_username_password_with_prompt(ctx) + _prompt_for_str(ctx, "client_id", "OAuth2 client ID", "") + _prompt_for_str(ctx, "client_secret", "OAuth2 client secret", "") + auth_config = OAuth2AuthConfig(**_current_auth_params(ctx)) + result = obtain_oauth2_tokens(auth_config) + ctx.curr_params["access_token"] = result.access_token if result.refresh_token: ctx.curr_params["refresh_token"] = result.refresh_token _configure_token_type_with_prompt(ctx) def _configure_token_auth_with_prompt(ctx: _Context) -> None: - _prompt_for_str(ctx, "token", "API access token", "") + _prompt_for_str(ctx, "access_token", "API access token", "") _configure_token_type_with_prompt(ctx) @@ -110,7 +139,16 @@ def _configure_username_password_with_prompt(ctx: _Context) -> None: def _configure_token_type_with_prompt(ctx: _Context) -> None: use_bearer = _prompt_for_bool(ctx, "use_bearer", "Use bearer token?", True) if not use_bearer: - _prompt_for_str(ctx, "token_header", "Access token header", "X-Auth-Token") + _prompt_for_str( + ctx, + "access_token_header", + "Access token header", + "X-Auth-Token", + ) + + +def _current_auth_params(ctx: _Context) -> dict[str, Any]: + return {key: value for key, value in ctx.curr_params.items() if key != "api_url"} def _prompt_for_auth_type(ctx: _Context) -> str: @@ -128,6 +166,21 @@ def _prompt_for_auth_type(ctx: _Context) -> str: return auth_type +def _prompt_for_oauth2_grant_type(ctx: _Context) -> OAuth2GrantType: + grant_type = _prompt_for_str( + ctx, + "grant_type", + f"OAuth2 grant type ({'|'.join(OAUTH2_GRANT_TYPE_NAMES)})", + "password", + ).casefold() + if grant_type not in OAUTH2_GRANT_TYPE_NAMES: + raise ValueError( + f"Invalid OAuth2 grant type: {grant_type}. " + f"Expected one of: {', '.join(OAUTH2_GRANT_TYPE_NAMES)}." + ) + return cast(OAuth2GrantType, grant_type) + + def _prompt_for_str(ctx: _Context, key: str, text: str, default: str) -> str: value: str | None = ctx.cli_params.get(key) if value is None: diff --git a/cuiman/tests/api/auth/test_config.py b/cuiman/tests/api/auth/test_config.py index 2c537d95..223fdf95 100644 --- a/cuiman/tests/api/auth/test_config.py +++ b/cuiman/tests/api/auth/test_config.py @@ -2,202 +2,217 @@ # Permissions are hereby granted under the terms of the Apache 2.0 License: # https://opensource.org/license/apache-2-0. +# ruff: noqa: S105, S106 + import base64 from unittest.mock import AsyncMock, MagicMock, patch import pytest - -from cuiman.api.auth import AuthConfig -from cuiman.api.auth.login import LoginResult +from pydantic import TypeAdapter, ValidationError + +from cuiman.api.auth import ( + ApiKeyAuthConfig, + AuthConfig, + BasicAuthConfig, + LoginAuthConfig, + NoAuthConfig, + OAuth2AuthConfig, + TokenAuthConfig, + TokenResult, +) + + +@pytest.mark.parametrize( + ("data", "expected_type"), + [ + ({"auth_type": "none"}, NoAuthConfig), + ( + {"auth_type": "basic", "username": "u", "password": "p"}, + BasicAuthConfig, + ), + ({"auth_type": "token", "access_token": "t"}, TokenAuthConfig), + ( + { + "auth_type": "login", + "login_url": "https://example.test/login", + "username": "u", + "password": "p", + }, + LoginAuthConfig, + ), + ( + { + "auth_type": "oauth2", + "token_url": "https://example.test/token", + "username": "u", + "password": "p", + }, + OAuth2AuthConfig, + ), + ({"auth_type": "api-key", "api_key": "k"}, ApiKeyAuthConfig), + ], +) +def test_auth_config_discriminator(data, expected_type): + config = TypeAdapter(AuthConfig).validate_python(data) + assert isinstance(config, expected_type) + + +def test_auth_config_rejects_fields_from_another_auth_type(): + with pytest.raises(ValidationError, match="token_url"): + TypeAdapter(AuthConfig).validate_python( + {"auth_type": "none", "token_url": "https://example.test/token"} + ) + + +def test_no_auth_headers(): + assert NoAuthConfig().auth_headers == {} + + +def test_basic_auth_headers(): + config = BasicAuthConfig(username="user", password="pass") + expected = base64.b64encode(b"user:pass").decode() + assert config.auth_headers == {"Authorization": f"Basic {expected}"} -def test_auth_headers_none(): - config = AuthConfig(auth_type=None) - assert config.auth_headers == {} - config = AuthConfig(auth_type="none") - assert config.auth_headers == {} +@pytest.mark.parametrize(("username", "password"), [("", "p"), ("u", "")]) +def test_basic_auth_headers_require_non_empty_credentials(username, password): + config = BasicAuthConfig(username=username, password=password) + with pytest.raises(ValueError, match="username/password required"): + _ = config.auth_headers -def test_auth_headers_token_custom_header(): - config = AuthConfig( - auth_type="token", - token="abc123", - token_header="X-Auth-Token", +def test_access_token_headers(): + assert TokenAuthConfig(access_token="abc").auth_headers == { + "Authorization": "Bearer abc" + } + assert TokenAuthConfig( + access_token="abc", use_bearer=False, - ) - assert config.auth_headers == {"X-Auth-Token": "abc123"} + access_token_header="X-Token", + ).auth_headers == {"X-Token": "abc"} -def test_auth_headers_token_bearer(): - config = AuthConfig( - auth_type="token", - token="abc123", - use_bearer=True, +def test_login_requires_access_token_for_headers(): + config = LoginAuthConfig( + login_url="https://example.test/login", + username="u", + password="p", ) - assert config.auth_headers == {"Authorization": "Bearer abc123"} + with pytest.raises(ValueError, match="Missing access token"): + _ = config.auth_headers -def test_auth_headers_login_bearer(): - config = AuthConfig(auth_type="login", token="xyz") - assert config.auth_headers == {"Authorization": "Bearer xyz"} - - -def test_auth_headers_login_custom_header(): - config = AuthConfig( - auth_type="login", - token="xyz", - use_bearer=False, - token_header="X-Token", - ) - assert config.auth_headers == {"X-Token": "xyz"} +def test_api_key_headers(): + assert ApiKeyAuthConfig(api_key="key").auth_headers == {"X-API-Key": "key"} + assert ApiKeyAuthConfig( + api_key="key", api_key_header="X-Custom-Key" + ).auth_headers == {"X-Custom-Key": "key"} -def test_auth_headers_api_key(): - config = AuthConfig( - auth_type="api-key", - api_key="mykey", - api_key_header="X-API-Key", - ) - assert config.auth_headers == {"X-API-Key": "mykey"} +def test_api_key_requires_non_empty_value(): + with pytest.raises(ValueError, match="api_key must be set"): + _ = ApiKeyAuthConfig(api_key="").auth_headers -def test_auth_headers_basic_auth(): - config = AuthConfig( - auth_type="basic", - username="user", - password="pass", - ) - headers = config.auth_headers - assert "Authorization" in headers +def test_oauth2_password_grant_requires_user_credentials(): + with pytest.raises(ValidationError, match="Username and password"): + OAuth2AuthConfig(token_url="https://example.test/token") - expected = base64.b64encode(b"user:pass").decode() - assert headers["Authorization"] == f"Basic {expected}" +def test_oauth2_client_credentials_grant_requires_client_credentials(): + with pytest.raises(ValidationError, match="Client ID and client secret"): + OAuth2AuthConfig( + token_url="https://example.test/token", + grant_type="client_credentials", + ) -def test_auth_headers_fail(): - assert_auth_headers_fail( - AuthConfig(auth_type="token", token=""), "Missing API token." - ) - assert_auth_headers_fail( - AuthConfig(auth_type="login", token=""), - "Token is missing. Run CLI 'configure' first.", - ) - assert_auth_headers_fail( - AuthConfig(auth_type="api-key", api_key=""), - "api_key must be set for authentication type 'api-key'.", - ) - assert_auth_headers_fail( - AuthConfig(auth_type="basic", username="jo", password=""), - "username/password required for basic authentication.", - ) - assert_auth_headers_fail( - AuthConfig(auth_type="basic", username="", password="123"), - "username/password required for basic authentication.", - ) +def test_non_oauth_configs_have_no_refreshers(): + config = NoAuthConfig() + assert config.make_token_refresher() is None + assert config.make_async_token_refresher() is None -def test_make_token_refresher_returns_none_when_not_login(): - config = AuthConfig(auth_type="token", token="abc") - assert config._maybe_make_token_refresher() is None - -def test_make_token_refresher_returns_none_when_no_refresh_token(): - config = AuthConfig(auth_type="login", token="abc") - assert config._maybe_make_token_refresher() is None - - -@patch("cuiman.api.auth.login.refresh_login") -def test_make_token_refresher_calls_refresh_login(mock_refresh: MagicMock): - mock_refresh.return_value = LoginResult( - access_token="new-token", refresh_token="new-refresh" +@patch("cuiman.api.auth.oauth2.renew_oauth2_tokens") +def test_oauth2_refresher_updates_tokens(mock_renew: MagicMock): + mock_renew.return_value = TokenResult( + access_token="new-access", refresh_token="new-refresh" ) - config = AuthConfig( - auth_type="login", - auth_url="https://acme.com/token", - token="old-token", + config = OAuth2AuthConfig( + token_url="https://example.test/token", + username="u", + password="p", + access_token="old-access", refresh_token="old-refresh", use_bearer=False, - token_header="X-Auth-Token", + access_token_header="X-Token", ) - refresher = config._maybe_make_token_refresher() - assert refresher is not None - headers = refresher() - mock_refresh.assert_called_once_with(config) - assert config.token == "new-token" + refresher = config.make_token_refresher() + + assert refresher() == {"X-Token": "new-access"} + mock_renew.assert_called_once_with(config) + assert config.access_token == "new-access" assert config.refresh_token == "new-refresh" - assert headers == {"X-Auth-Token": "new-token"} -@patch("cuiman.api.auth.login.refresh_login") -def test_make_token_refresher_without_new_refresh_token(mock_refresh: MagicMock): - mock_refresh.return_value = LoginResult( - access_token="new-token", refresh_token=None - ) - config = AuthConfig( - auth_type="login", - auth_url="https://acme.com/token", - token="old-token", +@patch("cuiman.api.auth.oauth2.renew_oauth2_tokens") +def test_oauth2_refresher_preserves_unrotated_refresh_token(mock_renew: MagicMock): + mock_renew.return_value = TokenResult(access_token="new-access") + config = OAuth2AuthConfig( + token_url="https://example.test/token", + username="u", + password="p", refresh_token="old-refresh", ) - refresher = config._maybe_make_token_refresher() - refresher() - assert config.token == "new-token" - assert config.refresh_token == "old-refresh" - - -def test_make_async_token_refresher_returns_none_when_not_login(): - config = AuthConfig(auth_type="token", token="abc") - assert config._make_async_token_refresher() is None + config.make_token_refresher()() -def test_make_async_token_refresher_returns_none_when_no_refresh_token(): - config = AuthConfig(auth_type="login", token="abc") - assert config._make_async_token_refresher() is None + assert config.refresh_token == "old-refresh" @pytest.mark.asyncio -@patch("cuiman.api.auth.login_async.refresh_login_async") -async def test_make_async_token_refresher_calls_refresh(mock_refresh: MagicMock): - mock_refresh.return_value = LoginResult( - access_token="new-token", refresh_token="new-refresh" - ) - config = AuthConfig( - auth_type="login", - auth_url="https://acme.com/token", - token="old-token", +@patch( + "cuiman.api.auth.oauth2_async.renew_oauth2_tokens_async", + new_callable=AsyncMock, +) +async def test_oauth2_async_refresher_updates_tokens(mock_renew: AsyncMock): + mock_renew.return_value = TokenResult( + access_token="new-access", refresh_token="new-refresh" + ) + config = OAuth2AuthConfig( + token_url="https://example.test/token", + username="u", + password="p", + access_token="old-access", refresh_token="old-refresh", - use_bearer=False, - token_header="X-Auth-Token", ) - refresher = config._make_async_token_refresher() - assert refresher is not None - headers = await refresher() - mock_refresh.assert_called_once_with(config) - assert config.token == "new-token" + + headers = await config.make_async_token_refresher()() + + assert headers == {"Authorization": "Bearer new-access"} assert config.refresh_token == "new-refresh" - assert headers == {"X-Auth-Token": "new-token"} @pytest.mark.asyncio -@patch("cuiman.api.auth.login_async.refresh_login_async") -async def test_make_async_token_refresher_without_new_refresh_token( - mock_refresh: MagicMock, +@patch( + "cuiman.api.auth.oauth2_async.renew_oauth2_tokens_async", + new_callable=AsyncMock, +) +async def test_client_credentials_refresher_ignores_refresh_token( + mock_renew: AsyncMock, ): - mock_refresh.return_value = LoginResult( - access_token="new-token", refresh_token=None + mock_renew.return_value = TokenResult( + access_token="new-access", refresh_token="unused-refresh" ) - config = AuthConfig( - auth_type="login", - auth_url="https://acme.com/token", - token="old-token", - refresh_token="old-refresh", + config = OAuth2AuthConfig( + token_url="https://example.test/token", + grant_type="client_credentials", + client_id="client", + client_secret="secret", + access_token="old-access", ) - refresher = config._make_async_token_refresher() - await refresher() - assert config.token == "new-token" - assert config.refresh_token == "old-refresh" + await config.make_async_token_refresher()() -def assert_auth_headers_fail(config: AuthConfig, match: str): - with pytest.raises(ValueError, match=match): - _headers = config.auth_headers + assert config.access_token == "new-access" + assert config.refresh_token is None diff --git a/cuiman/tests/api/auth/test_login.py b/cuiman/tests/api/auth/test_login.py index ac1977b8..b7891ad1 100644 --- a/cuiman/tests/api/auth/test_login.py +++ b/cuiman/tests/api/auth/test_login.py @@ -2,307 +2,125 @@ # Permissions are hereby granted under the terms of the Apache 2.0 License: # https://opensource.org/license/apache-2-0. +# ruff: noqa: S105, S106 + import json from unittest.mock import MagicMock, patch import pytest -from cuiman.api.auth import AuthConfig, login +from cuiman.api.auth import LoginAuthConfig, TokenResult, login from cuiman.api.auth.login import ( - LoginResult, login_for_tokens, parse_token, - prepare_refresh, + prepare_login, process_login_response, - refresh_login, ) -def test_login_json_response(): - cfg = AuthConfig( - auth_type="login", - auth_url="https://acme.com/api/auth/login", - client_id="my-client", - client_secret="my-secret", +def make_config(**kwargs) -> LoginAuthConfig: + return LoginAuthConfig( + login_url="https://example.test/login", username="u", password="p", + **kwargs, ) - mock_response = MagicMock() - mock_response.json.return_value = {"token": "abc123"} - mock_response.raise_for_status.return_value = None - - with patch("httpx.Client.post", return_value=mock_response) as mock_post: - token = login(cfg) - assert token == "abc123" - mock_post.assert_called_once_with( - "https://acme.com/api/auth/login", - data={ - "grant_type": "password", - "username": "u", - "password": "p", - "client_id": "my-client", - "client_secret": "my-secret", - }, - ) +def test_prepare_login_uses_proprietary_payload(): + url, data = prepare_login(make_config()) + assert url == "https://example.test/login" + assert data == {"username": "u", "password": "p"} -def test_login_plaintext_response(): - cfg = AuthConfig( - auth_type="login", - auth_url="https://acme.com/api/auth/login", - client_id="my-client", - client_secret="my-secret", - username="u", - password="p", - ) +@pytest.mark.parametrize(("username", "password"), [("", "p"), ("u", "")]) +def test_prepare_login_requires_credentials(username, password): + with pytest.raises(ValueError, match="Username and password"): + prepare_login( + LoginAuthConfig( + login_url="https://example.test/login", + username=username, + password=password, + ) + ) - mock_response = MagicMock() - mock_response.json.side_effect = json.JSONDecodeError("not json", "", 0) - mock_response.text = "plaintext-token" - mock_response.raise_for_status.return_value = None - with patch("httpx.Client.post", return_value=mock_response): - token = login(cfg) - - assert token == "plaintext-token" +def test_login_json_response(): + response = MagicMock() + response.json.return_value = {"token": "abc123"} + with patch("httpx.Client.post", return_value=response) as post: + token = login(make_config()) -def test_login_without_client_credentials(): - cfg = AuthConfig( - auth_type="login", - auth_url="https://acme.com/api/auth/login", - username="u", - password="p", + assert token == "abc123" + post.assert_called_once_with( + "https://example.test/login", + data={"username": "u", "password": "p"}, ) - mock_response = MagicMock() - mock_response.json.return_value = {"token": "abc123"} - mock_response.raise_for_status.return_value = None - with patch("httpx.Client.post", return_value=mock_response) as mock_post: - token = login(cfg) +def test_login_plaintext_response(): + response = MagicMock() + response.json.side_effect = json.JSONDecodeError("not json", "", 0) + response.text = "plaintext-token" - assert token == "abc123" - mock_post.assert_called_once_with( - "https://acme.com/api/auth/login", - data={ - "grant_type": "password", - "username": "u", - "password": "p", - }, - ) + with patch("httpx.Client.post", return_value=response): + assert login(make_config()) == "plaintext-token" -def test_login_missing_auth_url(): - cfg = AuthConfig( - auth_type="login", - auth_url=None, - username="max", - password="1234", - ) +def test_login_for_tokens_parses_optional_refresh_token(): + response = MagicMock() + response.json.return_value = { + "access_token": "access", + "refresh_token": "refresh", + } + with patch("httpx.Client.post", return_value=response): + result = login_for_tokens(make_config()) - with pytest.raises(ValueError, match="Authentication URL must be set."): - login(cfg) + assert result == TokenResult(access_token="access", refresh_token="refresh") -def test_login_missing_user_pass(): - cfg = AuthConfig( - auth_type="login", - auth_url="https://acme.com/api/auth/login", - username=None, - password=None, - ) +def test_process_login_response(): + response = MagicMock() + response.json.return_value = {"authToken": "abc"} + assert process_login_response(response) == "abc" - with pytest.raises( - ValueError, - match="Username and password must be set for authentication type 'login'.", - ): - login(cfg) + +def test_process_login_response_plaintext(): + response = MagicMock() + response.json.side_effect = ValueError("not json") + response.text = " abc " + assert process_login_response(response) == "abc" -def test_parse_token_data_ok(): +def test_parse_token_common_shapes(): assert parse_token("a1b2") == "a1b2" assert parse_token({"token": "123"}) == "123" assert parse_token({"auth_token": "abc"}) == "abc" assert parse_token({"data": {"authToken": "xyz"}}) == "xyz" assert parse_token({"apiToken": "abc"}) == "abc" assert parse_token({"data": {"accessToken": "xyz"}}) == "xyz" + assert parse_token( + { + "metadata": "ignored", + "empty": {"value": 42}, + "data": {"access_token": "later-token"}, + } + ) == "later-token" -def test_parse_token_data_fail(): - with pytest.raises( - RuntimeError, - match="Login succeeded, but token returned by server has wrong type.", - ): - parse_token(137) +@pytest.mark.parametrize("token_data", [137, {"accessToken": True}]) +def test_parse_token_rejects_wrong_type(token_data): + with pytest.raises(RuntimeError, match="wrong type"): + parse_token(token_data) - with pytest.raises( - RuntimeError, - match="Login succeeded, but token returned by server has wrong type.", - ): - parse_token({"accessToken": True}) - with pytest.raises( - RuntimeError, match="Login succeeded, but no token has been returned by server." - ): +def test_parse_token_rejects_missing_token(): + with pytest.raises(RuntimeError, match="no token"): parse_token({}) - with pytest.raises( - RuntimeError, match="Login succeeded, but token returned by server is empty." - ): - parse_token("") - with pytest.raises( - RuntimeError, match="Login succeeded, but token returned by server is empty." - ): - parse_token({"token": ""}) - - -def test_login_for_tokens_with_refresh_token(): - cfg = AuthConfig( - auth_type="login", - auth_url="https://acme.com/api/auth/login", - client_id="my-client", - client_secret="my-secret", - username="u", - password="p", - ) - - mock_response = MagicMock() - mock_response.json.return_value = { - "access_token": "new-access", - "refresh_token": "new-refresh", - } - mock_response.raise_for_status.return_value = None - - with patch("httpx.Client.post", return_value=mock_response): - result = login_for_tokens(cfg) - - assert isinstance(result, LoginResult) - assert result.access_token == "new-access" - assert result.refresh_token == "new-refresh" - - -def test_login_for_tokens_without_refresh_token(): - cfg = AuthConfig( - auth_type="login", - auth_url="https://acme.com/api/auth/login", - username="u", - password="p", - ) - - mock_response = MagicMock() - mock_response.json.return_value = {"access_token": "only-access"} - mock_response.raise_for_status.return_value = None - - with patch("httpx.Client.post", return_value=mock_response): - result = login_for_tokens(cfg) - - assert result.access_token == "only-access" - assert result.refresh_token is None - - -def test_prepare_refresh(): - cfg = AuthConfig( - auth_type="login", - auth_url="https://acme.com/token", - client_id="my-client", - client_secret="my-secret", - refresh_token="old-refresh", - ) - - url, data = prepare_refresh(cfg) - assert url == "https://acme.com/token" - assert data == { - "grant_type": "refresh_token", - "refresh_token": "old-refresh", - "client_id": "my-client", - "client_secret": "my-secret", - } - - -def test_prepare_refresh_without_client_credentials(): - cfg = AuthConfig( - auth_type="login", - auth_url="https://acme.com/token", - refresh_token="old-refresh", - ) - - url, data = prepare_refresh(cfg) - assert data == { - "grant_type": "refresh_token", - "refresh_token": "old-refresh", - } - - -def test_prepare_refresh_missing_refresh_token(): - cfg = AuthConfig( - auth_type="login", - auth_url="https://acme.com/token", - ) - - with pytest.raises(ValueError, match="Refresh token must be set"): - prepare_refresh(cfg) - - -def test_prepare_refresh_missing_auth_url(): - cfg = AuthConfig( - auth_type="login", - auth_url=None, - refresh_token="some-token", - ) - with pytest.raises(ValueError, match="Authentication URL must be set."): - prepare_refresh(cfg) - - -def test_process_login_response_json(): - mock_response = MagicMock() - mock_response.json.return_value = {"token": "abc123"} - mock_response.raise_for_status.return_value = None - - token = process_login_response(mock_response) - assert token == "abc123" - - -def test_process_login_response_plaintext(): - mock_response = MagicMock() - mock_response.json.side_effect = json.JSONDecodeError("not json", "", 0) - mock_response.text = " plain-token " - mock_response.raise_for_status.return_value = None - - token = process_login_response(mock_response) - assert token == "plain-token" - - -def test_refresh_login(): - cfg = AuthConfig( - auth_type="login", - auth_url="https://acme.com/token", - client_id="my-client", - client_secret="my-secret", - refresh_token="old-refresh", - ) - - mock_response = MagicMock() - mock_response.json.return_value = { - "access_token": "refreshed-access", - "refresh_token": "rotated-refresh", - } - mock_response.raise_for_status.return_value = None - - with patch("httpx.Client.post", return_value=mock_response) as mock_post: - result = refresh_login(cfg) - - assert result.access_token == "refreshed-access" - assert result.refresh_token == "rotated-refresh" - mock_post.assert_called_once_with( - "https://acme.com/token", - data={ - "grant_type": "refresh_token", - "refresh_token": "old-refresh", - "client_id": "my-client", - "client_secret": "my-secret", - }, - ) +@pytest.mark.parametrize("token_data", ["", {"token": ""}]) +def test_parse_token_rejects_empty_token(token_data): + with pytest.raises(RuntimeError, match="empty"): + parse_token(token_data) diff --git a/cuiman/tests/api/auth/test_login_async.py b/cuiman/tests/api/auth/test_login_async.py index ca3d53c4..66bd6b11 100644 --- a/cuiman/tests/api/auth/test_login_async.py +++ b/cuiman/tests/api/auth/test_login_async.py @@ -2,134 +2,53 @@ # Permissions are hereby granted under the terms of the Apache 2.0 License: # https://opensource.org/license/apache-2-0. +# ruff: noqa: S105, S106 + import json from unittest.mock import AsyncMock, MagicMock, patch import pytest -from cuiman.api.auth import AuthConfig, login_async -from cuiman.api.auth.login import LoginResult -from cuiman.api.auth.login_async import login_async_for_tokens, refresh_login_async +from cuiman.api.auth import LoginAuthConfig, TokenResult, login_async +from cuiman.api.auth.login_async import login_async_for_tokens -@pytest.mark.asyncio -async def test_login_async_json(): - cfg = AuthConfig( - auth_type="login", - auth_url="https://acme.com/api/auth/login", - client_id="my-client", - client_secret="my-secret", +def make_config() -> LoginAuthConfig: + return LoginAuthConfig( + login_url="https://example.test/login", username="u", password="p", ) - # mock AsyncClient.post - mock_response = MagicMock() - mock_response.json.return_value = {"token": "abc123"} - mock_response.raise_for_status.return_value = None - - # noinspection PyUnusedLocal - async def fake_post(url, data): - return mock_response - - with patch("httpx.AsyncClient.post", new=AsyncMock(side_effect=fake_post)): - token = await login_async(cfg) - - assert token == "abc123" - @pytest.mark.asyncio -async def test_login_async_plaintext(): - cfg = AuthConfig( - auth_type="login", - auth_url="https://acme.com/api/auth/login", - client_id="my-client", - client_secret="my-secret", - username="u", - password="p", - ) - - mock_response = MagicMock() - mock_response.json.side_effect = json.JSONDecodeError("not json", "", 0) - mock_response.text = "plaintext-token" - mock_response.raise_for_status.return_value = None - - # noinspection PyUnusedLocal - async def fake_post(url, data): - return mock_response - - with patch("httpx.AsyncClient.post", new=AsyncMock(side_effect=fake_post)): - token = await login_async(cfg) +async def test_login_async_json(): + response = MagicMock() + response.json.return_value = {"token": "abc123"} - assert token == "plaintext-token" + with patch("httpx.AsyncClient.post", new=AsyncMock(return_value=response)): + assert await login_async(make_config()) == "abc123" @pytest.mark.asyncio -async def test_login_async_missing_credentials(): - cfg = AuthConfig( - auth_type="login", - auth_url="https://acme.com/api/auth/login", - username=None, - password=None, - ) +async def test_login_async_plaintext(): + response = MagicMock() + response.json.side_effect = json.JSONDecodeError("not json", "", 0) + response.text = "plaintext-token" - with pytest.raises(ValueError): - await login_async(cfg) + with patch("httpx.AsyncClient.post", new=AsyncMock(return_value=response)): + assert await login_async(make_config()) == "plaintext-token" @pytest.mark.asyncio async def test_login_async_for_tokens(): - cfg = AuthConfig( - auth_type="login", - auth_url="https://acme.com/api/auth/login", - client_id="my-client", - client_secret="my-secret", - username="u", - password="p", - ) - - mock_response = MagicMock() - mock_response.json.return_value = { - "access_token": "new-access", - "refresh_token": "new-refresh", + response = MagicMock() + response.json.return_value = { + "access_token": "access", + "refresh_token": "refresh", } - mock_response.raise_for_status.return_value = None - - # noinspection PyUnusedLocal - async def fake_post(url, data): - return mock_response - - with patch("httpx.AsyncClient.post", new=AsyncMock(side_effect=fake_post)): - result = await login_async_for_tokens(cfg) - - assert isinstance(result, LoginResult) - assert result.access_token == "new-access" - assert result.refresh_token == "new-refresh" - - -@pytest.mark.asyncio -async def test_refresh_login_async(): - cfg = AuthConfig( - auth_type="login", - auth_url="https://acme.com/token", - client_id="my-client", - client_secret="my-secret", - refresh_token="old-refresh", - ) - - mock_response = MagicMock() - mock_response.json.return_value = { - "access_token": "refreshed-access", - "refresh_token": "rotated-refresh", - } - mock_response.raise_for_status.return_value = None - - # noinspection PyUnusedLocal - async def fake_post(url, data): - return mock_response - with patch("httpx.AsyncClient.post", new=AsyncMock(side_effect=fake_post)): - result = await refresh_login_async(cfg) + with patch("httpx.AsyncClient.post", new=AsyncMock(return_value=response)): + result = await login_async_for_tokens(make_config()) - assert result.access_token == "refreshed-access" - assert result.refresh_token == "rotated-refresh" + assert result == TokenResult(access_token="access", refresh_token="refresh") diff --git a/cuiman/tests/api/auth/test_oauth2.py b/cuiman/tests/api/auth/test_oauth2.py new file mode 100644 index 00000000..7975a6dc --- /dev/null +++ b/cuiman/tests/api/auth/test_oauth2.py @@ -0,0 +1,136 @@ +# Copyright (c) 2026 by the Eozilla team and contributors +# Permissions are hereby granted under the terms of the Apache 2.0 License: +# https://opensource.org/license/apache-2-0. + +# ruff: noqa: S105, S106 + +from unittest.mock import MagicMock, patch + +import pytest + +from cuiman.api.auth import OAuth2AuthConfig, TokenResult +from cuiman.api.auth.oauth2 import ( + obtain_oauth2_tokens, + prepare_oauth2_renewal_request, + prepare_oauth2_token_request, + process_oauth2_token_response, + renew_oauth2_tokens, +) + + +def password_config(**kwargs) -> OAuth2AuthConfig: + return OAuth2AuthConfig( + token_url="https://identity.example.test/token", + username="u", + password="p", + client_id="client", + client_secret="secret", + **kwargs, + ) + + +def client_config(**kwargs) -> OAuth2AuthConfig: + return OAuth2AuthConfig( + token_url="https://identity.example.test/token", + grant_type="client_credentials", + client_id="client", + client_secret="secret", + **kwargs, + ) + + +def test_prepare_password_grant(): + url, data = prepare_oauth2_token_request(password_config()) + assert url == "https://identity.example.test/token" + assert data == { + "grant_type": "password", + "username": "u", + "password": "p", + "client_id": "client", + "client_secret": "secret", + } + + +def test_prepare_password_grant_without_client_credentials(): + config = OAuth2AuthConfig( + token_url="https://identity.example.test/token", + username="u", + password="p", + ) + _, data = prepare_oauth2_token_request(config) + assert data == {"grant_type": "password", "username": "u", "password": "p"} + + +def test_prepare_client_credentials_grant(): + _, data = prepare_oauth2_token_request(client_config()) + assert data == { + "grant_type": "client_credentials", + "client_id": "client", + "client_secret": "secret", + } + + +def test_password_grant_renewal_uses_refresh_token(): + _, data = prepare_oauth2_renewal_request(password_config(refresh_token="refresh")) + assert data == { + "grant_type": "refresh_token", + "refresh_token": "refresh", + "client_id": "client", + "client_secret": "secret", + } + + +def test_password_grant_without_refresh_token_is_reacquired(): + assert prepare_oauth2_renewal_request(password_config()) == ( + "https://identity.example.test/token", + { + "grant_type": "password", + "username": "u", + "password": "p", + "client_id": "client", + "client_secret": "secret", + }, + ) + + +def test_client_credentials_grant_is_reacquired_on_renewal(): + assert prepare_oauth2_renewal_request(client_config())[1]["grant_type"] == ( + "client_credentials" + ) + + +def test_obtain_oauth2_tokens(): + response = MagicMock() + response.json.return_value = { + "access_token": "access", + "refresh_token": "refresh", + } + with patch("httpx.Client.post", return_value=response) as post: + result = obtain_oauth2_tokens(password_config()) + + assert result == TokenResult(access_token="access", refresh_token="refresh") + post.assert_called_once() + + +def test_renew_oauth2_tokens(): + response = MagicMock() + response.json.return_value = {"access_token": "renewed"} + with patch("httpx.Client.post", return_value=response): + result = renew_oauth2_tokens(client_config()) + assert result == TokenResult(access_token="renewed") + + +@pytest.mark.parametrize( + ("payload", "message"), + [ + (["not", "an", "object"], "JSON object"), + ({}, "access_token"), + ({"access_token": ""}, "access_token"), + ({"access_token": "a", "refresh_token": 42}, "refresh_token"), + ], +) +def test_process_oauth2_token_response_rejects_invalid_payload(payload, message): + response = MagicMock() + response.json.return_value = payload + with pytest.raises(RuntimeError, match=message): + process_oauth2_token_response(response) diff --git a/cuiman/tests/api/auth/test_oauth2_async.py b/cuiman/tests/api/auth/test_oauth2_async.py new file mode 100644 index 00000000..a612eb6b --- /dev/null +++ b/cuiman/tests/api/auth/test_oauth2_async.py @@ -0,0 +1,43 @@ +# Copyright (c) 2026 by the Eozilla team and contributors +# Permissions are hereby granted under the terms of the Apache 2.0 License: +# https://opensource.org/license/apache-2-0. + +# ruff: noqa: S105, S106 + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from cuiman.api.auth import OAuth2AuthConfig, TokenResult +from cuiman.api.auth.oauth2_async import ( + obtain_oauth2_tokens_async, + renew_oauth2_tokens_async, +) + + +def make_config(**kwargs) -> OAuth2AuthConfig: + return OAuth2AuthConfig( + token_url="https://identity.example.test/token", + grant_type="client_credentials", + client_id="client", + client_secret="secret", + **kwargs, + ) + + +@pytest.mark.asyncio +async def test_obtain_oauth2_tokens_async(): + response = MagicMock() + response.json.return_value = {"access_token": "access"} + with patch("httpx.AsyncClient.post", new=AsyncMock(return_value=response)): + result = await obtain_oauth2_tokens_async(make_config()) + assert result == TokenResult(access_token="access") + + +@pytest.mark.asyncio +async def test_renew_oauth2_tokens_async(): + response = MagicMock() + response.json.return_value = {"access_token": "renewed"} + with patch("httpx.AsyncClient.post", new=AsyncMock(return_value=response)): + result = await renew_oauth2_tokens_async(make_config()) + assert result == TokenResult(access_token="renewed") diff --git a/cuiman/tests/api/test_async_client.py b/cuiman/tests/api/test_async_client.py index 5b1d9ab4..0396d0db 100644 --- a/cuiman/tests/api/test_async_client.py +++ b/cuiman/tests/api/test_async_client.py @@ -2,6 +2,8 @@ # Permissions are hereby granted under the terms of the Apache 2.0 License: # https://opensource.org/license/apache-2-0. +# ruff: noqa: S106 + import os from pathlib import Path from unittest import IsolatedAsyncioTestCase @@ -11,7 +13,7 @@ from cuiman import ClientConfig from cuiman.api.async_client import AsyncClient -from cuiman.api.auth.login import LoginResult +from cuiman.api.auth import OAuth2AuthConfig, TokenResult from gavicore.models import ( ApiError, Capabilities, @@ -104,7 +106,7 @@ def test_default_transport_is_created_from_config(self): self.assertIsNone(kwargs["async_token_refresher"]) self.assertTrue(kwargs["debug"]) - async def test_default_transport_receives_login_auth_and_refresh_callback(self): + async def test_default_transport_receives_oauth2_auth_and_refresh_callback(self): old_access = "old-access-token" old_refresh = "old-refresh-token" new_access = "new-access-token" @@ -118,9 +120,13 @@ async def test_default_transport_receives_login_auth_and_refresh_callback(self): ): client = AsyncClient( api_url="https://acme.ogc.org/api", - auth_type="login", - token=old_access, - refresh_token=old_refresh, + auth=OAuth2AuthConfig( + token_url="https://identity.acme.org/token", + username="user", + password="password", + access_token=old_access, + refresh_token=old_refresh, + ), ) _, kwargs = httpx_transport_cls.call_args @@ -132,22 +138,22 @@ async def test_default_transport_receives_login_auth_and_refresh_callback(self): self.assertIsNotNone(async_token_refresher) with patch( - "cuiman.api.auth.login_async.refresh_login_async", + "cuiman.api.auth.oauth2_async.renew_oauth2_tokens_async", new_callable=AsyncMock, - return_value=LoginResult( + return_value=TokenResult( access_token=new_access, refresh_token=new_refresh, ), - ) as refresh_login_async: + ) as renew_oauth2_tokens_async: refreshed_headers = await async_token_refresher() - refresh_login_async.assert_awaited_once_with(client.config) + renew_oauth2_tokens_async.assert_awaited_once_with(client.config.auth) self.assertEqual( {"Authorization": f"Bearer {new_access}"}, refreshed_headers, ) - self.assertEqual(new_access, client.config.token) - self.assertEqual(new_refresh, client.config.refresh_token) + self.assertEqual(new_access, client.config.auth.access_token) + self.assertEqual(new_refresh, client.config.auth.refresh_token) async def test_transport_args_for_all_endpoints(self): request = ProcessRequest(inputs={"bbox": [10, 20, 30, 40]}, outputs={}) diff --git a/cuiman/tests/api/test_client.py b/cuiman/tests/api/test_client.py index d846d0b5..705c95fd 100644 --- a/cuiman/tests/api/test_client.py +++ b/cuiman/tests/api/test_client.py @@ -2,6 +2,8 @@ # Permissions are hereby granted under the terms of the Apache 2.0 License: # https://opensource.org/license/apache-2-0. +# ruff: noqa: S106 + import os from pathlib import Path from unittest import TestCase @@ -10,7 +12,7 @@ import pytest from cuiman import ClientConfig -from cuiman.api.auth.login import LoginResult +from cuiman.api.auth import OAuth2AuthConfig, TokenResult from cuiman.api.client import Client from gavicore.models import ( ApiError, @@ -99,7 +101,7 @@ def test_default_transport_is_created_from_config(self): self.assertIsNone(kwargs["token_refresher"]) self.assertTrue(kwargs["debug"]) - def test_default_transport_receives_login_auth_and_refresh_callback(self): + def test_default_transport_receives_oauth2_auth_and_refresh_callback(self): old_access = "old-access-token" old_refresh = "old-refresh-token" new_access = "new-access-token" @@ -113,9 +115,13 @@ def test_default_transport_receives_login_auth_and_refresh_callback(self): ): client = Client( api_url="https://acme.ogc.org/api", - auth_type="login", - token=old_access, - refresh_token=old_refresh, + auth=OAuth2AuthConfig( + token_url="https://identity.acme.org/token", + username="user", + password="password", + access_token=old_access, + refresh_token=old_refresh, + ), ) _, kwargs = httpx_transport_cls.call_args @@ -127,21 +133,21 @@ def test_default_transport_receives_login_auth_and_refresh_callback(self): self.assertIsNotNone(token_refresher) with patch( - "cuiman.api.auth.login.refresh_login", - return_value=LoginResult( + "cuiman.api.auth.oauth2.renew_oauth2_tokens", + return_value=TokenResult( access_token=new_access, refresh_token=new_refresh, ), - ) as refresh_login: + ) as renew_oauth2_tokens: refreshed_headers = token_refresher() - refresh_login.assert_called_once_with(client.config) + renew_oauth2_tokens.assert_called_once_with(client.config.auth) self.assertEqual( {"Authorization": f"Bearer {new_access}"}, refreshed_headers, ) - self.assertEqual(new_access, client.config.token) - self.assertEqual(new_refresh, client.config.refresh_token) + self.assertEqual(new_access, client.config.auth.access_token) + self.assertEqual(new_refresh, client.config.auth.refresh_token) def test_transport_args_for_all_endpoints(self): request = ProcessRequest(inputs={"bbox": [10, 20, 30, 40]}, outputs={}) diff --git a/cuiman/tests/api/test_config.py b/cuiman/tests/api/test_config.py index 7d8a222d..43e3f58b 100644 --- a/cuiman/tests/api/test_config.py +++ b/cuiman/tests/api/test_config.py @@ -2,108 +2,232 @@ # Permissions are hereby granted under the terms of the Apache 2.0 License: # https://opensource.org/license/apache-2-0. +# ruff: noqa: S105, S106 + import os import tempfile from pathlib import Path from unittest import TestCase -from cuiman.api.config import ClientConfig +import yaml + +from cuiman.api.auth import ( + ApiKeyAuthConfig, + BasicAuthConfig, + LoginAuthConfig, + NoAuthConfig, + TokenAuthConfig, +) +from cuiman.api.config import ClientConfig, _update_if_not_none from cuiman.api.defaults import DEFAULT_API_URL class ClientConfigTest(TestCase): def setUp(self): self.saved_environ = { - k: v for k, v in os.environ.items() if k.startswith("EOZILLA_") + key: value + for key, value in os.environ.items() + if key.startswith("EOZILLA_") } - for k in self.saved_environ.keys(): - del os.environ[k] + for key in self.saved_environ: + del os.environ[key] def tearDown(self): - self.saved_environ = { - k: v for k, v in os.environ.items() if k.startswith("EOZILLA_") - } - for k, v in self.saved_environ.items(): - os.environ[k] = v + for key in tuple(os.environ): + if key.startswith("EOZILLA_"): + del os.environ[key] + os.environ.update(self.saved_environ) def test_ctor(self): config = ClientConfig() - self.assertEqual(None, config.api_url) - self.assertEqual(None, config.auth_url) - self.assertEqual(None, config.auth_type) + self.assertIsNone(config.api_url) + self.assertEqual(NoAuthConfig(), config.auth) def test_create_empty(self): with tempfile.TemporaryDirectory() as tmp_dir_name: - config_path = Path(tmp_dir_name) / "config" - config = ClientConfig.create(config_path=config_path) - self.assertIsInstance(config, ClientConfig) - self.assertEqual("http://127.0.0.1:8008/", config.api_url) - self.assertEqual(None, config.auth_type) + config = ClientConfig.create( + config_path=Path(tmp_dir_name) / "missing-config" + ) + self.assertEqual(DEFAULT_API_URL, config.api_url) + self.assertEqual(NoAuthConfig(), config.auth) def test_create_from_env(self): os.environ.update( - dict( - EOZILLA_API_URL="https://eozilla.pippo.api", - EOZILLA_AUTH_TYPE="none", - ) + { + "EOZILLA_API_URL": "https://eozilla.example.test", + "EOZILLA_AUTH__AUTH_TYPE": "login", + "EOZILLA_AUTH__LOGIN_URL": "https://eozilla.example.test/auth/login", + "EOZILLA_AUTH__USERNAME": "pippo", + "EOZILLA_AUTH__PASSWORD": "poppi", + "EOZILLA_AUTH__ACCESS_TOKEN": "0f8915a4", + } + ) + + config = ClientConfig() + + self.assertEqual("https://eozilla.example.test/", config.api_url) + self.assertIsInstance(config.auth, LoginAuthConfig) + self.assertEqual("pippo", config.auth.username) + self.assertEqual("poppi", config.auth.password) + self.assertEqual("0f8915a4", config.auth.access_token) + + def test_create_from_file(self): + original = ClientConfig( + api_url="https://eozilla.example.test", + auth=LoginAuthConfig( + login_url="https://eozilla.example.test/login", + username="u", + password="p", + access_token="token", + ), ) with tempfile.TemporaryDirectory() as tmp_dir_name: config_path = Path(tmp_dir_name) / "config" + original.write(config_path) config = ClientConfig.create(config_path=config_path) - self.assertIsInstance(config, ClientConfig) - self.assertEqual("https://eozilla.pippo.api/", config.api_url) - self.assertEqual("none", config.auth_type) - def test_create_from_env_with_auth(self): - os.environ.update( - dict( - EOZILLA_API_URL="https://eozilla.pippo.api/processes", - EOZILLA_AUTH_TYPE="login", - EOZILLA_AUTH_URL="https://eozilla.pippo.api/auth/login", - EOZILLA_USERNAME="pippo", - EOZILLA_PASSWORD="poppi", - EOZILLA_TOKEN="0f8915a4", - ) + self.assertEqual(original, config) + + def test_from_file_converts_legacy_flat_auth_configurations(self): + common = { + "api_url": "https://eozilla.example.test", + "api_key_header": "X-API-Key", + "grant_type": "password", + "token_header": "X-Auth-Token", + "use_bearer": True, + } + cases = [ + ( + {"auth_type": "none"}, + NoAuthConfig(), + ), + ( + { + "auth_type": "basic", + "auth_url": "https://ignored.example.test", + "username": "basic-user", + "password": "basic-password", + }, + BasicAuthConfig( + username="basic-user", + password="basic-password", + ), + ), + ( + { + "auth_type": "token", + "token": "legacy-token", + "use_bearer": False, + "token_header": "X-Legacy-Token", + }, + TokenAuthConfig( + access_token="legacy-token", + use_bearer=False, + access_token_header="X-Legacy-Token", + ), + ), + ( + { + "auth_type": "api-key", + "api_key": "legacy-key", + "api_key_header": "X-Legacy-Key", + }, + ApiKeyAuthConfig( + api_key="legacy-key", + api_key_header="X-Legacy-Key", + ), + ), + ] + + with tempfile.TemporaryDirectory() as tmp_dir_name: + for index, (legacy_auth, expected_auth) in enumerate(cases): + with self.subTest(auth_type=legacy_auth["auth_type"]): + config_path = Path(tmp_dir_name) / f"legacy-{index}.yaml" + contents = yaml.safe_dump({**common, **legacy_auth}) + config_path.write_text(contents) + + config = ClientConfig.from_file(config_path) + + self.assertIsNotNone(config) + self.assertEqual(expected_auth, config.auth) + self.assertEqual(contents, config_path.read_text()) + + def test_from_file_rejects_legacy_login_auth_configuration(self): + legacy_config = { + "api_url": "https://eozilla.example.test", + "auth_type": "login", + "auth_url": "https://identity.example.test/token", + "username": "user", + "password": "password", + } + + with tempfile.TemporaryDirectory() as tmp_dir_name: + config_path = Path(tmp_dir_name) / "config.yaml" + contents = yaml.safe_dump(legacy_config) + config_path.write_text(contents) + + with self.assertRaisesRegex( + ValueError, + "Legacy configuration format detected, please run 'cuiman configure'", + ): + ClientConfig.from_file(config_path) + + self.assertEqual(contents, config_path.read_text()) + + def test_create_merges_nested_auth_overrides(self): + original = ClientConfig( + api_url="https://eozilla.example.test", + auth=LoginAuthConfig( + login_url="https://eozilla.example.test/login", + username="u", + password="p", + access_token="old-token", + ), ) - config = ClientConfig() - self.assertIsInstance(config, ClientConfig) - self.assertEqual("https://eozilla.pippo.api/processes", config.api_url) - self.assertEqual("login", config.auth_type) - self.assertEqual("https://eozilla.pippo.api/auth/login", config.auth_url) - self.assertEqual("pippo", config.username) - self.assertEqual("poppi", config.password) - self.assertEqual("0f8915a4", config.token) - def test_create_from_file(self): - config = ClientConfig( - api_url="https://eozilla.pippo.api", + config = ClientConfig.create( + config=original, + auth={"access_token": "new-token"}, ) + + self.assertIsInstance(config.auth, LoginAuthConfig) + self.assertEqual("new-token", config.auth.access_token) + self.assertEqual("u", config.auth.username) + + def test_create_from_env_and_file(self): + os.environ["EOZILLA_API_URL"] = "https://environment.example.test" + original = ClientConfig(api_url="https://file.example.test") with tempfile.TemporaryDirectory() as tmp_dir_name: config_path = Path(tmp_dir_name) / "config" - config.write(config_path=config_path) + original.write(config_path) config = ClientConfig.create(config_path=config_path) - self.assertEqual("https://eozilla.pippo.api/", config.api_url) - self.assertEqual(None, config.auth_type) - def test_create_from_env_and_file(self): - os.environ.update( - dict( - # Should take precedence! - EOZILLA_API_URL="https://eozilla.test.api", - ) + self.assertEqual("https://environment.example.test/", config.api_url) + + def test_partial_nested_env_auth_override_is_merged_with_file(self): + original = ClientConfig( + api_url="https://file.example.test", + auth=LoginAuthConfig( + login_url="https://file.example.test/login", + username="u", + password="p", + access_token="file-token", + ), ) - config = ClientConfig(api_url="https://eozilla.pippo.api") with tempfile.TemporaryDirectory() as tmp_dir_name: config_path = Path(tmp_dir_name) / "config" - config.write(config_path=config_path) + original.write(config_path) + os.environ["EOZILLA_AUTH__ACCESS_TOKEN"] = "environment-token" config = ClientConfig.create(config_path=config_path) - self.assertEqual("https://eozilla.test.api/", config.api_url) - self.assertEqual(None, config.auth_type) + + self.assertIsInstance(config.auth, LoginAuthConfig) + self.assertEqual("environment-token", config.auth.access_token) + self.assertEqual("u", config.auth.username) def test_normalize_config_path(self): path = Path("i/am/a/path") self.assertIs(path, ClientConfig.normalize_config_path(path)) - self.assertEqual(path, ClientConfig.normalize_config_path("i/am/a/path")) + self.assertEqual(path, ClientConfig.normalize_config_path(str(path))) self.assertEqual( ClientConfig.default_path, ClientConfig.normalize_config_path("") ) @@ -111,3 +235,8 @@ def test_normalize_config_path(self): def test_default_config(self): self.assertIsInstance(ClientConfig.default_config, ClientConfig) self.assertEqual(DEFAULT_API_URL, ClientConfig.default_config.api_url) + + def test_update_if_not_none_skips_none(self): + target = {"value": "original"} + _update_if_not_none(target, {"value": None}) + self.assertEqual({"value": "original"}, target) diff --git a/cuiman/tests/app/test_service.py b/cuiman/tests/app/test_service.py index bf49a2f5..7502d957 100644 --- a/cuiman/tests/app/test_service.py +++ b/cuiman/tests/app/test_service.py @@ -2,8 +2,18 @@ # Permissions are hereby granted under the terms of the Apache 2.0 License: # https://opensource.org/license/apache-2-0. +# ruff: noqa: S106 + import pytest +from cuiman.api.auth import ( + ApiKeyAuthConfig, + BasicAuthConfig, + LoginAuthConfig, + NoAuthConfig, + OAuth2AuthConfig, + TokenAuthConfig, +) from cuiman.api.config import ClientConfig from cuiman.app.service import ( ServiceProvider, @@ -16,8 +26,12 @@ def test_create_app_service_provider(): provider = create_app_service_provider( ClientConfig( api_url="https://process.example.test/api", - auth_type="login", - auth_url="https://auth.example.test/login", + auth=LoginAuthConfig( + login_url="https://auth.example.test/login", + username="user", + password="secret", + access_token="resolved-token", + ), ) ) @@ -30,247 +44,107 @@ def test_create_app_service_provider(): ) assert provider.options == { "apiUrl": "https://process.example.test/api", - "authType": "login", - "authUrl": "https://auth.example.test/login", - "grantType": "password", + "authType": "token", + "accessToken": "resolved-token", "useBearer": True, } @pytest.mark.parametrize( - ("config_kwargs", "expected_options"), + ("auth", "expected_auth_options"), [ + pytest.param(NoAuthConfig(), {"authType": "none"}, id="none"), pytest.param( - { - "api_url": "https://process.example.test/api", - "auth_type": None, - "auth_url": "https://auth.example.test/auth", - "username": "user", - "password": "secret", - "grant_type": "password", - "token": "token-123", - "refresh_token": "refresh-123", - "use_bearer": False, - "token_header": "X-Custom-Token", - "api_key": "api-key-123", - "api_key_header": "X-Custom-Api-Key", - }, - { - "apiUrl": "https://process.example.test/api", - }, - id="auth-type-none-omits-auth-options", - ), - pytest.param( - { - "api_url": "https://process.example.test/api", - "auth_type": "none", - "auth_url": "https://auth.example.test/auth", - "username": "user", - "password": "secret", - "grant_type": "password", - "token": "token-123", - "refresh_token": "refresh-123", - "use_bearer": False, - "token_header": "X-Custom-Token", - "api_key": "api-key-123", - "api_key_header": "X-Custom-Api-Key", - }, - { - "apiUrl": "https://process.example.test/api", - "authType": "none", - }, - id="explicit-auth-type-none-keeps-auth-type", - ), - pytest.param( - { - "api_url": "https://process.example.test/api", - "auth_type": "basic", - "auth_url": "https://auth.example.test/auth", - "username": "user", - "password": "secret", - "token": "token-123", - "refresh_token": "refresh-123", - "use_bearer": False, - "token_header": "X-Custom-Token", - "api_key": "api-key-123", - "api_key_header": "X-Custom-Api-Key", - }, - { - "apiUrl": "https://process.example.test/api", - "authType": "basic", - "authUrl": "https://auth.example.test/auth", - "username": "user", - "password": "secret", - }, - id="basic-auth-keeps-basic-fields", + BasicAuthConfig(username="user", password="secret"), + {"authType": "basic", "username": "user", "password": "secret"}, + id="basic", ), pytest.param( - { - "api_url": "https://process.example.test/api", - "auth_type": "login", - "auth_url": "https://auth.example.test/auth", - "username": "user", - "password": "secret", - "token": "token-123", - "refresh_token": "refresh-123", - "client_id": "client-id", - "client_secret": "client-secret", - "grant_type": "client_credentials", - "token_header": "X-Custom-Token", - "api_key": "api-key-123", - }, - { - "apiUrl": "https://process.example.test/api", - "authType": "token", - "authUrl": "https://auth.example.test/auth", - "token": "token-123", - "refreshToken": "refresh-123", - "useBearer": True, - }, - id="login-with-resolved-token-forwards-as-token-removes-token-header", + TokenAuthConfig(access_token="token"), + {"authType": "token", "accessToken": "token", "useBearer": True}, + id="bearer-token", ), pytest.param( + TokenAuthConfig( + access_token="token", + use_bearer=False, + access_token_header="X-Custom-Token", + ), { - "api_url": "https://process.example.test/api", - "auth_type": "login", - "auth_url": "https://auth.example.test/auth", - "token": "token-123", - "use_bearer": False, - "token_header": "X-Custom-Token", - }, - { - "apiUrl": "https://process.example.test/api", "authType": "token", - "authUrl": "https://auth.example.test/auth", - "token": "token-123", + "accessToken": "token", "useBearer": False, - "tokenHeader": "X-Custom-Token", + "accessTokenHeader": "X-Custom-Token", }, - id="login-with-resolved-token-and-custom-header-forwards-as-token", + id="custom-header-token", ), pytest.param( + LoginAuthConfig( + login_url="https://auth.example.test/login", + username="user", + password="secret", + ), { - "api_url": "https://process.example.test/api", - "auth_type": "login", - "auth_url": "https://auth.example.test/auth", - "username": "user", - "password": "secret", - "client_id": "client-id", - "grant_type": "password", - }, - { - "apiUrl": "https://process.example.test/api", "authType": "login", - "authUrl": "https://auth.example.test/auth", - "grantType": "password", + "loginUrl": "https://auth.example.test/login", "username": "user", "password": "secret", - "clientId": "client-id", "useBearer": True, }, - id="login-without-resolved-token-keeps-login", + id="unresolved-login", ), pytest.param( - { - "api_url": "https://process.example.test/api", - "auth_type": "token", - "auth_url": "https://auth.example.test/auth", - "token": "token-123", - "refresh_token": "refresh-123", - "token_header": "X-Custom-Token", + OAuth2AuthConfig( + token_url="https://auth.example.test/token", + username="user", + password="secret", + client_id="client", + ), + { + "authType": "oauth2", + "tokenUrl": "https://auth.example.test/token", + "grantType": "password", "username": "user", "password": "secret", - }, - { - "apiUrl": "https://process.example.test/api", - "authType": "token", - "authUrl": "https://auth.example.test/auth", - "token": "token-123", - "refreshToken": "refresh-123", + "clientId": "client", "useBearer": True, }, - id="token-bearer-removes-token-header", + id="unresolved-oauth2", ), pytest.param( - { - "api_url": "https://process.example.test/api", - "auth_type": "token", - "auth_url": "https://auth.example.test/auth", - "token": "token-123", - "use_bearer": False, - "token_header": "X-Custom-Token", - }, - { - "apiUrl": "https://process.example.test/api", - "authType": "token", - "authUrl": "https://auth.example.test/auth", - "token": "token-123", - "useBearer": False, - "tokenHeader": "X-Custom-Token", - }, - id="token-custom-header-keeps-token-header", - ), - pytest.param( - { - "api_url": "https://process.example.test/api", - "auth_type": "api-key", - "auth_url": "https://auth.example.test/auth", - "api_key": "api-key-123", - "username": "user", - "password": "secret", - "token": "token-123", - }, - { - "apiUrl": "https://process.example.test/api", - "authType": "api-key", - "authUrl": "https://auth.example.test/auth", - "apiKey": "api-key-123", - "apiKeyHeader": "X-API-Key", - }, - id="api-key-auth-keeps-default-api-key-header", + ApiKeyAuthConfig(api_key="key", api_key_header="X-Key"), + {"authType": "api-key", "apiKey": "key", "apiKeyHeader": "X-Key"}, + id="api-key", ), ], ) -def test_create_app_service_provider_keeps_only_applicable_options( - config_kwargs, expected_options -): - provider = create_app_service_provider(ClientConfig(**config_kwargs)) - - assert provider.options == expected_options +def test_service_options_for_auth_models(auth, expected_auth_options): + provider = create_app_service_provider( + ClientConfig(api_url="https://process.example.test/api", auth=auth) + ) + assert provider.options == { + "apiUrl": "https://process.example.test/api", + **expected_auth_options, + } -def test_service_provider_models_accept_optional_metadata_and_options(): - provider = ServiceProvider( - id="dev", - meta=ServiceProviderMeta( - type="dev", - title="Development", - disabled=False, - hidden=True, - ), - options={ - "enabled": True, - "retries": 2, - "timeout": 3.5, - "label": "local", - "token": None, - }, +def test_resolved_oauth2_is_forwarded_as_token_without_credentials(): + provider = create_app_service_provider( + ClientConfig( + api_url="https://process.example.test/api", + auth=OAuth2AuthConfig( + token_url="https://auth.example.test/token", + grant_type="client_credentials", + client_id="client", + client_secret="secret", + access_token="resolved-token", + refresh_token="refresh", + ), + ) ) - - assert provider.model_dump() == { - "id": "dev", - "meta": { - "type": "dev", - "title": "Development", - "description": None, - "disabled": False, - "hidden": True, - }, - "options": { - "enabled": True, - "retries": 2, - "timeout": 3.5, - "label": "local", - "token": None, - }, + assert provider.options == { + "apiUrl": "https://process.example.test/api", + "authType": "token", + "accessToken": "resolved-token", + "useBearer": True, } diff --git a/cuiman/tests/cli/test_cli.py b/cuiman/tests/cli/test_cli.py index 35dc6dcb..a531de56 100644 --- a/cuiman/tests/cli/test_cli.py +++ b/cuiman/tests/cli/test_cli.py @@ -11,7 +11,7 @@ import yaml from cuiman import Client, __version__ -from cuiman.api.auth.login import LoginResult +from cuiman.api.auth import TokenResult # noinspection PyProtectedMember from cuiman.cli.cli import _wait_until_interrupted, cli, new_cli @@ -41,7 +41,7 @@ def test_version(self): @patch("cuiman.cli.config.login_for_tokens") def test_configure(self, mock_login): - mock_login.return_value = LoginResult( + mock_login.return_value = TokenResult( access_token="dummy-token", # noqa: S106 refresh_token="dummy-refresh", # noqa: S106 ) @@ -54,12 +54,8 @@ def test_configure(self, mock_login): "http://localhorst:2357", "--auth-type", "login", - "--auth-url", + "--login-url", "http://localhorst:2357/auth/login", - "--client-id", - "my-client", - "--client-secret", - "my-secret", "--username", "bibo", "--password", @@ -74,18 +70,15 @@ def test_configure(self, mock_login): self.assertEqual( { "api_url": "http://localhorst:2357/", - "auth_type": "login", - "auth_url": "http://localhorst:2357/auth/login", - "client_id": "my-client", - "client_secret": "my-secret", - "grant_type": "password", - "refresh_token": "dummy-refresh", - "username": "bibo", - "password": "1234", - "token": "dummy-token", - "use_bearer": True, - "token_header": "X-Auth-Token", - "api_key_header": "X-API-Key", + "auth": { + "auth_type": "login", + "login_url": "http://localhorst:2357/auth/login", + "username": "bibo", + "password": "1234", + "access_token": "dummy-token", + "use_bearer": True, + "access_token_header": "X-Auth-Token", + }, }, config, ) @@ -101,7 +94,7 @@ def test_configure_with_invalid_auth_method(self): "http://localhost:2357", "--auth-type", "torken", # INVALID - "--token", + "--access-token", "x-lkdkadf878akj134lk1lk5lk432lkk", ) self.assertEqual(1, result.exit_code, msg=self.get_result_msg(result)) diff --git a/cuiman/tests/cli/test_config.py b/cuiman/tests/cli/test_config.py index 959e50db..9e790f58 100644 --- a/cuiman/tests/cli/test_config.py +++ b/cuiman/tests/cli/test_config.py @@ -2,51 +2,47 @@ # Permissions are hereby granted under the terms of the Apache 2.0 License: # https://opensource.org/license/apache-2-0. +# ruff: noqa: S105, S106 + import os import unittest from pathlib import Path from unittest.mock import MagicMock, patch import pytest +import yaml from cuiman import ClientConfig -from cuiman.api.auth.login import LoginResult -from cuiman.api.defaults import DEFAULT_CONFIG_PATH -from cuiman.cli.config import configure_client_with_prompt, get_config -from gavicore.util.testing import set_env, set_env_cm - -DEFAULT_CONFIG_BACKUP_PATH = DEFAULT_CONFIG_PATH.parent / ( - str(DEFAULT_CONFIG_PATH.name) + ".backup" +from cuiman.api.auth import ( + ApiKeyAuthConfig, + BasicAuthConfig, + LoginAuthConfig, + NoAuthConfig, + OAuth2AuthConfig, + TokenAuthConfig, + TokenResult, +) +from cuiman.cli.config import ( + _Context, + _configure_auth_with_prompt, + configure_client_with_prompt, + get_config, ) +from gavicore.util.testing import set_env # noinspection PyAttributeOutsideInit,PyPep8Naming class ConfigTestMixin: def setUp(self): self.restore_env = set_env( - **{k: None for k, v in os.environ.items() if k.startswith("EOZILLA_")} + **{key: None for key in os.environ if key.startswith("EOZILLA_")} ) - self.must_restore_config = False - # If a config backup exists, delete it - if DEFAULT_CONFIG_BACKUP_PATH.exists(): - os.remove(DEFAULT_CONFIG_BACKUP_PATH) - # If default config exists, rename it into the backup config - if DEFAULT_CONFIG_PATH.exists(): - DEFAULT_CONFIG_PATH.rename(DEFAULT_CONFIG_BACKUP_PATH) def tearDown(self): self.restore_env() - # If config backup exists, rename it into the default - if DEFAULT_CONFIG_BACKUP_PATH.exists(): - # If default config exists, remove it, - # so we can rename the backup - if DEFAULT_CONFIG_PATH.exists(): - os.remove(DEFAULT_CONFIG_PATH) - DEFAULT_CONFIG_BACKUP_PATH.rename(DEFAULT_CONFIG_PATH) class GetConfigTest(ConfigTestMixin, unittest.TestCase): - # noinspection PyMethodMayBeStatic def test_get_config_custom(self): with pytest.raises( ValueError, @@ -54,7 +50,6 @@ def test_get_config_custom(self): ): get_config("fantasia.cfg") - # noinspection PyMethodMayBeStatic def test_get_config_no_default(self): with pytest.raises( ValueError, @@ -65,224 +60,225 @@ def test_get_config_no_default(self): ): get_config(None) + def test_get_config_rejects_legacy_cli_login_auth(self): + legacy_config = { + "api_url": "https://eozilla.example.test", + "auth_type": "login", + "auth_url": "https://identity.example.test/token", + "username": "user", + "password": "password", + "client_id": "client", + "client_secret": "secret", + "grant_type": "password", + "token": "access", + "refresh_token": "refresh", + "use_bearer": True, + "token_header": "X-Auth-Token", + "api_key_header": "X-API-Key", + } + ClientConfig.default_path.write_text(yaml.safe_dump(legacy_config)) + + with self.assertRaisesRegex( + ValueError, + "Legacy configuration format detected, please run 'cuiman configure'", + ): + get_config(None) + + self.assertEqual( + legacy_config, yaml.safe_load(ClientConfig.default_path.read_text()) + ) + class ConfigureClientWithPromptTest(ConfigTestMixin, unittest.TestCase): def assert_is_default_config_path(self, config_path: Path): - self.assertEqual(DEFAULT_CONFIG_PATH, config_path) - self.assertTrue(DEFAULT_CONFIG_PATH.exists()) + self.assertEqual(ClientConfig.default_path, config_path) + self.assertTrue(ClientConfig.default_path.exists()) + + def test_none_auth_needs_no_additional_configuration(self): + context = _Context(cli_params={}, prev_params={}, curr_params={}) + _configure_auth_with_prompt(context, "none") + self.assertEqual({}, context.curr_params) @patch("typer.prompt") - def test_auth_type_none(self, mock_prompt: MagicMock): - # Simulate sequential responses to typer.prompt() - mock_prompt.side_effect = [ - "http://localhost:9090", - "none", - ] + def test_auth_type_none(self, prompt: MagicMock): + prompt.side_effect = ["http://localhost:9090", "none"] + config_path = configure_client_with_prompt() - self.assertEqual(2, mock_prompt.call_count) + self.assert_is_default_config_path(config_path) - config = get_config(None) self.assertEqual( - ClientConfig(api_url="http://localhost:9090", auth_type="none"), - config, + ClientConfig(api_url="http://localhost:9090", auth=NoAuthConfig()), + get_config(None), ) @patch("typer.prompt") - def test_auth_type_invalid(self, mock_prompt: MagicMock): - mock_prompt.side_effect = [ - "http://localhost:9090", - "torken", - ] - - with pytest.raises( - ValueError, - match=( - r"Invalid authentication type: torken\. " - r"Expected one of:" - ), - ): + def test_auth_type_invalid(self, prompt: MagicMock): + prompt.side_effect = ["http://localhost:9090", "torken"] + with pytest.raises(ValueError, match="Invalid authentication type: torken"): configure_client_with_prompt() @patch("typer.prompt") - def test_auth_type_basic(self, mock_prompt: MagicMock): - # Simulate sequential responses to typer.prompt() - mock_prompt.side_effect = [ - "http://localhorst:9999", # api_url - "basic", # auth_type - "udo", # username - "987", # password + def test_auth_type_basic(self, prompt: MagicMock): + prompt.side_effect = [ + "http://localhorst:9999", + "basic", + "udo", + "987", ] - actual_config_path = configure_client_with_prompt() - self.assertEqual(4, mock_prompt.call_count) - self.assert_is_default_config_path(actual_config_path) - config = get_config(None) + + config_path = configure_client_with_prompt() + + self.assert_is_default_config_path(config_path) self.assertEqual( ClientConfig( api_url="http://localhorst:9999", - auth_type="basic", - username="udo", - password="987", + auth=BasicAuthConfig(username="udo", password="987"), ), - config, + get_config(None), ) @patch("cuiman.cli.config.login_for_tokens") @patch("typer.confirm") @patch("typer.prompt") def test_auth_type_login( - self, mock_prompt: MagicMock, mock_confirm: MagicMock, mock_login: MagicMock + self, prompt: MagicMock, confirm: MagicMock, login: MagicMock ): - mock_login.return_value = LoginResult( - access_token="dummy-token", refresh_token="dummy-refresh" - ) - # Simulate sequential responses to typer.prompt() - mock_prompt.side_effect = [ + login.return_value = TokenResult(access_token="dummy-token") + prompt.side_effect = [ "http://localhorst:9999", "login", "http://localhorst:9999/signin", - "my-client", - "my-secret", "bibo", "1234", - "X-Auth-Token", - ] - # Simulate response to typer.confirm() - mock_confirm.side_effect = [ - False, + "X-Custom-Token", ] - actual_config_path = configure_client_with_prompt() - mock_login.assert_called_once() - mock_confirm.assert_called_once() - self.assertEqual(8, mock_prompt.call_count) - self.assert_is_default_config_path(actual_config_path) - config = get_config(None) + confirm.return_value = False + + config_path = configure_client_with_prompt() + + self.assert_is_default_config_path(config_path) + login.assert_called_once() self.assertEqual( ClientConfig( api_url="http://localhorst:9999", - auth_type="login", - auth_url="http://localhorst:9999/signin", - client_id="my-client", - client_secret="my-secret", - username="bibo", - password="1234", - token="dummy-token", - refresh_token="dummy-refresh", - use_bearer=False, - token_header="X-Auth-Token", + auth=LoginAuthConfig( + login_url="http://localhorst:9999/signin", + username="bibo", + password="1234", + access_token="dummy-token", + use_bearer=False, + access_token_header="X-Custom-Token", + ), ), - config, + get_config(None), ) + @patch("cuiman.cli.config.obtain_oauth2_tokens") @patch("typer.confirm") @patch("typer.prompt") - def test_auth_type_token(self, mock_prompt: MagicMock, mock_confirm: MagicMock): - # Simulate sequential responses to typer.prompt() - mock_prompt.side_effect = [ + def test_oauth2_password_grant( + self, prompt: MagicMock, confirm: MagicMock, obtain: MagicMock + ): + obtain.return_value = TokenResult( + access_token="access", refresh_token="refresh" + ) + prompt.side_effect = [ "http://localhorst:9999", - "token", - "phu-8934kmnl24509kl209245902jk", - ] - # Simulate response to typer.confirm() - mock_confirm.side_effect = [ - True, + "oauth2", + "https://identity.example.test/token", + "password", + "bibo", + "1234", + "client", + "secret", ] - actual_config_path = configure_client_with_prompt() - mock_confirm.assert_called_once() - self.assertEqual(3, mock_prompt.call_count) - self.assert_is_default_config_path(actual_config_path) - config = get_config(None) + confirm.return_value = True + + configure_client_with_prompt() + + obtain.assert_called_once() self.assertEqual( - ClientConfig( - api_url="http://localhorst:9999", - auth_type="token", - token="phu-8934kmnl24509kl209245902jk", - use_bearer=True, - token_header="X-Auth-Token", + OAuth2AuthConfig( + token_url="https://identity.example.test/token", + username="bibo", + password="1234", + client_id="client", + client_secret="secret", + access_token="access", + refresh_token="refresh", ), - config, + get_config(None).auth, ) + @patch("cuiman.cli.config.obtain_oauth2_tokens") + @patch("typer.confirm") @patch("typer.prompt") - def test_auth_type_api_key(self, mock_prompt: MagicMock): - # Simulate sequential responses to typer.prompt() - mock_prompt.side_effect = [ + def test_oauth2_client_credentials_grant( + self, prompt: MagicMock, confirm: MagicMock, obtain: MagicMock + ): + obtain.return_value = TokenResult(access_token="access") + prompt.side_effect = [ "http://localhorst:9999", - "api-key", - "AB4E2629967EF3DDE", - "X-API-Key", + "oauth2", + "https://identity.example.test/token", + "client_credentials", + "client", + "secret", ] - actual_config_path = configure_client_with_prompt() - self.assertEqual(4, mock_prompt.call_count) - self.assert_is_default_config_path(actual_config_path) - config = get_config(None) + confirm.return_value = True + + configure_client_with_prompt() + self.assertEqual( - ClientConfig( - api_url="http://localhorst:9999", - auth_type="api-key", - api_key="AB4E2629967EF3DDE", - api_key_header="X-API-Key", + OAuth2AuthConfig( + token_url="https://identity.example.test/token", + grant_type="client_credentials", + client_id="client", + client_secret="secret", + access_token="access", ), - config, + get_config(None).auth, ) - @patch("cuiman.cli.config.login_for_tokens") + @patch("typer.prompt") + def test_invalid_oauth2_grant(self, prompt: MagicMock): + prompt.side_effect = [ + "http://localhorst:9999", + "oauth2", + "https://identity.example.test/token", + "magic", + ] + with pytest.raises(ValueError, match="Invalid OAuth2 grant type: magic"): + configure_client_with_prompt() + @patch("typer.confirm") @patch("typer.prompt") - def test_defaults_are_used( - self, mock_prompt: MagicMock, mock_confirm: MagicMock, mock_login: MagicMock - ): - mock_login.return_value = LoginResult( - access_token="dummy-token", refresh_token="dummy-refresh" + def test_auth_type_token(self, prompt: MagicMock, confirm: MagicMock): + prompt.side_effect = ["http://localhorst:9999", "token", "token-value"] + confirm.return_value = True + + configure_client_with_prompt() + + self.assertEqual( + TokenAuthConfig(access_token="token-value"), get_config(None).auth ) - # Use default password "9823hc!" - expected_password = "9823hc!" - with set_env_cm(EOZILLA_PASSWORD=expected_password): - # Simulate sequential responses to typer.prompt() - mock_prompt.side_effect = [ - "http://localhorst:2357", - "login", - "http://localhorst:2357/auth/login", - "my-client", - "my-secret", - "bibo", - "******", - "bibo", - "X-Auth-Token", - ] - # Simulate response to typer.confirm() - mock_confirm.side_effect = [ - True, - ] - mock_prompt.assert_not_called() - mock_confirm.assert_not_called() - mock_login.assert_not_called() - actual_config_path = configure_client_with_prompt() - self.assert_is_default_config_path(actual_config_path) - config = get_config(None) - self.assertEqual( - ClientConfig( - auth_type="login", - api_url="http://localhorst:2357", - auth_url="http://localhorst:2357/auth/login", - client_id="my-client", - client_secret="my-secret", - username="bibo", - password=expected_password, - token="dummy-token", - refresh_token="dummy-refresh", - use_bearer=True, - ), - config, - ) @patch("typer.prompt") - def test_prompt_for_pw_uses_prev_password_on_hidden_input( - self, mock_prompt: MagicMock - ): - """When user enters '******' and a previous password exists, the previous - password is reused.""" - # First configure with basic auth + password - mock_prompt.side_effect = [ + def test_auth_type_api_key(self, prompt: MagicMock): + prompt.side_effect = [ + "http://localhorst:9999", + "api-key", + "key-value", + "X-API-Key", + ] + + configure_client_with_prompt() + + self.assertEqual(ApiKeyAuthConfig(api_key="key-value"), get_config(None).auth) + + @patch("typer.prompt") + def test_prompt_for_pw_reuses_previous_password(self, prompt: MagicMock): + prompt.side_effect = [ "http://localhost:9090", "basic", "alice", @@ -290,58 +286,53 @@ def test_prompt_for_pw_uses_prev_password_on_hidden_input( ] configure_client_with_prompt() - # Now reconfigure, keeping the old password via "******" - mock_prompt.reset_mock() - mock_prompt.side_effect = [ + prompt.reset_mock() + prompt.side_effect = [ "http://localhost:9090", "basic", "alice", - "******", # sentinel → should reuse "secret123" + "******", ] config_path = configure_client_with_prompt() - config = get_config(None) - self.assertEqual("secret123", config.password) + + self.assertEqual("secret123", get_config(None).auth.password) self.assert_is_default_config_path(config_path) @patch("typer.confirm") @patch("typer.prompt") def test_prompt_for_bool_uses_env_value( - self, mock_prompt: MagicMock, mock_confirm: MagicMock + self, prompt: MagicMock, confirm: MagicMock ): - """When an env var provides a bool value, it surfaces as the default in the - confirm prompt rather than silently bypassing it, so the user can override it.""" - with set_env_cm(EOZILLA_USE_BEARER="True"): - mock_prompt.side_effect = [ + with patch.dict( + os.environ, + { + "EOZILLA_AUTH__AUTH_TYPE": "token", + "EOZILLA_AUTH__ACCESS_TOKEN": "environment-token", + "EOZILLA_AUTH__USE_BEARER": "True", + }, + ): + prompt.side_effect = [ "http://localhost:9090", "token", "my-token", ] - # confirm is still called — env var value becomes the pre-filled default - mock_confirm.return_value = True - config_path = configure_client_with_prompt() - self.assert_is_default_config_path(config_path) - config = get_config(None) - self.assertTrue(config.use_bearer) - mock_confirm.assert_called_once() - _, kwargs = mock_confirm.call_args - self.assertTrue(kwargs.get("default")) + confirm.return_value = True + + configure_client_with_prompt() + + _, kwargs = confirm.call_args + self.assertTrue(kwargs["default"]) @patch("typer.prompt") - def test_using_custom_config_path(self, mock_prompt: MagicMock): - # Simulate sequential responses to typer.prompt() - mock_prompt.side_effect = ["http://localhost:9090", "none"] + def test_using_custom_config_path(self, prompt: MagicMock): + prompt.side_effect = ["http://localhost:9090", "none"] custom_config_path = Path("my-config.cfg") try: - actual_config_path = configure_client_with_prompt( - config_path=custom_config_path - ) - self.assertEqual(2, mock_prompt.call_count) - self.assertEqual(custom_config_path, actual_config_path) - self.assertTrue(custom_config_path.exists()) - config = get_config(custom_config_path) + actual_path = configure_client_with_prompt(config_path=custom_config_path) + self.assertEqual(custom_config_path, actual_path) self.assertEqual( - ClientConfig(api_url="http://localhost:9090", auth_type="none"), - config, + ClientConfig(api_url="http://localhost:9090"), + get_config(custom_config_path), ) finally: if custom_config_path.exists(): diff --git a/cuiman/tests/conftest.py b/cuiman/tests/conftest.py new file mode 100644 index 00000000..e87c9848 --- /dev/null +++ b/cuiman/tests/conftest.py @@ -0,0 +1,18 @@ +# Copyright (c) 2026 by the Eozilla team and contributors +# Permissions are hereby granted under the terms of the Apache 2.0 License: +# https://opensource.org/license/apache-2-0. + +from pathlib import Path + +import pytest + +from cuiman.api.config import ClientConfig + + +@pytest.fixture(autouse=True) +def isolate_default_client_config( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """Prevent tests from loading a developer's real client configuration.""" + monkeypatch.setattr(ClientConfig, "default_path", tmp_path / "config")