Full Module Index

class AesGcmBackend:
    """AES-256-GCM encryption backend conforming to ``EncryptionBackend``.

    Accepts a key as exactly 32 bytes (256 bits). Use
    ``AESGCM.generate_key(bit_length=256)`` to generate a valid key.

    Attributes:
        _aesgcm (AESGCM): Internal AESGCM instance for encrypt/decrypt
            operations.

    Examples:
        Initialize with a generated key:

        ```python
        from cryptography.hazmat.primitives.ciphers.aead import AESGCM

        key = AESGCM.generate_key(bit_length=256)
        backend = AesGcmBackend(key=key)
        ```
    """

    def __init__(self, key: bytes) -> None:
        """Initialize AesGcmBackend with the given key.

        Args:
            key: Encryption key as exactly 32 bytes (256 bits).

        Raises:
            ConfigurationError: If *key* is not ``bytes`` or is not
                exactly 32 bytes long.
        """
        if not isinstance(key, bytes):
            msg = f"key must be bytes, got {type(key).__name__}"
            raise ConfigurationError(msg)

        if len(key) != _KEY_LENGTH:
            msg = f"key must be exactly {_KEY_LENGTH} bytes, got {len(key)}"
            raise ConfigurationError(msg)

        self._aesgcm = AESGCM(key)

    @property
    def backend_id(self) -> int:
        """Unique backend identifier for the envelope header.

        Returns:
            ``BACKEND_AES_GCM`` (``0x02``).
        """
        return BACKEND_AES_GCM

    def sync_encrypt(self, plaintext: bytes) -> bytes:
        """Encrypt plaintext bytes synchronously.

        Generates a fresh 12-byte random nonce per call. Returns
        ``nonce || ciphertext + tag``.

        Args:
            plaintext: Raw bytes to encrypt.

        Returns:
            Encrypted bytes as ``nonce (12) || ciphertext + tag``.

        Raises:
            TypeError: If *plaintext* is not ``bytes``.

        Examples:
            ```python
            ciphertext = backend.sync_encrypt(b"hello")
            ```
        """
        if not isinstance(plaintext, bytes):
            msg = f"plaintext must be bytes, got {type(plaintext).__name__}"
            raise TypeError(msg)

        nonce = os.urandom(_NONCE_LENGTH)
        ciphertext_and_tag = self._aesgcm.encrypt(nonce, plaintext, None)
        return nonce + ciphertext_and_tag

    def sync_decrypt(self, ciphertext: bytes) -> bytes:
        """Decrypt ciphertext bytes synchronously.

        Expects ``nonce (12 bytes) || ciphertext + tag``.

        Args:
            ciphertext: Encrypted bytes to decrypt.

        Returns:
            Decrypted plaintext as bytes.

        Raises:
            TypeError: If *ciphertext* is not ``bytes``.
            DecryptionError: If decryption fails due to wrong key,
                tampered ciphertext, or malformed input.

        Examples:
            ```python
            plaintext = backend.sync_decrypt(ciphertext)
            ```
        """
        if not isinstance(ciphertext, bytes):
            msg = f"ciphertext must be bytes, got {type(ciphertext).__name__}"
            raise TypeError(msg)

        if len(ciphertext) < _MIN_CIPHERTEXT_LENGTH:
            msg = "Decryption failed: ciphertext too short"
            raise DecryptionError(msg)

        nonce = ciphertext[:_NONCE_LENGTH]
        ct_and_tag = ciphertext[_NONCE_LENGTH:]

        try:
            return self._aesgcm.decrypt(nonce, ct_and_tag, None)
        except InvalidTag:
            msg = "Decryption failed: invalid tag or wrong key"
            raise DecryptionError(msg) from None

    async def encrypt(self, plaintext: bytes) -> bytes:
        """Encrypt plaintext bytes asynchronously.

        Args:
            plaintext: Raw bytes to encrypt.

        Returns:
            Encrypted bytes as ``nonce (12) || ciphertext + tag``.

        Raises:
            TypeError: If *plaintext* is not ``bytes``.

        Examples:
            ```python
            ciphertext = await backend.encrypt(b"hello")
            ```
        """
        if not isinstance(plaintext, bytes):
            msg = f"plaintext must be bytes, got {type(plaintext).__name__}"
            raise TypeError(msg)

        return await asyncio.to_thread(self.sync_encrypt, plaintext)

    async def decrypt(self, ciphertext: bytes) -> bytes:
        """Decrypt ciphertext bytes asynchronously.

        Args:
            ciphertext: Encrypted bytes to decrypt.

        Returns:
            Decrypted plaintext as bytes.

        Raises:
            TypeError: If *ciphertext* is not ``bytes``.
            DecryptionError: If decryption fails due to wrong key,
                tampered ciphertext, or malformed input.

        Examples:
            ```python
            plaintext = await backend.decrypt(ciphertext)
            ```
        """
        if not isinstance(ciphertext, bytes):
            msg = f"ciphertext must be bytes, got {type(ciphertext).__name__}"
            raise TypeError(msg)

        return await asyncio.to_thread(self.sync_decrypt, ciphertext)

