Authentication¶

Authentication primitives for JWT tokens and encryption.

Overview¶

The auth module provides core authentication functionality that can be used by any service in the Imbi ecosystem:

JWT Tokens: Access and refresh token creation/verification
Token Encryption: Fernet encryption for sensitive data at rest
Config Encryption: Fernet encryption for persisted configuration secrets, keyed separately from token encryption

JWT Tokens¶

from imbi.common.auth import core

# Create an access token
access_token = core.create_access_token(
    subject="user@example.com",
    extra_claims={"role": "admin"}
)

# Create a refresh token
refresh_token = core.create_refresh_token(
    subject="user@example.com"
)

# Verify and decode a token
try:
    payload = core.verify_token(access_token)
    user_id = payload["sub"]
    role = payload.get("role")
except Exception as e:
    print(f"Invalid token: {e}")

Token Encryption¶

from imbi.common.auth import encryption

# Encrypt sensitive data (e.g., OAuth tokens)
encrypted = encryption.encrypt_token("sensitive_oauth_token")

# Decrypt
decrypted = encryption.decrypt_token(encrypted)

Config Encryption¶

Persisted configuration secrets (e.g. external MCP server credentials) are encrypted with a dedicated key sourced from IMBI_CONFIG_ENCRYPTION_KEY, separate from the token-encryption key, so the two can be rotated independently. decrypt_config_value returns None for None or invalid/corrupt ciphertext.

from imbi.common.auth import encryption

# Encrypt a configuration secret
encrypted = encryption.encrypt_config_value("client-secret")

# Decrypt (None on invalid ciphertext)
decrypted = encryption.decrypt_config_value(encrypted)

API Reference¶

Core Functions¶

create_access_token ¶

create_access_token(
    subject: str,
    extra_claims: dict[str, Any] | None = None,
    auth_settings: Auth | None = None,
) -> str

Create JWT access token.

Parameters:

Name	Type	Description	Default
`subject`	`str`	Subject (user identifier) to encode in token	required
`extra_claims`	`dict[str, Any] \| None`	Optional additional claims to include	`None`
`auth_settings`	`Auth \| None`	Optional auth settings for JWT configuration (uses singleton if not provided)	`None`

Returns:

Type	Description
`str`	JWT token string

Source code in libraries/common/src/imbi/common/auth/core.py

def create_access_token(
    subject: str,
    extra_claims: dict[str, typing.Any] | None = None,
    auth_settings: settings.Auth | None = None,
) -> str:
    """Create JWT access token.

    Args:
        subject: Subject (user identifier) to encode in token
        extra_claims: Optional additional claims to include
        auth_settings: Optional auth settings for JWT configuration
            (uses singleton if not provided)

    Returns:
        JWT token string

    """
    if auth_settings is None:
        auth_settings = settings.get_auth_settings()
    return _create_token(
        subject,
        token_type='access',
        ttl_seconds=auth_settings.access_token_expire_seconds,
        extra_claims=extra_claims,
        auth_settings=auth_settings,
    )

create_refresh_token ¶

create_refresh_token(
    subject: str,
    extra_claims: dict[str, Any] | None = None,
    auth_settings: Auth | None = None,
) -> str

Create JWT refresh token.

Parameters:

Name	Type	Description	Default
`subject`	`str`	Subject (user identifier) to encode in token	required
`extra_claims`	`dict[str, Any] \| None`	Optional additional claims to include	`None`
`auth_settings`	`Auth \| None`	Optional auth settings for JWT configuration (uses singleton if not provided)	`None`

Returns:

Type	Description
`str`	JWT token string

Source code in libraries/common/src/imbi/common/auth/core.py

