domain.protocols.encryption_protocol

src.domain.protocols.encryption_protocol

Encryption protocol for provider credentials.

Defines the port for encryption/decryption operations. Infrastructure layer implements this protocol to provide AES-256-GCM encryption.

Architecture

• Domain layer protocol (port)
• Infrastructure adapter: src/infrastructure/providers/encryption_service.py
• Used by sync handlers to decrypt provider credentials

Reference

• docs/architecture/hexagonal.md
• docs/architecture/provider-integration-architecture.md

Classes

EncryptionError dataclass

Bases: DomainError

Base encryption error.

Used when encryption or decryption fails. Does NOT inherit from Exception - used in Result types.

Source code in src/domain/protocols/encryption_protocol.py

@dataclass(frozen=True, slots=True, kw_only=True)
class EncryptionError(DomainError):
    """Base encryption error.

    Used when encryption or decryption fails.
    Does NOT inherit from Exception - used in Result types.
    """

    pass

EncryptionKeyError dataclass

Bases: EncryptionError

Invalid encryption key.

Occurs when key doesn't meet requirements (wrong length, etc.).

Source code in src/domain/protocols/encryption_protocol.py

@dataclass(frozen=True, slots=True, kw_only=True)
class EncryptionKeyError(EncryptionError):
    """Invalid encryption key.

    Occurs when key doesn't meet requirements (wrong length, etc.).
    """

    pass

DecryptionError dataclass

Bases: EncryptionError

Decryption failure.

Occurs when: - Wrong encryption key - Data has been tampered with - Invalid encrypted data format

Source code in src/domain/protocols/encryption_protocol.py

@dataclass(frozen=True, slots=True, kw_only=True)
class DecryptionError(EncryptionError):
    """Decryption failure.

    Occurs when:
    - Wrong encryption key
    - Data has been tampered with
    - Invalid encrypted data format
    """

    pass

SerializationError dataclass

Bases: EncryptionError

Serialization/deserialization failure.

Occurs when data cannot be serialized to JSON or decrypted data cannot be parsed as JSON.

Source code in src/domain/protocols/encryption_protocol.py

@dataclass(frozen=True, slots=True, kw_only=True)
class SerializationError(EncryptionError):
    """Serialization/deserialization failure.

    Occurs when data cannot be serialized to JSON
    or decrypted data cannot be parsed as JSON.
    """

    pass

EncryptionProtocol

Bases: Protocol

Protocol for encryption/decryption operations.

Abstracts the encryption service used by application layer handlers. Infrastructure layer provides concrete implementation.

Example

class SyncAccountsHandler: def init( self, encryption_service: EncryptionProtocol, ... ) -> None: self._encryption = encryption_service

async def handle(self, cmd: SyncAccounts) -> Result[...]:
    result = self._encryption.decrypt(encrypted_data)
    ...

Source code in src/domain/protocols/encryption_protocol.py

class EncryptionProtocol(Protocol):
    """Protocol for encryption/decryption operations.

    Abstracts the encryption service used by application layer handlers.
    Infrastructure layer provides concrete implementation.

    Example:
        class SyncAccountsHandler:
            def __init__(
                self,
                encryption_service: EncryptionProtocol,
                ...
            ) -> None:
                self._encryption = encryption_service

            async def handle(self, cmd: SyncAccounts) -> Result[...]:
                result = self._encryption.decrypt(encrypted_data)
                ...
    """

    def encrypt(self, data: dict[str, Any]) -> Result[bytes, EncryptionError]:
        """Encrypt credentials dictionary to bytes.

        Args:
            data: Credentials dictionary to encrypt. Must be JSON-serializable.

        Returns:
            Success(bytes) with encrypted data.
            Failure(EncryptionError) if encryption fails.
        """
        ...

    def decrypt(self, encrypted: bytes) -> Result[dict[str, Any], EncryptionError]:
        """Decrypt bytes back to credentials dictionary.

        Args:
            encrypted: Encrypted bytes from encrypt().

        Returns:
            Success(dict) with original credentials dictionary.
            Failure(DecryptionError) if decryption fails.
        """
        ...

Functions

encrypt

encrypt(
    data: dict[str, Any],
) -> Result[bytes, EncryptionError]

Encrypt credentials dictionary to bytes.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Credentials dictionary to encrypt. Must be JSON-serializable.	required

Returns:

Type	Description
Result[bytes, EncryptionError]	Success(bytes) with encrypted data.
Result[bytes, EncryptionError]	Failure(EncryptionError) if encryption fails.

Source code in src/domain/protocols/encryption_protocol.py

def encrypt(self, data: dict[str, Any]) -> Result[bytes, EncryptionError]:
    """Encrypt credentials dictionary to bytes.

    Args:
        data: Credentials dictionary to encrypt. Must be JSON-serializable.

    Returns:
        Success(bytes) with encrypted data.
        Failure(EncryptionError) if encryption fails.
    """
    ...

decrypt

decrypt(
    encrypted: bytes,
) -> Result[dict[str, Any], EncryptionError]

Decrypt bytes back to credentials dictionary.

Parameters:

Name	Type	Description	Default
encrypted	bytes	Encrypted bytes from encrypt().	required

Returns:

Type	Description
Result[dict[str, Any], EncryptionError]	Success(dict) with original credentials dictionary.
Result[dict[str, Any], EncryptionError]	Failure(DecryptionError) if decryption fails.

Source code in src/domain/protocols/encryption_protocol.py

def decrypt(self, encrypted: bytes) -> Result[dict[str, Any], EncryptionError]:
    """Decrypt bytes back to credentials dictionary.

    Args:
        encrypted: Encrypted bytes from encrypt().

    Returns:
        Success(dict) with original credentials dictionary.
        Failure(DecryptionError) if decryption fails.
    """
    ...