Auth¶

class ApiKeyNotFoundError(NexusError):
    """Raised when no authentication api key is found."""

    def __init__(
        self,
        reason: Optional[str] = None,
        details: Optional[Dict[str, Any]] = None,
    ) -> None:
        """
        Initialize an ApiKeyNotFoundError.

        Args:
            reason: Optional custom error message
            details: Optional dictionary with additional context
        """
        default_reason = (
            "No authentication api key found. "
            "Use 'miura-auth token set' or set MIURA_NEXUS_API_KEY environment variable."
        )
        super().__init__(ErrorCode.AUTH_ERROR, reason or default_reason, details)

class ApiKeyExpiredError(NexusError):
    """Raised when the authentication api key has expired."""

    def __init__(
        self,
        reason: Optional[str] = None,
        details: Optional[Dict[str, Any]] = None,
    ) -> None:
        """
        Initialize an ApiKeyExpiredError.

        Args:
            reason: Optional custom error message
            details: Optional dictionary with additional context
        """
        default_reason = "Authentication api key has expired. " "Use 'miura-auth token set' to set a new api key."
        super().__init__(ErrorCode.TOKEN_EXPIRED, reason or default_reason, details)

class ApiKeyInvalidError(NexusError):
    """Raised when the authentication api key is invalid or malformed."""

    def __init__(
        self,
        reason: Optional[str] = None,
        details: Optional[Dict[str, Any]] = None,
    ) -> None:
        """
        Initialize an ApiKeyInvalidError.

        Args:
            reason: Optional custom error message
            details: Optional dictionary with additional context
        """
        default_reason = (
            "Authentication api key is invalid or malformed. " "Check the token format and ensure it's a valid JWT."
        )
        super().__init__(ErrorCode.TOKEN_INVALID, reason or default_reason, details)

class KeyringError(NexusError):
    """Raised when there's an error accessing the system keyring."""

    def __init__(
        self,
        reason: Optional[str] = None,
        details: Optional[Dict[str, Any]] = None,
    ) -> None:
        """
        Initialize a KeyringError.

        Args:
            reason: Optional custom error message
            details: Optional dictionary with additional context
        """
        default_reason = (
            "Failed to access system keyring. "
            "Install a keyring backend (e.g., keyrings.alt) or use environment variables."
        )
        super().__init__(ErrorCode.KEYRING_ERROR, reason or default_reason, details)