def create_refresh_token(
    subject: str,
    extra_claims: dict[str, typing.Any] | None = None,
    auth_settings: settings.Auth | None = None,
) -> str:
    """Create JWT refresh token.

    Args:
        subject: Subject (user identifier) to encode in token
        extra_claims: Optional additional claims to include
        auth_settings: Optional auth settings for JWT configuration
            (uses singleton if not provided)

    Returns:
        JWT token string

    """
    if auth_settings is None:
        auth_settings = settings.get_auth_settings()
    return _create_token(
        subject,
        token_type='refresh',
        ttl_seconds=auth_settings.refresh_token_expire_seconds,
        extra_claims=extra_claims,
        auth_settings=auth_settings,
    )

verify_token ¶

verify_token(
    token: str, auth_settings: Auth | None = None
) -> dict[str, typing.Any]

Decode and validate JWT token.

Parameters:

Name	Type	Description	Default
`token`	`str`	JWT token string to decode	required
`auth_settings`	`Auth \| None`	Optional auth settings for JWT configuration (uses singleton if not provided)	`None`

Returns:

Type	Description
`dict[str, Any]`	Decoded token claims

Raises:

Type	Description
`ExpiredSignatureError`	If token has expired
`InvalidTokenError`	If token is invalid

Source code in libraries/common/src/imbi/common/auth/core.py

def verify_token(
    token: str, auth_settings: settings.Auth | None = None
) -> dict[str, typing.Any]:
    """Decode and validate JWT token.

    Args:
        token: JWT token string to decode
        auth_settings: Optional auth settings for JWT configuration
            (uses singleton if not provided)

    Returns:
        Decoded token claims

    Raises:
        jwt.ExpiredSignatureError: If token has expired
        jwt.InvalidTokenError: If token is invalid

    """
    if auth_settings is None:
        auth_settings = settings.get_auth_settings()

    decoded: dict[str, typing.Any] = jwt.decode(
        token,
        auth_settings.jwt_secret,
        algorithms=[auth_settings.jwt_algorithm],
        options={'require': ['sub', 'jti', 'type', 'exp']},
    )
    return decoded

Encryption Functions¶

TokenEncryption ¶

TokenEncryption(encryption_key: str)

Handles encryption/decryption of sensitive tokens using Fernet.

This class follows the singleton pattern to ensure a single encryption instance is used throughout the application lifecycle.

Example

encryptor = TokenEncryption.get_instance() encrypted = encryptor.encrypt('my-secret-token') decrypted = encryptor.decrypt(encrypted) assert decrypted == 'my-secret-token'

Initialize with base64-encoded Fernet key.

Parameters:

Name	Type	Description	Default
`encryption_key`	`str`	Base64-encoded Fernet encryption key	required

Raises:

Type	Description
`ValueError`	If encryption key format is invalid

Source code in libraries/common/src/imbi/common/auth/encryption.py

def __init__(self, encryption_key: str) -> None:
    """Initialize with base64-encoded Fernet key.

    Args:
        encryption_key: Base64-encoded Fernet encryption key

    Raises:
        ValueError: If encryption key format is invalid
    """
    try:
        key_bytes = encryption_key.encode('ascii')
        self._fernet = fernet.Fernet(key_bytes)
    except Exception as err:
        LOGGER.error('Invalid encryption key format: %s', err)
        raise ValueError('Invalid encryption key') from err

decrypt ¶

decrypt(ciphertext: str | None) -> str | None

Decrypt a token string.

Parameters:

Name	Type	Description	Default
`ciphertext`	`str \| None`	Base64-encoded encrypted token, or None	required

Returns:

Type	Description
`str \| None`	Decrypted plaintext token, or None if input is None or decryption fails

Note

Returns None instead of raising exception on decryption failure to handle corrupted/invalid ciphertext gracefully. Handles both new format (Fernet base64) and legacy format (double base64 encoding) for backward compatibility.

Source code in libraries/common/src/imbi/common/auth/encryption.py

