Adapters

An adapter wraps one cryptographic backend. It takes and returns plain Python types rather than request models, so it can be unit-tested without the web stack, and so a missing optional backend breaks only its own operation.

Base interface

base

Common interface shared by all cryptographic adapters.

Adapters deliberately speak plain Python types rather than the API's Pydantic models. Translation between the request envelope and an adapter call happens in core.manager. Keeping that boundary means an adapter can be imported and unit-tested without the web stack installed, and a backend that is missing from the environment only breaks its own operation.

CryptographicAdapter

Bases: ABC

Base class for every cryptographic backend adapter.

execute abstractmethod

execute(
    *,
    action: str,
    input_data: dict[str, Any],
    parameters: dict[str, Any],
    key_config: dict[str, Any] | None = None,
) -> AdapterResult

Run one operation against the backend.

Parameters:

Name	Type	Description	Default
action	str	Sub-operation to perform, e.g. "encrypt".	required
input_data	dict[str, Any]	Data to be processed.	required
parameters	dict[str, Any]	Operation-specific configuration.	required
key_config	dict[str, Any] | None	Key generation or key loading configuration.	None

Returns:

Type	Description
AdapterResult	A (result, metadata) pair.

Raises:

Type	Description
AdapterError	If the operation cannot be completed.

Source code in src/cryptography_manager/adapters/base.py

@abstractmethod
def execute(
    self,
    *,
    action: str,
    input_data: dict[str, Any],
    parameters: dict[str, Any],
    key_config: dict[str, Any] | None = None,
) -> AdapterResult:
    """Run one operation against the backend.

    Args:
        action: Sub-operation to perform, e.g. ``"encrypt"``.
        input_data: Data to be processed.
        parameters: Operation-specific configuration.
        key_config: Key generation or key loading configuration.

    Returns:
        A ``(result, metadata)`` pair.

    Raises:
        AdapterError: If the operation cannot be completed.
    """
    raise NotImplementedError

Differential privacy

differential_privacy

Differential privacy adapter backed by PyDP.

PyDP wraps Google's differential privacy library, so the noise calibration and sensitivity analysis are inherited from a well-reviewed implementation rather than reimplemented here. This adapter's job is to translate a configuration or request into a correctly-parameterised PyDP algorithm and to keep epsilon accounting honest.

Two execution paths are offered. execute_all runs a configured plan of queries in one go and debits the adapter's own budget; it exists for library and scripted use. execute_query runs a single query and leaves budget accounting to the caller, which is what the service layer needs because there the budget belongs to a user rather than to an adapter instance.

DifferentialPrivacyAdapter

DifferentialPrivacyAdapter(config: Config)

Bases: CryptographicAdapter

Differentially private aggregate queries over numeric data.

Initialise the adapter from configuration.

The query plan and data bounds are optional: a service handling per-request queries supplies them with each request, while a script driving a configured plan supplies them up front.

Parameters:

Name	Type	Description	Default
config	Config	Loaded configuration object.	required

Raises:

Type	Description
ConfigurationError	If the differential privacy section is absent or malformed.
DifferentialPrivacyError	If the configured plan cannot fit inside the configured budget.

Source code in src/cryptography_manager/adapters/differential_privacy.py