class AuthHandler:
    """Main authentication handler that orchestrates all auth operations."""

    def __init__(
        self,
        config_manager: Optional[ConfigurationManager] = None,
        storage_backend: Optional[str] = None,
        verify_ssl: Optional[bool] = None,
    ):
        """
        Initialize authentication handler.

        Args:
            config_manager: Optional ConfigurationManager instance
            storage_backend: Storage backend type ("file", "keyring", or "none"). Defaults to "file".
            verify_ssl: Whether to verify SSL certificates. If None, reads from
                MIURA_KEYCLOAK_VERIFY_SSL env var (default: True)
        """
        # If verify_ssl not explicitly provided, read from environment variable
        if verify_ssl is None:
            verify_ssl = True
            verify_ssl_env = os.environ.get("MIURA_KEYCLOAK_VERIFY_SSL", "").lower()
            if verify_ssl_env in ("false", "0", "no", "off"):
                verify_ssl = False

        self._config_manager = config_manager
        self.token_manager = TokenManager()
        self._keyring_manager: Optional[Any] = None  # Lazy initialization - only create when needed
        self.jwt_utils = JWTUtils()
        self.keycloak_client = KeycloakClient(config_manager=config_manager, verify_ssl=verify_ssl)

        # Determine storage backend
        if storage_backend is None:
            # Default to file storage
            storage_backend = os.environ.get("MIURA_AUTH_STORAGE", "file")

        # Create storage backend
        if storage_backend == "keyring":
            # Only create KeyringManager when using keyring backend
            from .keyring import KeyringManager

            self._keyring_manager = KeyringManager()
            self.storage_backend = create_storage_backend("keyring", self._keyring_manager)
        elif storage_backend == "none":
            self.storage_backend = create_storage_backend("none")
        else:
            # Default to file - no need to create KeyringManager
            self.storage_backend = create_storage_backend("file")

        # Keyring cache (read once per command execution)
        self._keyring_cache: Optional[Dict[str, str]] = None
        self._keyring_cache_initialized = False

    def _get_keyring_cache(self) -> Dict[str, str]:
        """
        Get keyring cache (read once per command execution).

        Returns:
            Dictionary of cached keyring values
        """
        if not self._keyring_cache_initialized:
            self._keyring_cache = {}
            # Only try to read from keyring if using keyring backend
            from .storage import KeyringStorageBackend

            if isinstance(self.storage_backend, KeyringStorageBackend):
                try:
                    # KeyringManager is guaranteed to exist when using keyring backend
                    assert self._keyring_manager is not None
                    token = self._keyring_manager.get_password("auth_api_key")
                    if token:
                        self._keyring_cache["auth_api_key"] = token
                except Exception:
                    # Keyring not available or error - cache remains empty
                    pass
            self._keyring_cache_initialized = True
        # After initialization, _keyring_cache is guaranteed to be a dict
        assert self._keyring_cache is not None
        return self._keyring_cache

    def _clear_keyring_cache(self) -> None:
        """Clear keyring cache (call on command completion)."""
        self._keyring_cache = None
        self._keyring_cache_initialized = False

    def _sync_refresh_token_from_sources(self, storage_backend_type: Optional[str] = None) -> Optional[str]:
        """
        Ensure the in-memory refresh token reflects the latest value from sources.

        Resolution order:
        1. Environment variable (MIURA_NEXUS_API_KEY) - highest priority
        2. Storage backend (file, keyring, or none) - based on --storage flag
        3. Keyring (if --storage=keyring or MIURA_AUTH_STORAGE=keyring)

        Args:
            storage_backend_type: Optional override for storage backend type

        Returns:
            Optional[str]: The current refresh token if available, otherwise None.

        Raises:
            ApiKeyInvalidError: If a token from any source is malformed.
            KeyringError: If keyring access fails unexpectedly.
        """
        # Lazy import to avoid slow startup times
        from miura.nexus.infrastructure.config import ConfigurationManager

        try:
            # Trigger environment loading (if needed)
            ConfigurationManager(load_dotenv=False)
        except Exception as exc:
            logger.debug(f"Could not load application config during token sync: {exc}")

        current_token: Optional[str] = None
        token_source: Optional[str] = None

        # 1. Environment variable has highest priority
        env_refresh_token = os.environ.get("MIURA_NEXUS_API_KEY")
        if env_refresh_token:
            self.jwt_utils.validate_jwt_format(env_refresh_token)
            current_token = env_refresh_token
            token_source = "env"
        else:
            # 2. Try storage backend (file, keyring, or none)
            try:
                stored_token = self.storage_backend.get("auth_api_key")
                if stored_token:
                    self.jwt_utils.validate_jwt_format(stored_token)
                    current_token = stored_token
                    token_source = "storage"
            except Exception as e:
                logger.debug(f"Storage backend access failed: {e}")

            # 3. If using keyring backend and storage didn't work, try keyring cache
            if not current_token and storage_backend_type == "keyring":
                try:
                    cache = self._get_keyring_cache()
                    cached_token = cache.get("auth_api_key")
                    if cached_token:
                        self.jwt_utils.validate_jwt_format(cached_token)
                        current_token = cached_token
                        token_source = "keyring_cache"
                except Exception as e:
                    logger.debug(f"Keyring cache access failed: {e}")

        if current_token is not None:
            refresh_changed = current_token != self.token_manager.refresh_token
            self.token_manager.set_refresh_token(
                current_token,
                reason="Refresh token changed, invalidating cached access token",
            )
            if refresh_changed:
                if token_source == "env":
                    logger.debug("Using refresh token from environment variable (MIURA_NEXUS_API_KEY)")
                elif token_source == "storage":
                    logger.debug(f"Using refresh token from {type(self.storage_backend).__name__} storage")
                elif token_source == "keyring_cache":
                    logger.debug("Using refresh token from keyring cache")
            return self.token_manager.refresh_token

        # No token found in any source; clear cached tokens to avoid stale values
        if self.token_manager.has_tokens():
            logger.debug("No refresh token available, clearing cached tokens")
            self.token_manager.set_refresh_token(
                None,
                reason="Refresh token cleared; removing cached access token",
                force_invalidate=True,
            )
        else:
            self.token_manager.set_refresh_token(None)
        return None

    def get_auth_api_key(self) -> str:
        """
        Get an access token for API authentication.
        Uses refresh token to fetch fresh access tokens transparently.
        Priority order: 1) Environment variable, 2) Memory cache, 3) Keyring.

        Returns:
            str: The access token for API authentication

        Raises:
            ApiKeyNotFoundError: If no refresh token is found in any source
            KeyringError: If there's an error accessing keyring
            ApiKeyInvalidError: If refresh token format is invalid
            PermissionError: If token refresh fails
        """
        # Ensure cached refresh token reflects latest source state
        # Determine storage backend type from handler
        from .storage import KeyringStorageBackend, FileStorageBackend, NoneStorageBackend

        storage_type = None
        if isinstance(self.storage_backend, KeyringStorageBackend):
            storage_type = "keyring"
        elif isinstance(self.storage_backend, FileStorageBackend):
            storage_type = "file"
        elif isinstance(self.storage_backend, NoneStorageBackend):
            storage_type = "none"
        refresh_token = self._sync_refresh_token_from_sources(storage_backend_type=storage_type)

        # Check if we have a valid access token in memory first
        if self.token_manager.access_token:
            # Check if the access token has expired
            try:
                if not self.jwt_utils.is_expired(self.token_manager.access_token):
                    logger.debug("Using cached access token")
                    return self.token_manager.access_token
                else:
                    logger.debug("Access token expired or expiring soon, refreshing...")
            except Exception as e:
                logger.debug(f"Could not check access token expiration: {e}, refreshing...")

            # If we reach here, the token is expired or we couldn't check it
            self.token_manager.invalidate_access_token("Access token expired or could not be verified; clearing cache")
        else:
            logger.debug("Access token cache empty; fetching fresh token")

        # Now use the refresh token to get an access token
        if refresh_token:
            access_token = self.keycloak_client.refresh_access_token(refresh_token)
            self.token_manager.set_access_token(access_token)
            return access_token

        # Determine which error to surface for missing refresh token
        if os.environ.get("MIURA_NEXUS_API_KEY"):
            raise ApiKeyInvalidError(
                "MIURA_NEXUS_API_KEY is set but could not be validated",
                details={"source": "environment_variable"},
            )

        # Note: Keyring availability is checked lazily when operations are attempted
        # If keyring operations fail, appropriate errors will be raised

        raise ApiKeyNotFoundError(
            "No refresh token found. Please set MIURA_NEXUS_API_KEY environment variable "
            "or use 'miura-auth token set' to store a refresh token.",
            details={"sources_checked": ["environment_variable", "keyring"]},
        )

    def set_auth_api_key(self, api_key: str) -> None:
        """
        Store a refresh token (presented as API key to user) in storage backend.

        When storing in a specific backend, clears tokens from other backends
        to maintain a single source of truth.

        Args:
            api_key: The refresh token to store (user sees this as API key)

        Raises:
            ApiKeyInvalidError: If refresh token format is invalid
            KeyringError: If there's an error storing in keyring (when using keyring backend)
        """
        # Validate api key format first
        self.jwt_utils.validate_jwt_format(api_key)

        try:
            # Store in selected storage backend
            self.storage_backend.set("auth_api_key", api_key)

            # Clear tokens from other storage backends to maintain single source of truth
            from .storage import FileStorageBackend, KeyringStorageBackend, NoneStorageBackend

            current_backend_type = None
            if isinstance(self.storage_backend, FileStorageBackend):
                current_backend_type = "file"
            elif isinstance(self.storage_backend, KeyringStorageBackend):
                current_backend_type = "keyring"
            elif isinstance(self.storage_backend, NoneStorageBackend):
                current_backend_type = "none"

            # Clear from other backends
            if current_backend_type != "file":
                try:
                    file_backend = create_storage_backend("file")
                    if file_backend.exists("auth_api_key"):
                        file_backend.delete("auth_api_key")
                        logger.debug("Cleared token from file storage")
                except Exception as e:
                    logger.debug(f"Could not clear file storage: {e}")

            if current_backend_type != "keyring":
                try:
                    # Only create KeyringManager when needed for cleanup
                    if self._keyring_manager is None:
                        from .keyring import KeyringManager

                        self._keyring_manager = KeyringManager()
                    keyring_backend = create_storage_backend("keyring", self._keyring_manager)
                    if keyring_backend.exists("auth_api_key"):
                        keyring_backend.delete("auth_api_key")
                        logger.debug("Cleared token from keyring storage")
                except Exception as e:
                    logger.debug(f"Could not clear keyring storage: {e}")

            # Also update keyring cache if using keyring backend
            if isinstance(self.storage_backend, KeyringStorageBackend):
                cache = self._get_keyring_cache()
                cache["auth_api_key"] = api_key

            # Update in-memory cache and invalidate any cached access token if necessary
            self.token_manager.set_refresh_token(
                api_key,
                reason="Stored new refresh token, invalidating cached access token",
            )
        except KeyringError as e:
            logger.error(f"Storage failed: {e}")
            raise
        except Exception as e:
            logger.error(f"Unexpected error storing api key: {e}")
            raise NetworkError(
                f"Failed to store authentication api key: {e}",
                details={"error_type": type(e).__name__},
            )

    def clear_api_keys(self) -> None:
        """
        Clear all stored tokens from memory and storage backend.

        Raises:
            KeyringError: If there's an error clearing from keyring (when using keyring backend)
        """
        # Clear from storage backend
        try:
            self.storage_backend.delete("auth_api_key")
        except Exception as e:
            logger.warning(f"Unexpected error clearing api key from storage: {e}")
            # Don't raise here as we still want to clear from memory

        # Clear keyring cache
        self._clear_keyring_cache()

        # Clear from memory
        self.token_manager.clear_all("Clearing cached tokens from memory")

    def get_api_key_info(self) -> Dict[str, Any]:
        """
        Get information about the current refresh token and access token.

        Returns:
            dict: Token information including source (env var, keyring, etc.)
        """
        # Determine storage backend type
        from .storage import KeyringStorageBackend

        storage_type = None
        if isinstance(self.storage_backend, KeyringStorageBackend):
            storage_type = "keyring"
        self._sync_refresh_token_from_sources(storage_backend_type=storage_type)

        info = {
            "has_refresh_token": self.token_manager.refresh_token is not None,
            "has_access_token": self.token_manager.access_token is not None,
            "has_refresh_token_in_env": os.environ.get("MIURA_NEXUS_API_KEY") is not None,
        }

        # Check if refresh token exists in storage backend
        try:
            info["has_refresh_token_in_storage"] = self.storage_backend.exists("auth_api_key")
        except Exception:
            info["has_refresh_token_in_storage"] = False

        # Also check keyring for backward compatibility info (only if using keyring backend)
        if isinstance(self.storage_backend, KeyringStorageBackend):
            try:
                # KeyringManager is guaranteed to exist when using keyring backend
                assert self._keyring_manager is not None
                info["has_refresh_token_in_keyring"] = self._keyring_manager.has_password("auth_api_key")
            except Exception:
                info["has_refresh_token_in_keyring"] = False
        else:
            # Not using keyring backend, so no need to check keyring
            info["has_refresh_token_in_keyring"] = False

        return info

    def get_api_key_details(self) -> Dict[str, Any]:
        """
        Get detailed information about the current refresh token including JWT payload.

        Returns:
            dict: Detailed token information including JWT claims

        Raises:
            ApiKeyNotFoundError: If no refresh token is found
            ApiKeyInvalidError: If refresh token format is invalid
        """
        from .storage import FileStorageBackend, KeyringStorageBackend, NoneStorageBackend

        storage_type = None
        if isinstance(self.storage_backend, KeyringStorageBackend):
            storage_type = "keyring"
        elif isinstance(self.storage_backend, FileStorageBackend):
            storage_type = "file"
        elif isinstance(self.storage_backend, NoneStorageBackend):
            storage_type = "none"
        refresh_token = self._sync_refresh_token_from_sources(storage_backend_type=storage_type)

        if not refresh_token:
            raise ApiKeyNotFoundError(
                "No refresh token currently loaded",
                details={"operation": "get_api_key_details"},
            )

        try:
            # Always parse the refresh token payload for token details (what user stored)
            refresh_payload = self.jwt_utils.parse_jwt_payload(refresh_token)

            # Status/details must report the stored refresh token. Avoid a network refresh here:
            # resolving Keycloak config can load local .env values and override isolated storage tests.
            access_payload = None
            access_token_fetch_error = None
            if self.token_manager.access_token:
                try:
                    access_payload = self.jwt_utils.parse_jwt_payload(self.token_manager.access_token)
                except Exception as e:
                    access_token_fetch_error = f"Failed to parse cached access token: {e}"
                    logger.debug(access_token_fetch_error)

            # Get basic info
            basic_info = self.get_api_key_info()

            # Calculate expiration status based on REFRESH TOKEN (what user stored)
            import time

            current_time = time.time()
            refresh_token_exp = refresh_payload.get("exp")
            is_expired = False

            if refresh_token_exp:
                is_expired = current_time >= refresh_token_exp

            # Use access token payload for user info if available, otherwise use refresh token
            user_info_payload = access_payload if access_payload else refresh_payload

            # Add JWT-specific information
            # jwt_payload should be the refresh token payload (what user stored)
            details = {
                **basic_info,
                "jwt_payload": refresh_payload,  # Always use refresh token payload
                "is_expired": is_expired,
                "has_offline_access": self.jwt_utils.has_offline_access(refresh_payload),
                **self.jwt_utils.extract_user_info(user_info_payload),  # Use access token for user info if available
            }
            # Add access token fetch error if present
            if access_token_fetch_error:
                details["access_token_fetch_error"] = access_token_fetch_error
            return details

        except ApiKeyInvalidError:
            # If JWT parsing fails, return basic info without JWT details
            basic_info = self.get_api_key_info()
            basic_info["jwt_parse_error"] = "API key is not a valid JWT or is malformed"
            basic_info["is_expired"] = False  # Can't determine expiration if JWT is invalid
            basic_info["has_offline_access"] = False  # Can't determine scope if JWT is invalid
            return basic_info

    def close(self) -> None:
        """Close resources and clean up."""
        if self.keycloak_client:
            self.keycloak_client.close()