def decrypt(self, ciphertext: str | None) -> str | None:
    """Decrypt a token string.

    Args:
        ciphertext: Base64-encoded encrypted token, or None

    Returns:
        Decrypted plaintext token, or None if input is None
            or decryption fails

    Note:
        Returns None instead of raising exception on decryption failure
        to handle corrupted/invalid ciphertext gracefully.
        Handles both new format (Fernet base64) and legacy format
        (double base64 encoding) for backward compatibility.
    """
    if ciphertext is None:
        return None

    try:
        # Try new format first (Fernet base64 only)
        encrypted_bytes = ciphertext.encode('ascii')
        plaintext_bytes: bytes = self._fernet.decrypt(encrypted_bytes)
        return plaintext_bytes.decode('utf-8')
    except fernet.InvalidToken:
        # Try legacy format (double base64 encoding)
        try:
            encrypted_bytes = base64.urlsafe_b64decode(
                ciphertext.encode('ascii')
            )
            plaintext_bytes = self._fernet.decrypt(encrypted_bytes)
            return plaintext_bytes.decode('utf-8')
        except (
            fernet.InvalidToken,
            binascii.Error,
            ValueError,
            UnicodeDecodeError,
        ):
            # All decryption attempts failed - invalid or corrupted
            LOGGER.warning(
                'Failed to decrypt token - invalid or corrupted ciphertext'
            )
            return None
    except (binascii.Error, ValueError, UnicodeDecodeError):
        # Handle decryption errors (base64 decode, unicode errors)
        LOGGER.warning(
            'Failed to decrypt token - invalid or corrupted ciphertext'
        )
        return None

encrypt ¶

encrypt(plaintext: str | None) -> str | None

Encrypt a token string.

Parameters:

Name	Type	Description	Default
`plaintext`	`str \| None`	Token string to encrypt, or None	required

Returns:

Type	Description
`str \| None`	Base64-encoded encrypted token, or None if input is None

Raises:

Type	Description
`Exception`	If encryption fails

Source code in libraries/common/src/imbi/common/auth/encryption.py

def encrypt(self, plaintext: str | None) -> str | None:
    """Encrypt a token string.

    Args:
        plaintext: Token string to encrypt, or None

    Returns:
        Base64-encoded encrypted token, or None if input is None

    Raises:
        Exception: If encryption fails
    """
    if plaintext is None:
        return None

    try:
        encrypted_bytes: bytes = self._fernet.encrypt(
            plaintext.encode('utf-8')
        )
        # Fernet already returns base64-encoded bytes, just decode to str
        return encrypted_bytes.decode('ascii')
    except Exception as err:
        LOGGER.exception('Encryption failed: %s', err)
        raise

get_instance `classmethod` ¶

get_instance() -> TokenEncryption

Get singleton instance of TokenEncryption.

Returns:

Name	Type	Description
`TokenEncryption`	`TokenEncryption`	The singleton instance

Raises:

Type	Description
`RuntimeError`	If encryption key not configured in settings

Source code in libraries/common/src/imbi/common/auth/encryption.py

@classmethod
def get_instance(cls) -> TokenEncryption:
    """Get singleton instance of TokenEncryption.

    Returns:
        TokenEncryption: The singleton instance

    Raises:
        RuntimeError: If encryption key not configured in settings
    """
    if cls._instance is None:
        auth_settings = settings.get_auth_settings()
        if not auth_settings.encryption_key:
            raise RuntimeError('Encryption key not configured')
        cls._instance = cls(auth_settings.encryption_key)
    return cls._instance

reset_instance `classmethod` ¶

reset_instance() -> None

Reset singleton instance (for testing).

This method is primarily used in test suites to reset the singleton state between tests.

Source code in libraries/common/src/imbi/common/auth/encryption.py

@classmethod
def reset_instance(cls) -> None:
    """Reset singleton instance (for testing).

    This method is primarily used in test suites to reset the singleton
    state between tests.
    """
    cls._instance = None

get_fernet ¶

get_fernet(
    auth_settings: Auth | None = None,
) -> fernet.Fernet

Get Fernet instance for encryption/decryption.

Parameters:

Name	Type	Description	Default
`auth_settings`	`Auth \| None`	Optional auth settings (uses singleton if not provided)	`None`