def __init__(self, config: Config):
    """Initialise the adapter from configuration.

    The query plan and data bounds are optional: a service handling
    per-request queries supplies them with each request, while a script
    driving a configured plan supplies them up front.

    Args:
        config: Loaded configuration object.

    Raises:
        ConfigurationError: If the differential privacy section is absent
            or malformed.
        DifferentialPrivacyError: If the configured plan cannot fit inside
            the configured budget.
    """
    try:
        section = config.get_submodule_config("differential_privacy")
        if not section:
            raise ConfigurationError(
                "Missing 'differential_privacy' configuration section",
                config_key="differential_privacy",
            )

        self.config: dict[str, Any] = section
        self.mechanism: str = section.get("default_mechanism", "laplace")
        self.delta: float | None = section.get("default_delta")
        self.global_budget: float = float(
            section.get("privacy_budget_limit", 1.0)
        )

        self.queries: list[dict[str, Any]] = list(
            section.get("queries") or []
        )

        data_cfg = section.get("data") or {}
        self.lower_bound: float | int | None = data_cfg.get("lower_bound")
        self.upper_bound: float | int | None = data_cfg.get("upper_bound")
    except ConfigurationError:
        raise
    except Exception as exc:
        # Only genuine configuration lookup failures are translated here.
        # Wrapping everything would disguise a budget or validation error
        # as a configuration problem and mislead whoever debugs it.
        raise ConfigurationError(f"DP adapter init failed: {exc}")

    planned = sum(float(q.get("epsilon", 0.0)) for q in self.queries)
    if planned > self.global_budget:
        raise DifferentialPrivacyError(
            f"Configured queries require e={planned}, which exceeds the "
            f"budget limit e={self.global_budget}",
            mechanism=self.mechanism,
        )

    self.budget = PrivacyBudget(
        total_epsilon=self.global_budget,
        delta=self.delta,
        mechanism=self.mechanism,
    )

    logger.info(
        f"DP adapter initialised: budget e={self.global_budget}, "
        f"{len(self.queries)} configured quer"
        f"{'y' if len(self.queries) == 1 else 'ies'} "
        f"planning e={planned}"
    )

execute_query

execute_query(
    query_type: str,
    epsilon: float,
    data: Sequence[Any] | NDArray[Any],
    lower_bound: float | int | None = None,
    upper_bound: float | int | None = None,
    budget: PrivacyBudget | None = None,
) -> float | int

Run a single differentially private query.

Budget accounting is the caller's responsibility unless budget is supplied. The service layer debits a per-user store instead, and double-debiting would silently halve every user's budget.

Parameters:

Name	Type	Description	Default
query_type	str	Name of the aggregate to compute.	required
epsilon	float	Privacy budget to spend on this query.	required
data	Sequence[Any] | NDArray[Any]	Numeric values to aggregate.	required
lower_bound	float | int | None	Lower value bound; falls back to configuration.	None
upper_bound	float | int | None	Upper value bound; falls back to configuration.	None
budget	PrivacyBudget | None	Optional budget to debit before executing.	None

Returns:

Type	Description
float | int	The noisy aggregate.

Raises:

Type	Description
DifferentialPrivacyError	If the query is unsupported, bounds are missing or invalid, or the backend fails.

Source code in src/cryptography_manager/adapters/differential_privacy.py