class FernetBackend:
    """Fernet encryption backend conforming to ``EncryptionBackend``.

    Accepts a key as ``str`` or ``bytes``. If the key is a valid
    base64url-encoded 32-byte Fernet key, it is used directly (no
    derivation, no salt marker). Otherwise, a two-phase key derivation
    scheme is used:

    - **Init**: PBKDF2-HMAC-SHA256 (600,000 iterations) derives a master
      key from the passphrase.
    - **Per-operation**: HKDF-SHA256 with a fresh random salt expands
      the master key into a unique per-operation Fernet key.

    Backward compatibility: data encrypted with pre-3.2 fixed-salt
    derivation (480,000 iterations) is detected via marker byte and
    decrypted transparently.

    Attributes:
        _passphrase_key (bytes | None): Master key from PBKDF2 at 600k
            iterations (passphrase mode only, ``None`` in direct-key mode).
        _legacy_fernet (Fernet): Fernet instance for direct-key mode or
            legacy (480k iterations) backward-compatible decryption.
        _is_passphrase_mode (bool): Whether the backend was initialized
            with a passphrase (True) or a pre-generated Fernet key (False).

    Examples:
        Initialize with a passphrase:

        ```python
        backend = FernetBackend(key="my-secret")
        ```

        Initialize with a pre-generated Fernet key:

        ```python
        from cryptography.fernet import Fernet

        key = Fernet.generate_key()
        backend = FernetBackend(key=key)
        ```
    """

    def __init__(self, key: str | bytes) -> None:
        """Initialize FernetBackend with the given key.

        Args:
            key: Encryption key as ``str`` or ``bytes``. A valid Fernet
                key is used directly; arbitrary input is derived via
                two-phase PBKDF2 + HKDF.

        Raises:
            ConfigurationError: If *key* is not ``str`` or ``bytes``,
                or if *key* is empty.
        """
        if not isinstance(key, (str, bytes)):
            msg = f"key must be str or bytes, got {type(key).__name__}"
            raise ConfigurationError(msg)

        if isinstance(key, str):
            key = key.encode()

        if not key:
            msg = "key must not be empty"
            raise ConfigurationError(msg)

        if self._is_valid_fernet_key(key):
            self._passphrase_key: bytes | None = None
            self._legacy_fernet = Fernet(key)
            self._is_passphrase_mode = False
        else:
            self._passphrase_key = self._derive_master_key(key, _PBKDF2_ITERATIONS)
            legacy_raw = self._derive_master_key(key, _PBKDF2_ITERATIONS_LEGACY)
            self._legacy_fernet = Fernet(base64.urlsafe_b64encode(legacy_raw))
            self._is_passphrase_mode = True

    @property
    def backend_id(self) -> int:
        """Unique backend identifier for the envelope header.

        Returns:
            ``BACKEND_FERNET`` (``0x01``).
        """
        return BACKEND_FERNET

    def sync_encrypt(self, plaintext: bytes) -> bytes:
        """Encrypt plaintext bytes synchronously.

        In passphrase mode, generates a fresh 16-byte random salt,
        derives a per-operation Fernet key via HKDF, and returns
        ``SALT_MARKER || salt || fernet_token``. In direct-key mode,
        returns a standard Fernet token.

        Args:
            plaintext: Raw bytes to encrypt.

        Returns:
            Encrypted ciphertext as bytes.

        Raises:
            TypeError: If *plaintext* is not ``bytes``.

        Examples:
            ```python
            ciphertext = backend.sync_encrypt(b"hello")
            ```
        """
        if not isinstance(plaintext, bytes):
            msg = f"plaintext must be bytes, got {type(plaintext).__name__}"
            raise TypeError(msg)

        if not self._is_passphrase_mode:
            return self._legacy_fernet.encrypt(plaintext)

        salt = os.urandom(_SALT_LENGTH)
        per_op_key = self._derive_per_op_key(salt)
        token = Fernet(per_op_key).encrypt(plaintext)
        return _SALT_MARKER + salt + token

    def sync_decrypt(self, ciphertext: bytes) -> bytes:
        """Decrypt ciphertext bytes synchronously.

        Detects the ciphertext format via the first byte:

        - ``0x01``: new salted format — extracts salt, derives per-op
          key via HKDF, decrypts the Fernet token.
        - ``>= 0x2B``: legacy Fernet token — decrypts with the legacy
          Fernet instance.
        - Otherwise: raises ``DecryptionError``.

        Args:
            ciphertext: Encrypted bytes to decrypt.

        Returns:
            Decrypted plaintext as bytes.

        Raises:
            TypeError: If *ciphertext* is not ``bytes``.
            DecryptionError: If decryption fails due to wrong key,
                tampered ciphertext, or malformed input.

        Examples:
            ```python
            plaintext = backend.sync_decrypt(ciphertext)
            ```
        """
        if not isinstance(ciphertext, bytes):
            msg = f"ciphertext must be bytes, got {type(ciphertext).__name__}"
            raise TypeError(msg)

        if not ciphertext:
            msg = _DECRYPT_FAILED_MSG
            raise DecryptionError(msg)

        if ciphertext[0:1] == _SALT_MARKER and len(ciphertext) >= _MIN_SALTED_LENGTH:
            return self._decrypt_salted(ciphertext)

        if ciphertext[0] >= 0x2B:
            return self._decrypt_legacy(ciphertext)

        msg = "Decryption failed: invalid ciphertext format"
        raise DecryptionError(msg)

    async def encrypt(self, plaintext: bytes) -> bytes:
        """Encrypt plaintext bytes asynchronously.

        Delegates to ``sync_encrypt`` via ``asyncio.to_thread()``.

        Args:
            plaintext: Raw bytes to encrypt.

        Returns:
            Encrypted ciphertext as bytes.

        Raises:
            TypeError: If *plaintext* is not ``bytes``.

        Examples:
            ```python
            ciphertext = await backend.encrypt(b"hello")
            ```
        """
        if not isinstance(plaintext, bytes):
            msg = f"plaintext must be bytes, got {type(plaintext).__name__}"
            raise TypeError(msg)

        return await asyncio.to_thread(self.sync_encrypt, plaintext)

    async def decrypt(self, ciphertext: bytes) -> bytes:
        """Decrypt ciphertext bytes asynchronously.

        Delegates to ``sync_decrypt`` via ``asyncio.to_thread()``.

        Args:
            ciphertext: Encrypted bytes to decrypt.

        Returns:
            Decrypted plaintext as bytes.

        Raises:
            TypeError: If *ciphertext* is not ``bytes``.
            DecryptionError: If decryption fails due to wrong key,
                tampered ciphertext, or malformed input.

        Examples:
            ```python
            plaintext = await backend.decrypt(ciphertext)
            ```
        """
        if not isinstance(ciphertext, bytes):
            msg = f"ciphertext must be bytes, got {type(ciphertext).__name__}"
            raise TypeError(msg)

        return await asyncio.to_thread(self.sync_decrypt, ciphertext)

    def _derive_per_op_key(self, salt: bytes) -> bytes:
        """Derive a per-operation Fernet key via HKDF-SHA256.

        Validates that a master key exists, then expands it with the
        given salt to produce a unique Fernet key.

        Args:
            salt: Random salt for HKDF domain separation.

        Returns:
            A valid base64url-encoded 32-byte Fernet key.

        Raises:
            DecryptionError: If no master key is available (direct-key mode).
        """
        master = self._passphrase_key
        if master is None:
            msg = _DECRYPT_FAILED_MSG
            raise DecryptionError(msg)

        raw = HKDF(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt,
            info=_HKDF_INFO,
        ).derive(master)
        return base64.urlsafe_b64encode(raw)

    def _decrypt_salted(self, ciphertext: bytes) -> bytes:
        """Decrypt new-format salted ciphertext.

        Extracts the 16-byte salt, derives a per-op key via HKDF,
        and decrypts the remaining Fernet token.

        Args:
            ciphertext: Full ciphertext including marker and salt.

        Returns:
            Decrypted plaintext.

        Raises:
            DecryptionError: If decryption fails.
        """
        salt = ciphertext[1 : 1 + _SALT_LENGTH]
        token = ciphertext[1 + _SALT_LENGTH :]

        if self._passphrase_key is None:
            msg = _DECRYPT_FAILED_MSG
            raise DecryptionError(msg)

        per_op_key = self._derive_per_op_key(salt)
        try:
            return Fernet(per_op_key).decrypt(token)
        except InvalidToken:
            msg = _DECRYPT_FAILED_MSG
            raise DecryptionError(msg) from None

    def _decrypt_legacy(self, ciphertext: bytes) -> bytes:
        """Decrypt legacy Fernet token (pre-3.2 or direct-key mode).

        Delegates to ``self._legacy_fernet`` which uses the 480k-iteration
        PBKDF2-derived key (passphrase mode) or direct Fernet key.

        Args:
            ciphertext: Legacy Fernet token bytes.

        Returns:
            Decrypted plaintext.

        Raises:
            DecryptionError: If decryption fails.
        """
        try:
            return self._legacy_fernet.decrypt(ciphertext)
        except InvalidToken:
            msg = _DECRYPT_FAILED_MSG
            raise DecryptionError(msg) from None

    @staticmethod
    def _is_valid_fernet_key(key_bytes: bytes) -> bool:
        """Check whether key_bytes is a valid Fernet key.

        Attempts to construct a ``Fernet`` instance. Returns ``False``
        on ``ValueError`` (wrong key length) or ``binascii.Error``
        (non-base64 input).

        Args:
            key_bytes: Raw key material.

        Returns:
            True if the bytes are a valid Fernet key.
        """
        try:
            Fernet(key_bytes)
            return True
        except (ValueError, binascii.Error):
            return False

    @staticmethod
    def _derive_master_key(passphrase: bytes, iterations: int) -> bytes:
        """Derive a 32-byte master key via PBKDF2-HMAC-SHA256.

        Args:
            passphrase: Raw passphrase bytes.
            iterations: Number of PBKDF2 iterations.

        Returns:
            32-byte raw derived key.
        """
        return hashlib.pbkdf2_hmac(
            "sha256",
            passphrase,
            _PBKDF2_SALT,
            iterations,
        )