Returns:

Type	Description
`Fernet`	Fernet instance configured with encryption key

Source code in libraries/common/src/imbi/common/auth/encryption.py

def get_fernet(auth_settings: settings.Auth | None = None) -> fernet.Fernet:
    """Get Fernet instance for encryption/decryption.

    Args:
        auth_settings: Optional auth settings (uses singleton if not provided)

    Returns:
        Fernet instance configured with encryption key

    """
    if auth_settings is None:
        auth_settings = settings.get_auth_settings()

    if not auth_settings.encryption_key:
        raise RuntimeError('Encryption key not configured')

    return fernet.Fernet(auth_settings.encryption_key.encode('ascii'))

encrypt_token ¶

encrypt_token(
    plaintext: str, auth_settings: Auth | None = None
) -> str

Encrypt a token string.

Parameters:

Name	Type	Description	Default
`plaintext`	`str`	Token string to encrypt	required
`auth_settings`	`Auth \| None`	Optional auth settings (creates default if not provided)	`None`

Returns:

Type	Description
`str`	Base64-encoded encrypted token

Source code in libraries/common/src/imbi/common/auth/encryption.py

def encrypt_token(
    plaintext: str, auth_settings: settings.Auth | None = None
) -> str:
    """Encrypt a token string.

    Args:
        plaintext: Token string to encrypt
        auth_settings: Optional auth settings (creates default if not provided)

    Returns:
        Base64-encoded encrypted token

    """
    f = get_fernet(auth_settings)
    encrypted_bytes: bytes = f.encrypt(plaintext.encode('utf-8'))
    return encrypted_bytes.decode('ascii')

decrypt_token ¶

decrypt_token(
    ciphertext: str, auth_settings: Auth | None = None
) -> str

Decrypt a token string.

Parameters:

Name	Type	Description	Default
`ciphertext`	`str`	Base64-encoded encrypted token	required
`auth_settings`	`Auth \| None`	Optional auth settings (creates default if not provided)	`None`

Returns:

Type	Description
`str`	Decrypted plaintext token

Raises:

Type	Description
`Exception`	If decryption fails

Source code in libraries/common/src/imbi/common/auth/encryption.py

def decrypt_token(
    ciphertext: str, auth_settings: settings.Auth | None = None
) -> str:
    """Decrypt a token string.

    Args:
        ciphertext: Base64-encoded encrypted token
        auth_settings: Optional auth settings (creates default if not provided)

    Returns:
        Decrypted plaintext token

    Raises:
        Exception: If decryption fails

    """
    f = get_fernet(auth_settings)
    plaintext_bytes: bytes = f.decrypt(ciphertext.encode('ascii'))
    return plaintext_bytes.decode('utf-8')

ConfigEncryption ¶

ConfigEncryption(encryption_key: str)

Encrypt/decrypt persisted configuration secrets using Fernet.

Mirrors :class:TokenEncryption but is keyed off the dedicated IMBI_CONFIG_ENCRYPTION_KEY (via :func:imbi.common.settings.get_config_settings) so configuration secrets and auth tokens can be rotated independently.

Initialize with base64-encoded Fernet key.

Parameters:

Name	Type	Description	Default
`encryption_key`	`str`	Base64-encoded Fernet encryption key	required

Raises:

Type	Description
`ValueError`	If encryption key format is invalid

Source code in libraries/common/src/imbi/common/auth/encryption.py

def __init__(self, encryption_key: str) -> None:
    """Initialize with base64-encoded Fernet key.

    Args:
        encryption_key: Base64-encoded Fernet encryption key

    Raises:
        ValueError: If encryption key format is invalid
    """
    try:
        key_bytes = encryption_key.encode('ascii')
        self._fernet = fernet.Fernet(key_bytes)
    except Exception as err:
        LOGGER.error('Invalid config encryption key format: %s', err)
        raise ValueError('Invalid encryption key') from err