def execute_query(
    self,
    query_type: str,
    epsilon: float,
    data: Sequence[Any] | npt.NDArray[Any],
    lower_bound: float | int | None = None,
    upper_bound: float | int | None = None,
    budget: PrivacyBudget | None = None,
) -> float | int:
    """Run a single differentially private query.

    Budget accounting is the caller's responsibility unless ``budget`` is
    supplied. The service layer debits a per-user store instead, and
    double-debiting would silently halve every user's budget.

    Args:
        query_type: Name of the aggregate to compute.
        epsilon: Privacy budget to spend on this query.
        data: Numeric values to aggregate.
        lower_bound: Lower value bound; falls back to configuration.
        upper_bound: Upper value bound; falls back to configuration.
        budget: Optional budget to debit before executing.

    Returns:
        The noisy aggregate.

    Raises:
        DifferentialPrivacyError: If the query is unsupported, bounds are
            missing or invalid, or the backend fails.
    """
    if query_type not in self._QUERY_MAP:
        supported = ", ".join(sorted(self._QUERY_MAP))
        raise DifferentialPrivacyError(
            f"Unsupported query type '{query_type}'; "
            f"supported: {supported}",
            mechanism=self.mechanism,
            operation=query_type,
        )
    if epsilon <= 0:
        raise DifferentialPrivacyError(
            f"Epsilon must be > 0, got {epsilon}",
            mechanism=self.mechanism,
            operation=query_type,
        )

    values, dtype = self._prepare(data)

    kwargs: dict[str, Any] = {"epsilon": epsilon, "dtype": dtype}
    if query_type in self._REQUIRES_BOUNDS:
        low = lower_bound if lower_bound is not None else self.lower_bound
        high = upper_bound if upper_bound is not None else self.upper_bound
        if low is None or high is None:
            raise DifferentialPrivacyError(
                f"Query '{query_type}' requires lower_bound and "
                f"upper_bound, either per request or in configuration",
                mechanism=self.mechanism,
                operation=query_type,
            )
        if high <= low:
            raise DifferentialPrivacyError(
                f"upper_bound ({high}) must exceed lower_bound ({low})",
                mechanism=self.mechanism,
                operation=query_type,
            )
        kwargs["lower_bound"] = self._match_dtype(low, dtype)
        kwargs["upper_bound"] = self._match_dtype(high, dtype)

    # Debit before executing: a query that runs but fails to be accounted
    # for has already leaked privacy.
    if budget is not None:
        budget.spend(epsilon)

    try:
        algorithm = self._QUERY_MAP[query_type](**kwargs)
        return algorithm.quick_result(values)
    except DifferentialPrivacyError:
        raise
    except Exception as exc:
        raise DifferentialPrivacyError(
            f"Query '{query_type}' failed: {exc}",
            mechanism=self.mechanism,
            operation=query_type,
        )

execute_all

execute_all(
    data: Sequence[Any] | NDArray[Any],
) -> dict[str, float | int]

Run every configured query in order, debiting the adapter budget.

Parameters:

Name	Type	Description	Default
data	Sequence[Any] | NDArray[Any]	Numeric values to aggregate.	required

Returns:

Type	Description
dict[str, float | int]	Query name mapped to its noisy result.

Raises:

Type	Description
ConfigurationError	If no queries are configured.
DifferentialPrivacyError	If the budget is exceeded or a query fails.

Source code in src/cryptography_manager/adapters/differential_privacy.py

def execute_all(
    self, data: Sequence[Any] | npt.NDArray[Any]
) -> dict[str, float | int]:
    """Run every configured query in order, debiting the adapter budget.

    Args:
        data: Numeric values to aggregate.

    Returns:
        Query name mapped to its noisy result.

    Raises:
        ConfigurationError: If no queries are configured.
        DifferentialPrivacyError: If the budget is exceeded or a query
            fails.
    """
    if not self.queries:
        raise ConfigurationError(
            "No queries configured; provide a 'queries' list or call "
            "execute_query directly",
            config_key="differential_privacy.queries",
        )

    results: dict[str, float | int] = {}
    for query in self.queries:
        results[query["name"]] = self.execute_query(
            query_type=query["type"],
            epsilon=float(query["epsilon"]),
            data=data,
            lower_bound=query.get("lower_bound"),
            upper_bound=query.get("upper_bound"),
            budget=self.budget,
        )
    return results

execute

execute(
    *,
    action: str,
    input_data: dict[str, Any],
    parameters: dict[str, Any],
    key_config: dict[str, Any] | None = None,
) -> AdapterResult

Run one differentially private query for the service layer.

Parameters:

Name	Type	Description	Default
action	str	"query", or empty to default to it.	required
input_data	dict[str, Any]	{"data": [...]}.	required
parameters	dict[str, Any]	query_type and epsilon_cost, optionally lower_bound and upper_bound.	required
key_config	dict[str, Any] | None	Unused.	None

Returns:

Type	Description
AdapterResult	A (result, metadata) pair. Budget state is not included here;
AdapterResult	the service layer adds it from the per-user store.

Raises:

Type	Description
DifferentialPrivacyError	If the action is unknown or the query cannot be run.

