"""Authentication contracts and dependency-independent public models."""
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import StrEnum
from typing import TYPE_CHECKING
from .errors import (
AuthenticationError,
AuthenticationFailedError,
ConfigurationError,
QHarborToolsError,
RemoteAccessError,
)
from .models import require_text
if TYPE_CHECKING:
from collections.abc import Sequence
[docs]
class LoginPolicy(StrEnum):
"""Control whether a remote operation may start interactive login."""
ON_DEMAND = "on_demand"
EXPLICIT = "explicit"
[docs]
@dataclass(frozen=True, slots=True)
class AuthenticationStatus:
"""Portable authentication state returned to framework users."""
provider: str
authenticated: bool
supported: bool = True
principal_id: str | None = None
server_url: str | None = None
scopes: tuple[str, ...] | Sequence[str] = ()
def __post_init__(self) -> None:
object.__setattr__(self, "provider", require_text(self.provider, "authentication.provider"))
if self.authenticated and not self.supported:
raise ConfigurationError(
"An unsupported authentication provider cannot be authenticated.",
operation="validate",
)
if isinstance(self.scopes, str):
raise ConfigurationError("authentication.scopes must be a sequence of scope names.", operation="validate")
object.__setattr__(self, "scopes", tuple(str(scope) for scope in self.scopes))
[docs]
@classmethod
def unsupported(cls, provider: str) -> AuthenticationStatus:
"""Return an unauthenticated status for a backend without login support."""
return cls(provider=provider, authenticated=False, supported=False)
[docs]
@dataclass(frozen=True, slots=True)
class TokenCredentials:
"""Secret-safe token credentials for non-interactive authentication."""
token: str = field(repr=False)
server_url: str
def __post_init__(self) -> None:
object.__setattr__(self, "token", require_text(self.token, "credentials.token"))
object.__setattr__(self, "server_url", require_text(self.server_url, "credentials.server_url"))
[docs]
class AuthenticationBackend(ABC):
"""Abstract authentication provider injected into :class:`Experiment`."""
name = "authentication"
[docs]
def status(self) -> AuthenticationStatus:
"""Return current authentication state without opening a login prompt."""
try:
return self._status()
except QHarborToolsError:
raise
except Exception as error:
raise RemoteAccessError(
"Could not determine authentication status.",
operation="authentication_status",
backend_name=self.name,
retryable=True,
) from error
[docs]
def login(self) -> AuthenticationStatus:
"""Run interactive login and verify that it completed."""
try:
status = self._login()
except QHarborToolsError:
raise
except Exception as error:
raise AuthenticationFailedError(
"Interactive authentication failed.",
operation="login",
backend_name=self.name,
retryable=True,
) from error
return self._require_authenticated(status, operation="login")
[docs]
def login_with_token(self, credentials: TokenCredentials) -> AuthenticationStatus:
"""Authenticate without an interactive prompt and verify the result."""
try:
status = self._login_with_token(credentials)
except QHarborToolsError:
raise
except Exception as error:
raise AuthenticationFailedError(
"Token authentication failed.",
operation="login_with_token",
backend_name=self.name,
retryable=True,
) from error
return self._require_authenticated(status, operation="login_with_token")
[docs]
def logout(self) -> None:
"""End the provider session."""
try:
self._logout()
except QHarborToolsError:
raise
except Exception as error:
raise AuthenticationError(
"Logout failed.",
operation="logout",
backend_name=self.name,
retryable=True,
) from error
def _require_authenticated(self, status: AuthenticationStatus, *, operation: str) -> AuthenticationStatus:
if not status.authenticated:
raise AuthenticationFailedError(
"Authentication did not complete.",
operation=operation,
backend_name=self.name,
retryable=True,
)
return status
@abstractmethod
def _status(self) -> AuthenticationStatus: ...
@abstractmethod
def _login(self) -> AuthenticationStatus: ...
def _login_with_token(self, credentials: TokenCredentials) -> AuthenticationStatus:
raise AuthenticationFailedError(
"This authentication backend does not support token login.",
operation="login_with_token",
backend_name=self.name,
)
@abstractmethod
def _logout(self) -> None: ...