class ConfigurationError(SecureSessionError):
    """Raised when the service is misconfigured at startup.

    Covers invalid encryption keys, backends that do not conform to the
    ``EncryptionBackend`` protocol, invalid backend IDs, empty database
    paths, and database connection failures. Error messages never include
    key material or other sensitive data.

    Examples:
        Handle configuration failures at startup:

        ```python
        try:
            service = EncryptedSessionService(
                db_url="sqlite+aiosqlite:///sessions.db",
                backend=my_backend,
            )
        except ConfigurationError as exc:
            log.error("Service misconfigured: %s", exc)
        ```
    """

class DecryptionError(SecureSessionError, DontWrapMixin):
    """Raised when decryption fails.

    Inherits ``DontWrapMixin`` so SQLAlchemy propagates this directly
    instead of wrapping it in ``StatementError``.

    Possible causes include a wrong key, tampered ciphertext, or
    malformed input. Error messages intentionally exclude key material,
    ciphertext, and plaintext to prevent information leakage.

    Examples:
        Handle decryption failures specifically:

        ```python
        try:
            plaintext = await backend.decrypt(ciphertext)
        except DecryptionError:
            log.error("Decryption failed, check key")
        ```
    """

class EncryptionError(SecureSessionError, DontWrapMixin):
    """Raised when encryption fails.

    Inherits ``DontWrapMixin`` so SQLAlchemy propagates this directly
    instead of wrapping it in ``StatementError``.

    Possible causes include invalid plaintext input or backend-specific
    errors. Error messages intentionally exclude key material, plaintext,
    and ciphertext to prevent information leakage.

    Examples:
        Handle encryption failures specifically:

        ```python
        try:
            ciphertext = await backend.encrypt(plaintext)
        except EncryptionError:
            log.error("Encryption failed")
        ```
    """

class SecureSessionError(Exception):
    """Base exception for all adk-secure-sessions errors.

    All library-specific exceptions inherit from this class so callers
    can use a single ``except SecureSessionError`` clause to handle any
    failure originating from this package.

    Examples:
        Catch any library error regardless of type:

        ```python
        try:
            await backend.encrypt(plaintext)
        except SecureSessionError:
            log.error("adk-secure-sessions operation failed")
        ```
    """

class SerializationError(SecureSessionError, DontWrapMixin):
    """Raised when data cannot be serialized to JSON.

    Inherits ``DontWrapMixin`` so SQLAlchemy propagates this directly
    instead of wrapping it in ``StatementError``.

    This indicates a caller bug — the input contains types that are not
    JSON-serializable (e.g., ``datetime``, custom objects). This is
    distinct from encryption/decryption failures which indicate
    configuration or data integrity issues.

    Examples:
        Handle serialization failures:

        ```python
        try:
            envelope = await encrypt_session(data, backend, backend_id)
        except SerializationError:
            log.error("Data contains non-JSON-serializable values")
        ```
    """