Source code in src/cryptography_manager/adapters/differential_privacy.py

def execute(
    self,
    *,
    action: str,
    input_data: dict[str, Any],
    parameters: dict[str, Any],
    key_config: dict[str, Any] | None = None,
) -> AdapterResult:
    """Run one differentially private query for the service layer.

    Args:
        action: ``"query"``, or empty to default to it.
        input_data: ``{"data": [...]}``.
        parameters: ``query_type`` and ``epsilon_cost``, optionally
            ``lower_bound`` and ``upper_bound``.
        key_config: Unused.

    Returns:
        A ``(result, metadata)`` pair. Budget state is not included here;
        the service layer adds it from the per-user store.

    Raises:
        DifferentialPrivacyError: If the action is unknown or the query
            cannot be run.
    """
    if action not in ("", "query"):
        raise DifferentialPrivacyError(
            f"Unsupported differential privacy action: {action}",
            mechanism=self.mechanism,
            operation=action,
        )
    if "data" not in input_data:
        raise DifferentialPrivacyError(
            "input_data must contain 'data'", mechanism=self.mechanism
        )

    query_type = str(parameters.get("query_type", ""))
    epsilon = float(parameters.get("epsilon_cost", 0.0))
    result = self.execute_query(
        query_type=query_type,
        epsilon=epsilon,
        data=input_data["data"],
        lower_bound=parameters.get("lower_bound"),
        upper_bound=parameters.get("upper_bound"),
    )
    return (
        result,
        {
            "query_type": query_type,
            "epsilon_spent": epsilon,
            "mechanism": self.mechanism,
            "record_count": len(input_data["data"]),
        },
    )

Encryption and key management

encryption

Encryption, decryption and key management via Google Tink.

Tink is used rather than a raw cipher library because it is misuse-resistant: algorithm parameters, nonce generation and ciphertext tagging are handled internally, and only vetted primitives are reachable. This adapter exposes the AEAD primitive plus the keyset lifecycle operations (generation, serialisation, loading and rotation).

The byte-level operations are the core; the file helpers are a thin convenience layer for the library and example code. A REST caller cannot pass file paths, so nothing above this module depends on the filesystem.

EncryptionAdapter

Bases: CryptographicAdapter

Symmetric authenticated encryption and keyset management via Tink.

generate_keyset staticmethod

generate_keyset(
    template: str = DEFAULT_TEMPLATE,
) -> KeysetHandle

Generate a new keyset containing a single primary key.

Parameters:

Name	Type	Description	Default
template	str	Name of an allowlisted AEAD key template.	DEFAULT_TEMPLATE

Returns:

Type	Description
KeysetHandle	A handle to the newly generated keyset.

Source code in src/cryptography_manager/adapters/encryption.py

@staticmethod
def generate_keyset(
    template: str = DEFAULT_TEMPLATE,
) -> tink.KeysetHandle:
    """Generate a new keyset containing a single primary key.

    Args:
        template: Name of an allowlisted AEAD key template.

    Returns:
        A handle to the newly generated keyset.
    """
    return tink.new_keyset_handle(_template(template))

serialize_keyset staticmethod

serialize_keyset(handle: KeysetHandle) -> str

Serialise a keyset to cleartext JSON.

The output contains raw key material and must be stored in a secret manager or encrypted at rest. It is never written to an audit record.

Parameters:

Name	Type	Description	Default
handle	KeysetHandle	The keyset to serialise.	required

Returns:

Type	Description
str	The keyset as a JSON string.

Source code in src/cryptography_manager/adapters/encryption.py

@staticmethod
def serialize_keyset(handle: tink.KeysetHandle) -> str:
    """Serialise a keyset to cleartext JSON.

    The output contains raw key material and must be stored in a secret
    manager or encrypted at rest. It is never written to an audit record.

    Args:
        handle: The keyset to serialise.

    Returns:
        The keyset as a JSON string.
    """
    return json_proto_keyset_format.serialize(
        handle, secret_key_access.TOKEN
    )

