Cryptography

Overview

The Cryptography module provides a simple, high-level abstraction for symmetric (secret key) encryption and decryption. It is built to be secure by default, leveraging the robust and widely trusted Fernet implementation from the cryptography library.

Its primary use case is to provide an end-to-end encryption layer for message payloads as part of the Payload Processing Pipeline, but it can also be used for any general-purpose encryption needs within your application.

Key Features

• Strong Encryption: Uses Fernet (AES-128 in CBC mode with PKCS7 padding, signed with HMAC using SHA256).
• Secure Key Handling: The encryption key (fernet_key) is configured as a secret and resolved at runtime via the Secrets Management module.
• Lazy Initialization: The Fernet client is only initialized on its first use, preventing unnecessary key handling at startup.
• Pipeline Integration: Seamlessly integrates into the messaging payload pipeline to automatically encrypt and decrypt messages.

---

How It Works

The module is centered around the CryptoProviderProtocol, which defines a simple encrypt_bytes and decrypt_bytes interface. The default implementation is the FernetCryptoProvider.

The CryptoProviderFactory is a singleton factory used to obtain a configured instance of the provider. It ensures that the same key and client are used throughout the application.

---

Integration with the Messaging Pipeline

The most common way to use the crypto module is to add it as a step in the messaging payload pipeline. This automatically encrypts the entire message payload after serialization and before it is sent to the broker. On the consumer side, it automatically decrypts the payload before deserialization.

To enable this, simply add a "crypto" step to your pipeline configuration.

# In settings.toml
[default.integration.messaging.payload]

# The pipeline is executed in order for encoding, and in reverse for decoding.
# Flow: Pydantic Model -> [Serializer] -> Bytes -> [Crypto] -> Encrypted Bytes
[[default.integration.messaging.payload.pipeline]]
step = "serializer"

[[default.integration.messaging.payload.pipeline]]
step = "crypto"

With this configuration, all messages will be transparently encrypted and decrypted as they are produced and consumed.

---

Direct Usage Example

You can also use the crypto provider directly for general-purpose encryption tasks.

from nala.athomic.security.crypto import CryptoProviderFactory

async def encrypt_sensitive_data(data: str) -> bytes:
    crypto_provider = CryptoProviderFactory.create()

    data_bytes = data.encode('utf-8')
    encrypted_bytes = await crypto_provider.encrypt_bytes(data_bytes)

    return encrypted_bytes

async def decrypt_sensitive_data(encrypted_data: bytes) -> str:
    crypto_provider = CryptoProviderFactory.create()

    decrypted_bytes = await crypto_provider.decrypt_bytes(encrypted_data)

    return decrypted_bytes.decode('utf-8')

---

Configuration

The cryptography module requires a fernet_key to be configured under the [security.crypto] section in your settings.toml. For production, this must be a reference to a secret stored in a secure backend like Vault.

[default.security.crypto]
# The Fernet key must be 32 URL-safe base64-encoded bytes.
# In production, this should always be a secret reference.
fernet_key = { path = "global/crypto", key = "fernet-key" }

You can generate a new Fernet key using the cryptography library:

from cryptography.fernet import Fernet
key = Fernet.generate_key()
print(key.decode())

---

API Reference

nala.athomic.security.crypto.protocol.CryptoProviderProtocol

Bases: Protocol

Defines the contract for a cryptographic provider that operates on raw bytes.

This protocol ensures that any concrete implementation (e.g., Fernet, AES) conforms to a standardized asynchronous interface for encryption and decryption, adhering to the Dependency Inversion Principle (DIP).

decrypt_bytes(encrypted_data) async

Decrypts an encrypted byte string.

Parameters:

Name	Type	Description	Default
encrypted_data	bytes	The encrypted byte string.	required

Returns:

Name	Type	Description
bytes	bytes	The original, decrypted byte string.

Raises:

Type	Description
CryptoOperationError	If decryption fails (e.g., invalid key, invalid token).

encrypt_bytes(data) async

Encrypts a raw byte string.

Parameters:

Name	Type	Description	Default
data	bytes	The raw, unencrypted byte string.	required

Returns:

Name	Type	Description
bytes	bytes	The encrypted byte string.

nala.athomic.security.crypto.factory.CryptoProviderFactory

