Pipeline DSL¶

Pipeline DSL and module registry.

This module provides a small domain specific language for composing cryptographic workflows and a registry that allows backends to expose custom pipeline steps.

class cryptography_suite.pipeline.AESGCMDecrypt(password: str, kdf: str = 'argon2', backend: str | None = None)¶

Bases: CryptoModule[str, str]

Decrypt AES-GCM data.

Parameters:

password (str) – Password used to derive the decryption key.
kdf (str) – Key-derivation function name ("argon2" by default when available).
backend (str | None) – Optional backend name to use for this operation only.

Example

>>> from cryptography_suite.pipeline import AESGCMDecrypt
>>> AESGCMDecrypt(password="pw").run("...")
'hi'

See also

cryptography_suite.crypto_backends.use_backend: Selects the active backend providing the AES implementation.

backend: str | None = None¶

kdf: str = 'argon2'¶

password: str¶

run(data: str) → str¶: Run the module on the provided data.

to_proverif() → str¶: Return a ProVerif representation of the module.

to_tamarin() → str¶: Return a Tamarin representation of the module.

class cryptography_suite.pipeline.CryptoModule(*args, **kwargs)¶

Bases: Protocol[Input, Output]

Protocol for pipeline modules.

run(data: Input) → Output¶: Run the module on the provided data.

to_proverif() → str¶: Return a ProVerif representation of the module.

to_tamarin() → str¶: Return a Tamarin representation of the module.

class cryptography_suite.pipeline.ECIESX25519Decrypt(private_key: Any)¶

Bases: CryptoModule[str | bytes, bytes]

Decrypt ECIES X25519 ciphertext.

private_key: Any¶

run(data: str | bytes) → bytes¶: Run the module on the provided data.

to_proverif() → str¶: Return a ProVerif representation of the module.

to_tamarin() → str¶: Return a Tamarin representation of the module.

class cryptography_suite.pipeline.ECIESX25519Encrypt(public_key: Any, raw_output: bool = False)¶

Bases: CryptoModule[bytes, str | bytes]

Encrypt data using ECIES with X25519.

public_key: Any¶

raw_output: bool = False¶

run(data: bytes) → str | bytes¶: Run the module on the provided data.

to_proverif() → str¶: Return a ProVerif representation of the module.

to_tamarin() → str¶: Return a Tamarin representation of the module.

class cryptography_suite.pipeline.HybridDecrypt(private_key: Any)¶

Bases: CryptoModule[Any, bytes]

Decrypt data produced by HybridEncrypt.

private_key: Any¶

run(data: Any) → bytes¶: Run the module on the provided data.

to_proverif() → str¶: Return a ProVerif representation of the module.

to_tamarin() → str¶: Return a Tamarin representation of the module.

class cryptography_suite.pipeline.HybridEncrypt(public_key: Any, raw_output: bool = False)¶

Bases: CryptoModule[bytes, Any]

Encrypt data using hybrid RSA/ECIES + AES-GCM.

public_key: Any¶

raw_output: bool = False¶

run(data: bytes) → Any¶: Run the module on the provided data.

to_proverif() → str¶: Return a ProVerif representation of the module.

to_tamarin() → str¶: Return a Tamarin representation of the module.

class cryptography_suite.pipeline.KyberDecrypt(private_key: Any, level: int = 512)¶

Bases: CryptoModule[tuple[str | bytes, str | bytes], bytes]

Decrypt data produced by KyberEncrypt.

level: int = 512¶

private_key: Any¶

run(data: tuple[str | bytes, str | bytes]) → bytes¶: Run the module on the provided data.

to_proverif() → str¶: Return a ProVerif representation of the module.

to_tamarin() → str¶: Return a Tamarin representation of the module.

class cryptography_suite.pipeline.KyberEncrypt(public_key: bytes, level: int = 512, raw_output: bool = False)¶

Bases: CryptoModule[bytes, tuple[str | bytes, str | bytes]]

Encrypt data using Kyber and AES-GCM.

level: int = 512¶

public_key: bytes¶

raw_output: bool = False¶

run(data: bytes) → tuple[str | bytes, str | bytes]¶: Run the module on the provided data.

to_proverif() → str¶: Return a ProVerif representation of the module.

to_tamarin() → str¶: Return a Tamarin representation of the module.

class cryptography_suite.pipeline.Pipeline(modules: list[~cryptography_suite.pipeline.CryptoModule[~typing.Any, ~typing.Any]] = <factory>, tracked_secrets: list[str] = <factory>)¶

Bases: Generic[Input, Output]

Composable cryptographic pipeline.

describe() → list[dict[str, Any]]¶

dry_run(data: Any) → Any¶

classmethod from_config(config: Iterable[Callable[[], CryptoModule]]) → Pipeline¶

modules: list[CryptoModule[Any, Any]]¶

run(data: Any) → Any¶

run_with_logging(data: Any, logger: Logger | None = None) → Any¶: Run modules with Rich progress logging.

to_json() → str¶

to_proverif() → str¶: Export the pipeline as a simplistic ProVerif process.

to_tamarin() → str¶: Export the pipeline as a simplistic Tamarin model.

track_secret(name: str) → None¶: Mark a secret to be monitored in exported models.

tracked_secrets: list[str]¶

class cryptography_suite.pipeline.PipelineVisualizer(pipeline: Pipeline)¶

Bases: object

Simple ASCII pipeline visualizer.

render_ascii() → str¶

class cryptography_suite.pipeline.RSADecrypt(private_key: Any)¶

Bases: CryptoModule[str | bytes, bytes]

Decrypt RSA-OAEP ciphertext.

Parameters:: private_key (Any) – The RSAPrivateKey used for decryption.

private_key: Any¶

run(data: str | bytes) → bytes¶: Run the module on the provided data.

to_proverif() → str¶: Return a ProVerif representation of the module.

to_tamarin() → str¶: Return a Tamarin representation of the module.

class cryptography_suite.pipeline.RSAEncrypt(public_key: Any, raw_output: bool = False)¶

Bases: CryptoModule[bytes, str | bytes]

Encrypt data using RSA-OAEP.

Parameters:

public_key (Any) – The RSAPublicKey used for encryption.
raw_output (bool) – When True return raw bytes instead of Base64 encoded text.

public_key: Any¶

raw_output: bool = False¶

run(data: bytes) → str | bytes¶: Run the module on the provided data.

to_proverif() → str¶: Return a ProVerif representation of the module.

to_tamarin() → str¶: Return a Tamarin representation of the module.

cryptography_suite.pipeline.list_modules() → list[str]¶: Return the names of all registered pipeline modules.

cryptography_suite.pipeline.register_module(cls: type[CryptoModule[Any, Any]]) → type[CryptoModule[Any, Any]]¶

Backends can decorate classes with @register_module to expose custom pipeline steps. Registered classes appear in list_modules() and are importable from cryptography_suite.pipeline.