load_keyset staticmethod

load_keyset(serialized: str) -> KeysetHandle

Load a keyset from its cleartext JSON representation.

Parameters:

Name	Type	Description	Default
serialized	str	JSON produced by :meth:serialize_keyset.	required

Returns:

Type	Description
KeysetHandle	A handle to the loaded keyset.

Raises:

Type	Description
KeyManagementError	If the keyset cannot be parsed.

Source code in src/cryptography_manager/adapters/encryption.py

@staticmethod
def load_keyset(serialized: str) -> tink.KeysetHandle:
    """Load a keyset from its cleartext JSON representation.

    Args:
        serialized: JSON produced by :meth:`serialize_keyset`.

    Returns:
        A handle to the loaded keyset.

    Raises:
        KeyManagementError: If the keyset cannot be parsed.
    """
    try:
        return json_proto_keyset_format.parse(
            serialized, secret_key_access.TOKEN
        )
    except Exception as exc:
        raise KeyManagementError(
            f"Failed to parse keyset: {exc}", operation="load_keyset"
        )

rotate_keyset staticmethod

rotate_keyset(
    handle: KeysetHandle, template: str = DEFAULT_TEMPLATE
) -> KeysetHandle

Add a fresh key to a keyset and promote it to primary.

Existing keys are retained and stay enabled, so ciphertexts produced before the rotation remain decryptable. Tink prefixes each ciphertext with its key ID, which is what makes this work.

tink-py 1.12 exposes no KeysetManager, so the merge is done at the protobuf layer using the public tink.proto and proto_keyset_format modules rather than private attributes.

Parameters:

Name	Type	Description	Default
handle	KeysetHandle	The keyset to rotate.	required
template	str	Template for the new primary key.	DEFAULT_TEMPLATE

Returns:

Type	Description
KeysetHandle	A handle to the rotated keyset.

Raises:

Type	Description
KeyManagementError	If the rotation fails.

Source code in src/cryptography_manager/adapters/encryption.py

@staticmethod
def rotate_keyset(
    handle: tink.KeysetHandle,
    template: str = DEFAULT_TEMPLATE,
) -> tink.KeysetHandle:
    """Add a fresh key to a keyset and promote it to primary.

    Existing keys are retained and stay enabled, so ciphertexts produced
    before the rotation remain decryptable. Tink prefixes each ciphertext
    with its key ID, which is what makes this work.

    tink-py 1.12 exposes no ``KeysetManager``, so the merge is done at the
    protobuf layer using the public ``tink.proto`` and
    ``proto_keyset_format`` modules rather than private attributes.

    Args:
        handle: The keyset to rotate.
        template: Template for the new primary key.

    Returns:
        A handle to the rotated keyset.

    Raises:
        KeyManagementError: If the rotation fails.
    """
    try:
        current = _Keyset.FromString(
            proto_keyset_format.serialize(handle, secret_key_access.TOKEN)
        )
        fresh = _Keyset.FromString(
            proto_keyset_format.serialize(
                tink.new_keyset_handle(_template(template)),
                secret_key_access.TOKEN,
            )
        )

        existing_ids = {k.key_id for k in current.key}
        for key in fresh.key:
            if key.key_id in existing_ids:
                raise KeyManagementError(
                    f"Generated key ID {key.key_id} already present",
                    operation="rotate_keyset",
                )
            current.key.append(key)

        current.primary_key_id = fresh.primary_key_id

        rotated = proto_keyset_format.parse(
            current.SerializeToString(), secret_key_access.TOKEN
        )
    except KeyManagementError:
        raise
    except Exception as exc:
        raise KeyManagementError(
            f"Failed to rotate keyset: {exc}", operation="rotate_keyset"
        )

    logger.info(
        f"Keyset rotated: new primary key ID {current.primary_key_id}, "
        f"{len(current.key)} key(s) retained"
    )
    return rotated