decrypt ¶

decrypt(ciphertext: str | None) -> str | None

Decrypt a configuration value.

Parameters:

Name	Type	Description	Default
`ciphertext`	`str \| None`	Base64-encoded encrypted value, or None	required

Returns:

Type	Description
`str \| None`	Decrypted plaintext value, or None if input is None or decryption fails

Note

Returns None instead of raising on decryption failure to handle corrupted/invalid ciphertext gracefully.

Source code in libraries/common/src/imbi/common/auth/encryption.py

def decrypt(self, ciphertext: str | None) -> str | None:
    """Decrypt a configuration value.

    Args:
        ciphertext: Base64-encoded encrypted value, or None

    Returns:
        Decrypted plaintext value, or None if input is None or
            decryption fails

    Note:
        Returns None instead of raising on decryption failure to handle
        corrupted/invalid ciphertext gracefully.
    """
    if ciphertext is None:
        return None

    try:
        encrypted_bytes = ciphertext.encode('ascii')
        plaintext_bytes: bytes = self._fernet.decrypt(encrypted_bytes)
        return plaintext_bytes.decode('utf-8')
    except (
        fernet.InvalidToken,
        binascii.Error,
        ValueError,
        UnicodeDecodeError,
    ):
        LOGGER.warning(
            'Failed to decrypt config value - invalid or corrupt '
            'ciphertext'
        )
        return None

encrypt ¶

encrypt(plaintext: str | None) -> str | None

Encrypt a configuration value.

Parameters:

Name	Type	Description	Default
`plaintext`	`str \| None`	Value to encrypt, or None	required

Returns:

Type	Description
`str \| None`	Base64-encoded encrypted value, or None if input is None

Raises:

Type	Description
`Exception`	If encryption fails

Source code in libraries/common/src/imbi/common/auth/encryption.py

def encrypt(self, plaintext: str | None) -> str | None:
    """Encrypt a configuration value.

    Args:
        plaintext: Value to encrypt, or None

    Returns:
        Base64-encoded encrypted value, or None if input is None

    Raises:
        Exception: If encryption fails
    """
    if plaintext is None:
        return None

    try:
        encrypted_bytes: bytes = self._fernet.encrypt(
            plaintext.encode('utf-8')
        )
        return encrypted_bytes.decode('ascii')
    except Exception as err:
        LOGGER.exception('Config encryption failed: %s', err)
        raise

get_instance `classmethod` ¶

get_instance() -> ConfigEncryption

Get singleton instance of ConfigEncryption.

Returns:

Name	Type	Description
`ConfigEncryption`	`ConfigEncryption`	The singleton instance

Raises:

Type	Description
`RuntimeError`	If config encryption key not configured in settings

Source code in libraries/common/src/imbi/common/auth/encryption.py

@classmethod
def get_instance(cls) -> ConfigEncryption:
    """Get singleton instance of ConfigEncryption.

    Returns:
        ConfigEncryption: The singleton instance

    Raises:
        RuntimeError: If config encryption key not configured in settings
    """
    if cls._instance is None:
        config_settings = settings.get_config_settings()
        if not config_settings.encryption_key:
            raise RuntimeError('Encryption key not configured')
        cls._instance = cls(config_settings.encryption_key)
    return cls._instance

reset_instance `classmethod` ¶

reset_instance() -> None

Reset singleton instance (for testing).

Source code in libraries/common/src/imbi/common/auth/encryption.py

@classmethod
def reset_instance(cls) -> None:
    """Reset singleton instance (for testing)."""
    cls._instance = None

get_config_fernet ¶

get_config_fernet(
    config_settings: ConfigSecrets | None = None,
) -> fernet.Fernet

Get Fernet instance for configuration-secret encryption.

Parameters:

Name	Type	Description	Default
`config_settings`	`ConfigSecrets \| None`	Optional config settings (uses singleton if not provided)	`None`

Returns:

Type	Description
`Fernet`	Fernet instance configured with the config encryption key

