Refactor cuiman authentication configuration

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

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",
 ]

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>
-    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."""