keyset_info staticmethod

keyset_info(handle: KeysetHandle) -> dict[str, Any]

Describe a keyset without exposing key material.

Only key IDs, type URLs, status and prefix type are returned, making the result safe to log and to return over the API.

Parameters:

Name	Type	Description	Default
handle	KeysetHandle	The keyset to describe.	required

Returns:

Type	Description
dict[str, Any]	Keyset metadata suitable for audit and API responses.

Source code in src/cryptography_manager/adapters/encryption.py

@staticmethod
def keyset_info(handle: tink.KeysetHandle) -> dict[str, Any]:
    """Describe a keyset without exposing key material.

    Only key IDs, type URLs, status and prefix type are returned, making
    the result safe to log and to return over the API.

    Args:
        handle: The keyset to describe.

    Returns:
        Keyset metadata suitable for audit and API responses.
    """
    info = handle.keyset_info()
    return {
        "primary_key_id": info.primary_key_id,
        "key_count": len(info.key_info),
        "keys": [
            {
                "key_id": k.key_id,
                "type_url": k.type_url,
                "status": _KeyStatus.Name(k.status),
                "output_prefix_type": _OutputPrefix.Name(
                    k.output_prefix_type
                ),
            }
            for k in info.key_info
        ],
    }

encrypt_bytes

encrypt_bytes(
    plaintext: bytes,
    handle: KeysetHandle,
    associated_data: bytes = b"",
) -> bytes

Encrypt bytes with authenticated encryption.

Parameters:

Name	Type	Description	Default
plaintext	bytes	Data to encrypt.	required
handle	KeysetHandle	Keyset providing the AEAD primitive.	required
associated_data	bytes	Context bound to the ciphertext. It is authenticated but not encrypted, and the exact same value must be supplied to decrypt.	b''

Returns:

Type	Description
bytes	The ciphertext.

Raises:

Type	Description
StandardCryptographyError	If encryption fails.

Source code in src/cryptography_manager/adapters/encryption.py

def encrypt_bytes(
    self,
    plaintext: bytes,
    handle: tink.KeysetHandle,
    associated_data: bytes = b"",
) -> bytes:
    """Encrypt bytes with authenticated encryption.

    Args:
        plaintext: Data to encrypt.
        handle: Keyset providing the AEAD primitive.
        associated_data: Context bound to the ciphertext. It is
            authenticated but not encrypted, and the exact same value must
            be supplied to decrypt.

    Returns:
        The ciphertext.

    Raises:
        StandardCryptographyError: If encryption fails.
    """
    try:
        return self._primitive(handle).encrypt(
            plaintext, associated_data
        )
    except StandardCryptographyError:
        raise
    except Exception as exc:
        raise StandardCryptographyError(
            f"Encryption failed: {exc}",
            algorithm="aead",
            operation="encrypt",
        )

decrypt_bytes

decrypt_bytes(
    ciphertext: bytes,
    handle: KeysetHandle,
    associated_data: bytes = b"",
) -> bytes

Decrypt bytes produced by :meth:encrypt_bytes.

Parameters:

Name	Type	Description	Default
ciphertext	bytes	Data to decrypt.	required
handle	KeysetHandle	Keyset providing the AEAD primitive.	required
associated_data	bytes	The exact value used at encryption time.	b''

Returns:

Type	Description
bytes	The recovered plaintext.

Raises:

Type	Description
StandardCryptographyError	If decryption or authentication fails.

Source code in src/cryptography_manager/adapters/encryption.py