Raises:

Type	Description
`RuntimeError`	If the config encryption key is not configured

Source code in libraries/common/src/imbi/common/auth/encryption.py

def get_config_fernet(
    config_settings: settings.ConfigSecrets | None = None,
) -> fernet.Fernet:
    """Get Fernet instance for configuration-secret encryption.

    Args:
        config_settings: Optional config settings (uses singleton if not
            provided)

    Returns:
        Fernet instance configured with the config encryption key

    Raises:
        RuntimeError: If the config encryption key is not configured

    """
    if config_settings is None:
        config_settings = settings.get_config_settings()

    if not config_settings.encryption_key:
        raise RuntimeError('Encryption key not configured')

    return fernet.Fernet(config_settings.encryption_key.encode('ascii'))

encrypt_config_value ¶

encrypt_config_value(plaintext: str | None) -> str | None

Encrypt a configuration secret value.

Parameters:

Name	Type	Description	Default
`plaintext`	`str \| None`	Value to encrypt, or None	required

Returns:

Type	Description
`str \| None`	Base64-encoded ciphertext, or None if input is None

Raises:

Type	Description
`RuntimeError`	If the config encryption key is not configured

Source code in libraries/common/src/imbi/common/auth/encryption.py

def encrypt_config_value(plaintext: str | None) -> str | None:
    """Encrypt a configuration secret value.

    Args:
        plaintext: Value to encrypt, or None

    Returns:
        Base64-encoded ciphertext, or None if input is None

    Raises:
        RuntimeError: If the config encryption key is not configured

    """
    if plaintext is None:
        return None
    f = get_config_fernet()
    encrypted_bytes: bytes = f.encrypt(plaintext.encode('utf-8'))
    return encrypted_bytes.decode('ascii')

decrypt_config_value ¶

decrypt_config_value(ciphertext: str | None) -> str | None

Decrypt a configuration secret value.

Parameters:

Name	Type	Description	Default
`ciphertext`	`str \| None`	Base64-encoded ciphertext, or None	required

Returns:

Type	Description
`str \| None`	Decrypted plaintext, or None if input is None or the ciphertext is
`str \| None`	invalid/corrupt.

Raises:

Type	Description
`RuntimeError`	If the config encryption key is not configured

Source code in libraries/common/src/imbi/common/auth/encryption.py

def decrypt_config_value(ciphertext: str | None) -> str | None:
    """Decrypt a configuration secret value.

    Args:
        ciphertext: Base64-encoded ciphertext, or None

    Returns:
        Decrypted plaintext, or None if input is None or the ciphertext is
        invalid/corrupt.

    Raises:
        RuntimeError: If the config encryption key is not configured

    """
    if ciphertext is None:
        return None
    f = get_config_fernet()
    try:
        plaintext_bytes: bytes = f.decrypt(ciphertext.encode('ascii'))
        return plaintext_bytes.decode('utf-8')
    except (
        fernet.InvalidToken,
        binascii.Error,
        ValueError,
        UnicodeDecodeError,
    ):
        LOGGER.warning(
            'Failed to decrypt config value - invalid or corrupt ciphertext'
        )
        return None

Authentication¶

Overview¶

JWT Tokens¶

Token Encryption¶

Config Encryption¶

API Reference¶

Core Functions¶

create_access_token ¶

create_refresh_token ¶

verify_token ¶

Encryption Functions¶

TokenEncryption ¶

decrypt ¶

encrypt ¶

get_instance classmethod ¶

reset_instance classmethod ¶

get_fernet ¶

encrypt_token ¶

decrypt_token ¶

ConfigEncryption ¶

decrypt ¶

encrypt ¶

get_instance classmethod ¶

reset_instance classmethod ¶

get_config_fernet ¶

encrypt_config_value ¶

decrypt_config_value ¶

get_instance `classmethod` ¶

reset_instance `classmethod` ¶

get_instance `classmethod` ¶

reset_instance `classmethod` ¶