Manages the singleton instance of the CryptoProvider.

nala.athomic.security.crypto.fernet_provider.FernetCryptoProvider

Bases: CryptoProviderProtocol, CredentialResolve

A cryptographic provider implementation using the Fernet symmetric encryption standard.

This class adheres to the CryptoProviderProtocol and manages the entire lifecycle and security of the Fernet key: 1. Lazy Initialization: The key is resolved and the Fernet instance is created only on the first call to encrypt_bytes or decrypt_bytes. 2. Asynchronous Credential Resolution: It uses the CredentialResolve mixin to securely fetch the encryption key from an external source (e.g., Vault). 3. Integrated Observability: All encryption and decryption attempts are instrumented with distributed tracing and Prometheus metrics.

Attributes:

Name	Type	Description
settings	CryptoSettings	The configuration for the crypto module.
key_source		The source object (e.g., SecretStr, CredentialProxy) for the Fernet key.
fernet	Optional[Fernet]	The underlying Fernet encryption instance.
_init_lock	Lock	A lock to ensure thread-safe, single-time initialization.
tracer		An OpenTelemetry tracer instance.
provider_name	str	The name used for metric and span labeling.

__init__(settings=None)

Initializes the provider, but delays Fernet instance creation.

Parameters:

Name	Type	Description	Default
settings	Optional[CryptoSettings]	The configuration settings for the provider.	None

decrypt_bytes(encrypted_data) async

Decrypts an encrypted byte string, wrapped with tracing and metrics.

Parameters:

Name	Type	Description	Default
encrypted_data	bytes	The encrypted byte string.	required

Returns:

Name	Type	Description
bytes	bytes	The original, decrypted byte string.

Raises:

Type	Description
CryptoOperationError	If decryption fails due to an invalid token or any other internal error.

encrypt_bytes(data) async

Encrypts a raw byte string, wrapped with tracing and metrics.

Parameters:

Name	Type	Description	Default
data	bytes	The raw, unencrypted byte string.	required

Returns:

Name	Type	Description
bytes	bytes	The encrypted byte string.

Raises:

Type	Description
CryptoOperationError	If encryption fails for any reason (e.g., internal error).

nala.athomic.security.crypto.adapters.CryptoStepAdapter

Bases: ProcessingStepProtocol

An adapter that integrates a CryptoProviderProtocol into the payload processing pipeline.

This class acts as the Adapter for the encryption step (part of the Pipes and Filters pattern), mapping the generic encode/decode operations required by the PayloadProcessor to the concrete encrypt_bytes/decrypt_bytes methods of the underlying CryptoProviderProtocol. This adheres to the SRP by ensuring the CryptoProvider focuses only on the cryptographic operation, while the adapter focuses only on the interface transformation.

Attributes:

Name	Type	Description
crypto	CryptoProviderProtocol	The underlying provider instance that performs the actual encryption and decryption.

__init__(crypto_provider)

Initializes the adapter with the core cryptographic provider.

Parameters:

Name	Type	Description	Default
crypto_provider	CryptoProviderProtocol	An instance of the concrete cryptographic provider (e.g., FernetCryptoProvider).	required

decode(data, **kwargs) async

Decrypts the data (inbound pipeline flow).

Maps the pipeline's decode call to the provider's decrypt_bytes method.

Parameters:

Name	Type	Description	Default
data	Any	The encrypted payload to be decrypted, expected as bytes.	required
**kwargs	Any	Additional keyword arguments (ignored by this step).	{}

Returns:

Name	Type	Description
Any	Any	The decrypted, original data as bytes.

Raises:

Type	Description
TypeError	If the input data is not of type bytes.
CryptoOperationError	Propagated from the underlying provider if decryption fails.

encode(data, **kwargs) async

Encrypts the data (outbound pipeline flow).

Maps the pipeline's encode call to the provider's encrypt_bytes method.

Parameters:

Name	Type	Description	Default
data	Any	The payload to be encrypted, expected as bytes.	required
**kwargs	Any	Additional keyword arguments (ignored by this step).	{}

Returns:

Name	Type	Description
Any	Any	The encrypted data as bytes.

Raises:

Type	Description
TypeError	If the input data is not of type bytes.