def decrypt_bytes(
    self,
    ciphertext: bytes,
    handle: tink.KeysetHandle,
    associated_data: bytes = b"",
) -> bytes:
    """Decrypt bytes produced by :meth:`encrypt_bytes`.

    Args:
        ciphertext: Data to decrypt.
        handle: Keyset providing the AEAD primitive.
        associated_data: The exact value used at encryption time.

    Returns:
        The recovered plaintext.

    Raises:
        StandardCryptographyError: If decryption or authentication fails.
    """
    try:
        return self._primitive(handle).decrypt(
            ciphertext, associated_data
        )
    except StandardCryptographyError:
        raise
    except Exception as exc:
        # Covers a wrong key, mismatched associated data and tampering
        # alike; the distinction is deliberately not surfaced.
        raise StandardCryptographyError(
            f"Decryption failed: {exc}",
            algorithm="aead",
            operation="decrypt",
        )

encrypt_file

encrypt_file(
    path: str | Path,
    handle: KeysetHandle,
    output: str | Path | None = None,
) -> Path

Encrypt a file, binding the ciphertext to its basename.

Parameters:

Name	Type	Description	Default
path	str | Path	File to encrypt.	required
handle	KeysetHandle	Keyset providing the AEAD primitive.	required
output	str | Path | None	Destination; defaults to path with .enc appended.	None

Returns:

Type	Description
Path	Path to the ciphertext file.

Source code in src/cryptography_manager/adapters/encryption.py

def encrypt_file(
    self,
    path: str | Path,
    handle: tink.KeysetHandle,
    output: str | Path | None = None,
) -> Path:
    """Encrypt a file, binding the ciphertext to its basename.

    Args:
        path: File to encrypt.
        handle: Keyset providing the AEAD primitive.
        output: Destination; defaults to ``path`` with ``.enc`` appended.

    Returns:
        Path to the ciphertext file.
    """
    path = Path(path)
    destination = Path(output) if output else path.with_suffix(
        path.suffix + ".enc"
    )
    associated_data = path.name.encode("utf-8")
    ciphertext = self.encrypt_bytes(
        path.read_bytes(), handle, associated_data
    )
    destination.write_bytes(ciphertext)
    return destination

decrypt_file

decrypt_file(
    path: str | Path,
    handle: KeysetHandle,
    output: str | Path | None = None,
) -> Path

Decrypt a file produced by :meth:encrypt_file.

Parameters:

Name	Type	Description	Default
path	str | Path	Ciphertext file; must end in .enc.	required
handle	KeysetHandle	Keyset providing the AEAD primitive.	required
output	str | Path | None	Destination; defaults to path with .dec appended.	None

Returns:

Type	Description
Path	Path to the recovered plaintext file.

Raises:

Type	Description
StandardCryptographyError	If the filename is not .enc.

Source code in src/cryptography_manager/adapters/encryption.py

def decrypt_file(
    self,
    path: str | Path,
    handle: tink.KeysetHandle,
    output: str | Path | None = None,
) -> Path:
    """Decrypt a file produced by :meth:`encrypt_file`.

    Args:
        path: Ciphertext file; must end in ``.enc``.
        handle: Keyset providing the AEAD primitive.
        output: Destination; defaults to ``path`` with ``.dec`` appended.

    Returns:
        Path to the recovered plaintext file.

    Raises:
        StandardCryptographyError: If the filename is not ``.enc``.
    """
    path = Path(path)
    if path.suffix != ".enc":
        raise StandardCryptographyError(
            f"Expected a '.enc' file, got '{path.name}'",
            algorithm="aead",
            operation="decrypt_file",
        )
    # The associated data is the *original* filename, so strip '.enc'.
    associated_data = path.with_suffix("").name.encode("utf-8")
    destination = Path(output) if output else path.with_suffix(
        path.suffix + ".dec"
    )
    plaintext = self.decrypt_bytes(
        path.read_bytes(), handle, associated_data
    )
    destination.write_bytes(plaintext)
    return destination

execute

execute(
    *,
    action: str,
    input_data: dict[str, Any],
    parameters: dict[str, Any],
    key_config: dict[str, Any] | None = None,
) -> AdapterResult

Run an encryption or key-management action.

Parameters:

Name	Type	Description	Default
action	str	One of encrypt, decrypt, generate_key, rotate_key or key_info.	required
input_data	dict[str, Any]	For encrypt, {"plaintext": <str>}; for decrypt, {"ciphertext": <base64 str>}.	required
parameters	dict[str, Any]	May carry associated_data and template.	required
key_config	dict[str, Any] | None	Keyset selection; see :meth:_resolve_keyset.	None

Returns:

Type	Description
AdapterResult	A (result, metadata) pair. Ciphertext is base64-encoded so it
AdapterResult	survives a JSON round trip.

Raises:

Type	Description
StandardCryptographyError	If the action is unknown or fails.

Source code in src/cryptography_manager/adapters/encryption.py

def execute(
    self,
    *,
    action: str,
    input_data: dict[str, Any],
    parameters: dict[str, Any],
    key_config: dict[str, Any] | None = None,
) -> AdapterResult:
    """Run an encryption or key-management action.

    Args:
        action: One of ``encrypt``, ``decrypt``, ``generate_key``,
            ``rotate_key`` or ``key_info``.
        input_data: For ``encrypt``, ``{"plaintext": <str>}``; for
            ``decrypt``, ``{"ciphertext": <base64 str>}``.
        parameters: May carry ``associated_data`` and ``template``.
        key_config: Keyset selection; see :meth:`_resolve_keyset`.

    Returns:
        A ``(result, metadata)`` pair. Ciphertext is base64-encoded so it
        survives a JSON round trip.

    Raises:
        StandardCryptographyError: If the action is unknown or fails.
    """
    associated_data = parameters.get("associated_data", "")
    associated = (
        associated_data.encode("utf-8")
        if isinstance(associated_data, str)
        else bytes(associated_data)
    )

    if action == "generate_key":
        handle = self.generate_keyset(
            (key_config or {}).get("template", DEFAULT_TEMPLATE)
        )
        return (
            {"keyset": self.serialize_keyset(handle)},
            {"keyset_info": self.keyset_info(handle)},
        )

    if action == "rotate_key":
        handle = self._resolve_keyset(key_config)
        rotated = self.rotate_keyset(
            handle, (key_config or {}).get("template", DEFAULT_TEMPLATE)
        )
        return (
            {"keyset": self.serialize_keyset(rotated)},
            {"keyset_info": self.keyset_info(rotated)},
        )

    if action == "key_info":
        handle = self._resolve_keyset(key_config)
        return ({}, {"keyset_info": self.keyset_info(handle)})

    if action == "encrypt":
        if "plaintext" not in input_data:
            raise StandardCryptographyError(
                "input_data must contain 'plaintext'",
                algorithm="aead",
                operation="encrypt",
            )
        handle = self._resolve_keyset(key_config)
        ciphertext = self.encrypt_bytes(
            str(input_data["plaintext"]).encode("utf-8"),
            handle,
            associated,
        )
        return (
            {"ciphertext": base64.b64encode(ciphertext).decode("ascii")},
            {"keyset_info": self.keyset_info(handle)},
        )

    if action == "decrypt":
        if "ciphertext" not in input_data:
            raise StandardCryptographyError(
                "input_data must contain 'ciphertext'",
                algorithm="aead",
                operation="decrypt",
            )
        try:
            raw = base64.b64decode(
                str(input_data["ciphertext"]), validate=True
            )
        except Exception as exc:
            raise StandardCryptographyError(
                f"ciphertext is not valid base64: {exc}",
                algorithm="aead",
                operation="decrypt",
            )
        handle = self._resolve_keyset(key_config)
        plaintext = self.decrypt_bytes(raw, handle, associated)
        return (
            {"plaintext": plaintext.decode("utf-8", errors="replace")},
            {"keyset_info": self.keyset_info(handle)},
        )

    raise StandardCryptographyError(
        f"Unsupported encryption action: {action}",
        algorithm="aead",
        operation=action,
    )