Skip to content

Core

The core layer owns everything that happens around the cryptography: which adapter runs, whether the caller can afford the query, and what gets recorded.

Manager

manager

Unified entry point for every cryptographic operation.

The manager owns adapter lifecycles and translates the request envelope into an adapter call. Adapters are built lazily and cached: constructing one can be expensive, and a backend that is absent from the environment should only break the operation that needs it rather than preventing the service from starting.

CryptographyManager 

CryptographyManager(
    config: Config | None = None,
    config_file: str | None = None,
)

Routes cryptographic requests to the appropriate backend adapter.

Initialize the Cryptography Manager.

Parameters:

Name Type Description Default
config Config | None

Optional configuration object.

 None
config_file str | None

Optional path to a configuration file.

 None

Raises:

Type Description
ConfigurationError

If the configuration file cannot be loaded.
CryptographyManagerError

If initialization otherwise fails.
Source code in src/cryptography_manager/core/manager.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def __init__(
    self,
    config: Config | None = None,
    config_file: str | None = None,
) -> None:
    """Initialize the Cryptography Manager.

    Args:
        config: Optional configuration object.
        config_file: Optional path to a configuration file.

    Raises:
        ConfigurationError: If the configuration file cannot be loaded.
        CryptographyManagerError: If initialization otherwise fails.
    """
    try:
        if config is not None:
            self.config = config
        else:
            self.config = Config()
            if config_file:
                self.config.load_from_file(config_file)
    except ConfigurationError:
        raise
    except Exception as exc:
        raise CryptographyManagerError(
            f"Failed to initialize Cryptography Manager: {exc}"
        )

    self._instances: dict[str, Any] = {}
    logger.info("Cryptography Manager initialized successfully")
differential_privacy 
differential_privacy() -> 'DifferentialPrivacyAdapter'

Return the DP adapter, building it on first use.

Source code in src/cryptography_manager/core/manager.py 
84
85
86
87
88
89
90
91
92
def differential_privacy(self) -> "DifferentialPrivacyAdapter":
    """Return the DP adapter, building it on first use."""
    if "differential_privacy" not in self._instances:
        from ..adapters import DifferentialPrivacyAdapter

        self._instances["differential_privacy"] = (
            DifferentialPrivacyAdapter(self.config)
        )
    return self._instances["differential_privacy"]
encryption 
encryption() -> 'EncryptionAdapter'

Return the encryption adapter, building it on first use.

Source code in src/cryptography_manager/core/manager.py 
 94
 95
 96
 97
 98
 99
100
def encryption(self) -> "EncryptionAdapter":
    """Return the encryption adapter, building it on first use."""
    if "encryption" not in self._instances:
        from ..adapters import EncryptionAdapter

        self._instances["encryption"] = EncryptionAdapter()
    return self._instances["encryption"]
adapter_for 
adapter_for(operation: Operation) -> 'CryptographicAdapter'

Return the adapter servicing an operation.

Parameters:

Name Type Description Default
operation Operation

The requested operation.

 required

Returns:

Type Description
'CryptographicAdapter'

The adapter instance.

Raises:

Type Description
OperationNotImplementedError

If the operation is recognised but has no adapter yet.
Source code in src/cryptography_manager/core/manager.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def adapter_for(self, operation: Operation) -> "CryptographicAdapter":
    """Return the adapter servicing an operation.

    Args:
        operation: The requested operation.

    Returns:
        The adapter instance.

    Raises:
        OperationNotImplementedError: If the operation is recognised but
            has no adapter yet.
    """
    if operation in PENDING_OPERATIONS:
        raise OperationNotImplementedError(
            f"Operation '{operation.value}' is not implemented yet",
            operation=operation.value,
        )
    if operation is Operation.DIFFERENTIAL_PRIVACY:
        return self.differential_privacy()
    if operation in (Operation.ENCRYPTION, Operation.KEY_MANAGEMENT):
        return self.encryption()
    raise OperationNotImplementedError(
        f"No adapter registered for operation '{operation.value}'",
        operation=operation.value,
    )
backend_for 
backend_for(operation: Operation) -> Backend

Return the backend library servicing an operation.

Source code in src/cryptography_manager/core/manager.py 
129
130
131
132
133
134
135
136
137
def backend_for(self, operation: Operation) -> Backend:
    """Return the backend library servicing an operation."""
    if operation is Operation.DIFFERENTIAL_PRIVACY:
        return Backend.PYDP
    if operation in (Operation.ENCRYPTION, Operation.KEY_MANAGEMENT):
        return Backend.TINK
    if operation is Operation.HOMOMORPHIC_ENCRYPTION:
        return Backend.PYFHEL
    return Backend.MPYC
execute 
execute(
    request: CryptographyRequest,
) -> tuple[Any, dict[str, Any]]

Run a request against its backend.

Budget accounting and auditing are handled by the workflow layer, not here, so that this method stays usable directly from library code.

Parameters:

Name Type Description Default
request CryptographyRequest

The validated request envelope.

 required

Returns:

Type Description
tuple[Any, dict[str, Any]]

A (result, metadata) pair.

Raises:

Type Description
OperationNotImplementedError

If the operation has no adapter.
AdapterError

If the backend fails.
Source code in src/cryptography_manager/core/manager.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def execute(
    self, request: CryptographyRequest
) -> tuple[Any, dict[str, Any]]:
    """Run a request against its backend.

    Budget accounting and auditing are handled by the workflow layer, not
    here, so that this method stays usable directly from library code.

    Args:
        request: The validated request envelope.

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

    Raises:
        OperationNotImplementedError: If the operation has no adapter.
        AdapterError: If the backend fails.
    """
    adapter = self.adapter_for(request.operation)
    action = str(
        request.parameters.get(
            "action", DEFAULT_ACTIONS.get(request.operation, "")
        )
    )
    return adapter.execute(
        action=action,
        input_data=request.input_data,
        parameters=request.parameters,
        key_config=request.key_config,
    )

Workflow

workflow

Request workflow: validate, pre-check, execute, audit, respond.

This is where the pieces are sequenced. Every path through it -- including every rejection -- ends in exactly one audit record, so a request that was refused is as traceable as one that succeeded.

The ordering deviates from the obvious "check the budget first" in one place: a differential privacy request's budget cost lives in its parameters, so those parameters have to be validated before the budget can be checked against them. Validating first also means a malformed request never touches budget state.

WorkflowOutcome 

WorkflowOutcome(
    status: WorkflowStatus, response: CryptographyResponse
)

The result of running a workflow, ready to be turned into a response.

Source code in src/cryptography_manager/core/workflow.py 
53
54
55
56
57
58
59
def __init__(
    self,
    status: WorkflowStatus,
    response: CryptographyResponse,
) -> None:
    self.status = status
    self.response = response

CryptographyWorkflow 

CryptographyWorkflow(
    manager: CryptographyManager,
    audit: AuditLogger,
    budget_store: BudgetStore,
)

Sequences a single cryptographic request end to end.

Initialise the workflow.

Parameters:

Name Type Description Default
manager CryptographyManager

Dispatches requests to backend adapters.

 required
audit AuditLogger

Records every outcome.

 required
budget_store BudgetStore

Per-user privacy budget accounting.

 required
Source code in src/cryptography_manager/core/workflow.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def __init__(
    self,
    manager: CryptographyManager,
    audit: AuditLogger,
    budget_store: BudgetStore,
) -> None:
    """Initialise the workflow.

    Args:
        manager: Dispatches requests to backend adapters.
        audit: Records every outcome.
        budget_store: Per-user privacy budget accounting.
    """
    self.manager = manager
    self.audit = audit
    self.budget_store = budget_store
reject 
reject(
    status: WorkflowStatus,
    user_id: str | None = None,
    error: str | None = None,
    operation: str | None = None,
    backend: str | None = None,
) -> WorkflowOutcome

Record a rejection that happened before execution could start.

Used for failures the request never survives long enough to reach the main path, such as a body too large to parse or one that does not fit the schema. It still writes an audit record, because a refused request has to be as traceable as one that ran.

Parameters:

Name Type Description Default
status WorkflowStatus

The workflow outcome.

 required
user_id str | None

Authenticated user, if known at this point.

 None
error str | None

Human-readable reason for the rejection.

 None
operation str | None

Requested operation, if it could be determined.

 None
backend str | None

Requested backend, if it could be determined.

 None

Returns:

Type Description
WorkflowOutcome

The outcome, carrying both status and response body.
Source code in src/cryptography_manager/core/workflow.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def reject(
    self,
    status: WorkflowStatus,
    user_id: str | None = None,
    error: str | None = None,
    operation: str | None = None,
    backend: str | None = None,
) -> WorkflowOutcome:
    """Record a rejection that happened before execution could start.

    Used for failures the request never survives long enough to reach the
    main path, such as a body too large to parse or one that does not fit
    the schema. It still writes an audit record, because a refused request
    has to be as traceable as one that ran.

    Args:
        status: The workflow outcome.
        user_id: Authenticated user, if known at this point.
        error: Human-readable reason for the rejection.
        operation: Requested operation, if it could be determined.
        backend: Requested backend, if it could be determined.

    Returns:
        The outcome, carrying both status and response body.
    """
    record = self.audit.record(
        status=status.value,
        user_id=user_id,
        operation=operation,
        backend=backend,
        execution_time=0.0,
        parameters={},
        error_message=error,
    )
    return WorkflowOutcome(
        status,
        CryptographyResponse(
            status=status.value,
            execution_time=0.0,
            audit_id=record.audit_id,
            errors=error,
        ),
    )
run 
run(
    request: CryptographyRequest | None,
    user_id: str | None,
    auth_error: str | None = None,
) -> WorkflowOutcome

Execute one request and produce its response.

Parameters:

Name Type Description Default
request CryptographyRequest | None

The parsed request, or None if authorization failed before the body was considered.

 required
user_id str | None

Authenticated user, or None if unauthenticated.

 required
auth_error str | None

Why authorization failed, if it did.

 None

Returns:

Type Description
WorkflowOutcome

The workflow outcome, carrying both status and response body.
Source code in src/cryptography_manager/core/workflow.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
def run(
    self,
    request: CryptographyRequest | None,
    user_id: str | None,
    auth_error: str | None = None,
) -> WorkflowOutcome:
    """Execute one request and produce its response.

    Args:
        request: The parsed request, or ``None`` if authorization failed
            before the body was considered.
        user_id: Authenticated user, or ``None`` if unauthenticated.
        auth_error: Why authorization failed, if it did.

    Returns:
        The workflow outcome, carrying both status and response body.
    """
    started = time.perf_counter()
    operation = request.operation.value if request else None
    backend = request.backend.value if request else None
    parameters = dict(request.parameters) if request else {}

    def finish(
        status: WorkflowStatus,
        result: Any = None,
        metadata: dict[str, Any] | None = None,
        error: str | None = None,
    ) -> WorkflowOutcome:
        elapsed = time.perf_counter() - started
        record = self.audit.record(
            status=status.value,
            user_id=user_id,
            operation=operation,
            backend=backend,
            execution_time=elapsed,
            parameters=parameters,
            error_message=error,
        )
        return WorkflowOutcome(
            status,
            CryptographyResponse(
                status=status.value,
                result=result,
                metadata=metadata or {},
                execution_time=round(elapsed, 6),
                audit_id=record.audit_id,
                errors=error,
            ),
        )

    # 1. Authorization.
    if user_id is None or request is None:
        return finish(
            WorkflowStatus.UNAUTHORIZED,
            error=auth_error or "Missing or invalid authorization token",
        )

    # 2. Envelope validation: is this backend able to do this operation?
    incompatible = request.check_backend_compatible()
    if incompatible:
        return finish(WorkflowStatus.VALIDATION_ERROR, error=incompatible)

    action = str(
        request.parameters.get("action", "")
    ) or _default_action(request.operation)
    allowed = VALID_ACTIONS.get(request.operation)
    if allowed is not None and action not in allowed:
        return finish(
            WorkflowStatus.VALIDATION_ERROR,
            error=(
                f"Unsupported action '{action}' for operation "
                f"'{request.operation.value}'; supported: "
                f"{', '.join(sorted(allowed))}"
            ),
        )

    # 3. Operation-specific validation and pre-checks.
    epsilon_cost: float | None = None
    if request.operation is Operation.DIFFERENTIAL_PRIVACY:
        try:
            dp_params = DPParameters(
                **{
                    k: v
                    for k, v in request.parameters.items()
                    if k != "action"
                }
            )
        except ValidationError as exc:
            return finish(
                WorkflowStatus.VALIDATION_ERROR,
                error=_first_error(exc),
            )
        epsilon_cost = dp_params.epsilon_cost

        remaining = self.budget_store.remaining(user_id)
        if self.budget_store.is_exhausted(user_id):
            return finish(
                WorkflowStatus.BUDGET_EXHAUSTED,
                metadata=self.budget_store.info(user_id),
                error=(
                    f"Privacy budget exhausted for user '{user_id}'"
                ),
            )
        if epsilon_cost > remaining:
            return finish(
                WorkflowStatus.INSUFFICIENT_BUDGET,
                metadata=self.budget_store.info(user_id),
                error=(
                    f"Requested cost e={epsilon_cost} exceeds remaining "
                    f"budget e={remaining}"
                ),
            )

    # 4. Execution.
    try:
        result, metadata = self.manager.execute(request)
    except OperationNotImplementedError as exc:
        return finish(WorkflowStatus.NOT_IMPLEMENTED, error=str(exc))
    except (CryptographyManagerError, ValueError) as exc:
        # A backend refusing malformed input is the caller's problem, so
        # this is reported as a validation failure rather than a fault.
        return finish(WorkflowStatus.VALIDATION_ERROR, error=str(exc))
    except Exception as exc:  # pragma: no cover - unexpected fault
        logger.exception("Unexpected failure during execution")
        return finish(
            WorkflowStatus.FAILED, error=f"Execution failed: {exc}"
        )

    # 5. Budget is debited only after the query actually ran, so a failed
    # query does not cost the user any privacy.
    if epsilon_cost is not None:
        try:
            self.budget_store.spend(user_id, epsilon_cost)
        except DifferentialPrivacyError as exc:
            # Reachable only if a concurrent request drained the budget
            # between the check above and here.
            return finish(
                WorkflowStatus.INSUFFICIENT_BUDGET,
                metadata=self.budget_store.info(user_id),
                error=str(exc),
            )
        metadata = {**metadata, **self.budget_store.info(user_id)}

    return finish(WorkflowStatus.SUCCESS, result=result, metadata=metadata)

Privacy budgets

budget

Privacy budget accounting.

Differential privacy guarantees only hold if the total epsilon spent against a dataset is bounded. The remaining budget is retrieved per user before each query runs, which means accounting has to outlive a single adapter instance -- hence a store keyed by user rather than a counter living on the adapter.

Two rejection states are tracked separately: a budget already exhausted (nothing left at all) and one merely insufficient for the requested query. They map to different HTTP responses, and the distinction is useful to a caller deciding whether to retry with a cheaper query or give up entirely.

PrivacyBudget dataclass 

PrivacyBudget(
    total_epsilon: float,
    delta: float | None = None,
    mechanism: str = "laplace",
)

Epsilon accounting for a single dataset or session.

is_exhausted property 
is_exhausted: bool

Whether no budget remains at all.

can_afford 
can_afford(epsilon: float) -> bool

Whether epsilon can be spent without overdrawing.

Source code in src/cryptography_manager/core/budget.py 
57
58
59
def can_afford(self, epsilon: float) -> bool:
    """Whether ``epsilon`` can be spent without overdrawing."""
    return epsilon <= self.remaining_epsilon + EPSILON_TOLERANCE
spend 
spend(epsilon: float) -> None

Consume epsilon from the remaining budget.

Parameters:

Name Type Description Default
epsilon float

Privacy cost of the query.

 required

Raises:

Type Description
DifferentialPrivacyError

If the cost is non-positive or exceeds the remaining budget.
Source code in src/cryptography_manager/core/budget.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def spend(self, epsilon: float) -> None:
    """Consume epsilon from the remaining budget.

    Args:
        epsilon: Privacy cost of the query.

    Raises:
        DifferentialPrivacyError: If the cost is non-positive or exceeds
            the remaining budget.
    """
    if epsilon <= 0:
        raise DifferentialPrivacyError(
            "Spent epsilon must be > 0", mechanism=self.mechanism
        )
    if not self.can_afford(epsilon):
        raise DifferentialPrivacyError(
            f"Privacy budget exceeded: requested e={epsilon}, "
            f"remaining e={self.remaining_epsilon}",
            mechanism=self.mechanism,
        )
    # Clamp so repeated spending cannot drift below zero.
    self.remaining_epsilon = max(0.0, self.remaining_epsilon - epsilon)
reset 
reset() -> None

Restore the full budget. Only valid for a new dataset.

Source code in src/cryptography_manager/core/budget.py 
84
85
86
def reset(self) -> None:
    """Restore the full budget. Only valid for a new dataset."""
    self.remaining_epsilon = self.total_epsilon
info 
info() -> dict[str, float | None]

Return the current accounting state.

Source code in src/cryptography_manager/core/budget.py 
88
89
90
91
92
93
94
95
def info(self) -> dict[str, float | None]:
    """Return the current accounting state."""
    return {
        "total_epsilon": self.total_epsilon,
        "remaining_epsilon": self.remaining_epsilon,
        "spent_epsilon": self.total_epsilon - self.remaining_epsilon,
        "delta": self.delta,
    }

BudgetStore 

BudgetStore(
    total_epsilon: float, delta: float | None = None
)

Bases: ABC

Per-user privacy budget accounting.

Initialise the store.

Parameters:

Name Type Description Default
total_epsilon float

Budget granted to each user.

 required
delta float | None

Privacy leakage probability recorded alongside the budget.

 None
Source code in src/cryptography_manager/core/budget.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def __init__(self, total_epsilon: float, delta: float | None = None):
    """Initialise the store.

    Args:
        total_epsilon: Budget granted to each user.
        delta: Privacy leakage probability recorded alongside the budget.
    """
    if total_epsilon <= 0:
        raise DifferentialPrivacyError(
            "Total privacy budget epsilon must be > 0"
        )
    self.total_epsilon = total_epsilon
    self.delta = delta
    self._lock = threading.Lock()
remaining 
remaining(user_id: str) -> float

Return the epsilon still available to a user.

Source code in src/cryptography_manager/core/budget.py 
128
129
130
131
def remaining(self, user_id: str) -> float:
    """Return the epsilon still available to a user."""
    with self._lock:
        return self._load().get(user_id, self.total_epsilon)
spend 
spend(user_id: str, epsilon: float) -> float

Deduct epsilon from a user's budget.

Parameters:

Name Type Description Default
user_id str

Identifier of the requesting user.

 required
epsilon float

Privacy cost of the query.

 required

Returns:

Type Description
float

The remaining budget after the deduction.

Raises:

Type Description
DifferentialPrivacyError

If the cost is non-positive or exceeds the remaining budget.
Source code in src/cryptography_manager/core/budget.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def spend(self, user_id: str, epsilon: float) -> float:
    """Deduct epsilon from a user's budget.

    Args:
        user_id: Identifier of the requesting user.
        epsilon: Privacy cost of the query.

    Returns:
        The remaining budget after the deduction.

    Raises:
        DifferentialPrivacyError: If the cost is non-positive or exceeds
            the remaining budget.
    """
    if epsilon <= 0:
        raise DifferentialPrivacyError("Spent epsilon must be > 0")
    with self._lock:
        state = self._load()
        current = state.get(user_id, self.total_epsilon)
        if epsilon > current + EPSILON_TOLERANCE:
            raise DifferentialPrivacyError(
                f"Privacy budget exceeded: requested e={epsilon}, "
                f"remaining e={current}"
            )
        updated = current - epsilon
        # Snap a floating-point residue to zero. Without this a budget
        # spent down in fractional steps lands on something like 1e-16
        # instead of 0.0, and reads as "insufficient" forever rather than
        # "exhausted".
        if updated <= EPSILON_TOLERANCE:
            updated = 0.0
        state[user_id] = updated
        self._save(state)
        logger.info(
            f"Budget spent: user={user_id} e={epsilon} "
            f"remaining={state[user_id]}"
        )
        return state[user_id]
is_exhausted 
is_exhausted(user_id: str) -> bool

Whether a user has no usable budget left.

Uses the same tolerance as spending, so a budget drained by fractional steps is reported as exhausted rather than as an unusable sliver.

Source code in src/cryptography_manager/core/budget.py 
172
173
174
175
176
177
178
179
def is_exhausted(self, user_id: str) -> bool:
    """Whether a user has no usable budget left.

    Uses the same tolerance as spending, so a budget drained by
    fractional steps is reported as exhausted rather than as an
    unusable sliver.
    """
    return self.remaining(user_id) <= EPSILON_TOLERANCE
reset 
reset(user_id: str) -> None

Restore a user's full budget. Only valid for a new dataset.

Source code in src/cryptography_manager/core/budget.py 
181
182
183
184
185
186
def reset(self, user_id: str) -> None:
    """Restore a user's full budget. Only valid for a new dataset."""
    with self._lock:
        state = self._load()
        state[user_id] = self.total_epsilon
        self._save(state)
info 
info(user_id: str) -> dict[str, float | None]

Return a user's accounting state, for response metadata.

Source code in src/cryptography_manager/core/budget.py 
188
189
190
191
192
193
194
195
196
def info(self, user_id: str) -> dict[str, float | None]:
    """Return a user's accounting state, for response metadata."""
    remaining = self.remaining(user_id)
    return {
        "total_epsilon": self.total_epsilon,
        "remaining_epsilon": remaining,
        "spent_epsilon": self.total_epsilon - remaining,
        "delta": self.delta,
    }

InMemoryBudgetStore 

InMemoryBudgetStore(
    total_epsilon: float, delta: float | None = None
)

Bases: BudgetStore

Budget store backed by a process-local dictionary.

Budgets are lost on restart, and are not shared across workers. Suitable for tests, examples and single-process development only.

Source code in src/cryptography_manager/core/budget.py 
207
208
209
def __init__(self, total_epsilon: float, delta: float | None = None):
    super().__init__(total_epsilon, delta)
    self._state: dict[str, float] = {}
remaining 
remaining(user_id: str) -> float

Return the epsilon still available to a user.

Source code in src/cryptography_manager/core/budget.py 
128
129
130
131
def remaining(self, user_id: str) -> float:
    """Return the epsilon still available to a user."""
    with self._lock:
        return self._load().get(user_id, self.total_epsilon)
spend 
spend(user_id: str, epsilon: float) -> float

Deduct epsilon from a user's budget.

Parameters:

Name Type Description Default
user_id str

Identifier of the requesting user.

 required
epsilon float

Privacy cost of the query.

 required

Returns:

Type Description
float

The remaining budget after the deduction.

Raises:

Type Description
DifferentialPrivacyError

If the cost is non-positive or exceeds the remaining budget.
Source code in src/cryptography_manager/core/budget.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def spend(self, user_id: str, epsilon: float) -> float:
    """Deduct epsilon from a user's budget.

    Args:
        user_id: Identifier of the requesting user.
        epsilon: Privacy cost of the query.

    Returns:
        The remaining budget after the deduction.

    Raises:
        DifferentialPrivacyError: If the cost is non-positive or exceeds
            the remaining budget.
    """
    if epsilon <= 0:
        raise DifferentialPrivacyError("Spent epsilon must be > 0")
    with self._lock:
        state = self._load()
        current = state.get(user_id, self.total_epsilon)
        if epsilon > current + EPSILON_TOLERANCE:
            raise DifferentialPrivacyError(
                f"Privacy budget exceeded: requested e={epsilon}, "
                f"remaining e={current}"
            )
        updated = current - epsilon
        # Snap a floating-point residue to zero. Without this a budget
        # spent down in fractional steps lands on something like 1e-16
        # instead of 0.0, and reads as "insufficient" forever rather than
        # "exhausted".
        if updated <= EPSILON_TOLERANCE:
            updated = 0.0
        state[user_id] = updated
        self._save(state)
        logger.info(
            f"Budget spent: user={user_id} e={epsilon} "
            f"remaining={state[user_id]}"
        )
        return state[user_id]
is_exhausted 
is_exhausted(user_id: str) -> bool

Whether a user has no usable budget left.

Uses the same tolerance as spending, so a budget drained by fractional steps is reported as exhausted rather than as an unusable sliver.

Source code in src/cryptography_manager/core/budget.py 
172
173
174
175
176
177
178
179
def is_exhausted(self, user_id: str) -> bool:
    """Whether a user has no usable budget left.

    Uses the same tolerance as spending, so a budget drained by
    fractional steps is reported as exhausted rather than as an
    unusable sliver.
    """
    return self.remaining(user_id) <= EPSILON_TOLERANCE
reset 
reset(user_id: str) -> None

Restore a user's full budget. Only valid for a new dataset.

Source code in src/cryptography_manager/core/budget.py 
181
182
183
184
185
186
def reset(self, user_id: str) -> None:
    """Restore a user's full budget. Only valid for a new dataset."""
    with self._lock:
        state = self._load()
        state[user_id] = self.total_epsilon
        self._save(state)
info 
info(user_id: str) -> dict[str, float | None]

Return a user's accounting state, for response metadata.

Source code in src/cryptography_manager/core/budget.py 
188
189
190
191
192
193
194
195
196
def info(self, user_id: str) -> dict[str, float | None]:
    """Return a user's accounting state, for response metadata."""
    remaining = self.remaining(user_id)
    return {
        "total_epsilon": self.total_epsilon,
        "remaining_epsilon": remaining,
        "spent_epsilon": self.total_epsilon - remaining,
        "delta": self.delta,
    }

FileBudgetStore 

FileBudgetStore(
    total_epsilon: float,
    path: str | Path,
    delta: float | None = None,
)

Bases: BudgetStore

Budget store backed by a JSON file.

Survives restarts, which matters because a budget that resets when the container restarts is not a privacy guarantee. Writes are atomic via os.replace. This is still single-node: a multi-replica deployment needs a shared backend, and this class is the seam to swap it in at.

Source code in src/cryptography_manager/core/budget.py 
228
229
230
231
232
233
234
235
236
def __init__(
    self,
    total_epsilon: float,
    path: str | Path,
    delta: float | None = None,
):
    super().__init__(total_epsilon, delta)
    self.path = Path(path)
    self.path.parent.mkdir(parents=True, exist_ok=True)
remaining 
remaining(user_id: str) -> float

Return the epsilon still available to a user.

Source code in src/cryptography_manager/core/budget.py 
128
129
130
131
def remaining(self, user_id: str) -> float:
    """Return the epsilon still available to a user."""
    with self._lock:
        return self._load().get(user_id, self.total_epsilon)
spend 
spend(user_id: str, epsilon: float) -> float

Deduct epsilon from a user's budget.

Parameters:

Name Type Description Default
user_id str

Identifier of the requesting user.

 required
epsilon float

Privacy cost of the query.

 required

Returns:

Type Description
float

The remaining budget after the deduction.

Raises:

Type Description
DifferentialPrivacyError

If the cost is non-positive or exceeds the remaining budget.
Source code in src/cryptography_manager/core/budget.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def spend(self, user_id: str, epsilon: float) -> float:
    """Deduct epsilon from a user's budget.

    Args:
        user_id: Identifier of the requesting user.
        epsilon: Privacy cost of the query.

    Returns:
        The remaining budget after the deduction.

    Raises:
        DifferentialPrivacyError: If the cost is non-positive or exceeds
            the remaining budget.
    """
    if epsilon <= 0:
        raise DifferentialPrivacyError("Spent epsilon must be > 0")
    with self._lock:
        state = self._load()
        current = state.get(user_id, self.total_epsilon)
        if epsilon > current + EPSILON_TOLERANCE:
            raise DifferentialPrivacyError(
                f"Privacy budget exceeded: requested e={epsilon}, "
                f"remaining e={current}"
            )
        updated = current - epsilon
        # Snap a floating-point residue to zero. Without this a budget
        # spent down in fractional steps lands on something like 1e-16
        # instead of 0.0, and reads as "insufficient" forever rather than
        # "exhausted".
        if updated <= EPSILON_TOLERANCE:
            updated = 0.0
        state[user_id] = updated
        self._save(state)
        logger.info(
            f"Budget spent: user={user_id} e={epsilon} "
            f"remaining={state[user_id]}"
        )
        return state[user_id]
is_exhausted 
is_exhausted(user_id: str) -> bool

Whether a user has no usable budget left.

Uses the same tolerance as spending, so a budget drained by fractional steps is reported as exhausted rather than as an unusable sliver.

Source code in src/cryptography_manager/core/budget.py 
172
173
174
175
176
177
178
179
def is_exhausted(self, user_id: str) -> bool:
    """Whether a user has no usable budget left.

    Uses the same tolerance as spending, so a budget drained by
    fractional steps is reported as exhausted rather than as an
    unusable sliver.
    """
    return self.remaining(user_id) <= EPSILON_TOLERANCE
reset 
reset(user_id: str) -> None

Restore a user's full budget. Only valid for a new dataset.

Source code in src/cryptography_manager/core/budget.py 
181
182
183
184
185
186
def reset(self, user_id: str) -> None:
    """Restore a user's full budget. Only valid for a new dataset."""
    with self._lock:
        state = self._load()
        state[user_id] = self.total_epsilon
        self._save(state)
info 
info(user_id: str) -> dict[str, float | None]

Return a user's accounting state, for response metadata.

Source code in src/cryptography_manager/core/budget.py 
188
189
190
191
192
193
194
195
196
def info(self, user_id: str) -> dict[str, float | None]:
    """Return a user's accounting state, for response metadata."""
    remaining = self.remaining(user_id)
    return {
        "total_epsilon": self.total_epsilon,
        "remaining_epsilon": remaining,
        "spent_epsilon": self.total_epsilon - remaining,
        "delta": self.delta,
    }

Audit trail

audit

Audit record construction, sanitisation and delivery.

Every request produces exactly one audit record, whether it succeeded, was rejected before execution, or failed inside a backend -- a rejected request that leaves no trace is precisely the one an investigator will want later.

Two rules shape this module. Sensitive material never enters a record: not key material, not plaintext, not the dataset, not secret shares. And delivery never fails a request: an unreachable downstream service is logged and the record is still written locally, because losing the operation's result because the audit trail was briefly unavailable would be a worse outcome than a delayed record.

AuditLogger 

AuditLogger(
    log_path: str | Path = "./audit/crm-audit.jsonl",
    targets: dict[str, str] | None = None,
    timeout: float = 5.0,
)

Builds audit records, writes them locally and forwards them onward.

Initialise the logger.

Parameters:

Name Type Description Default
log_path str | Path

JSON-lines file receiving every record.

 './audit/crm-audit.jsonl'
targets dict[str, str] | None

Downstream services to forward to, keyed by name.

 None
timeout float

Per-request timeout when forwarding.

 5.0
Source code in src/cryptography_manager/core/audit.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def __init__(
    self,
    log_path: str | Path = "./audit/crm-audit.jsonl",
    targets: dict[str, str] | None = None,
    timeout: float = 5.0,
):
    """Initialise the logger.

    Args:
        log_path: JSON-lines file receiving every record.
        targets: Downstream services to forward to, keyed by name.
        timeout: Per-request timeout when forwarding.
    """
    self.log_path = Path(log_path)
    self.log_path.parent.mkdir(parents=True, exist_ok=True)
    self.targets = targets or {}
    self.timeout = timeout
    self._write_lock = threading.Lock()
    # Forwarding runs off the request path so a slow downstream service
    # cannot add its latency to every caller's response.
    self._pool = ThreadPoolExecutor(
        max_workers=4, thread_name_prefix="crm-audit"
    )
build 
build(
    *,
    status: str,
    user_id: str | None = None,
    operation: str | None = None,
    backend: str | None = None,
    execution_time: float = 0.0,
    parameters: dict[str, Any] | None = None,
    error_message: str | None = None,
) -> AuditRecord

Construct a sanitised audit record.

Parameters:

Name Type Description Default
status str

Workflow outcome.

 required
user_id str | None

Identifier of the requesting user, if known.

 None
operation str | None

Requested cryptographic operation.

 None
backend str | None

Library used.

 None
execution_time float

Duration of execution, in seconds.

 0.0
parameters dict[str, Any] | None

Raw parameters; sanitised before being stored.

 None
error_message str | None

Failure detail, if any.

 None

Returns:

Type Description
AuditRecord

The record, ready to be emitted.
Source code in src/cryptography_manager/core/audit.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def build(
    self,
    *,
    status: str,
    user_id: str | None = None,
    operation: str | None = None,
    backend: str | None = None,
    execution_time: float = 0.0,
    parameters: dict[str, Any] | None = None,
    error_message: str | None = None,
) -> AuditRecord:
    """Construct a sanitised audit record.

    Args:
        status: Workflow outcome.
        user_id: Identifier of the requesting user, if known.
        operation: Requested cryptographic operation.
        backend: Library used.
        execution_time: Duration of execution, in seconds.
        parameters: Raw parameters; sanitised before being stored.
        error_message: Failure detail, if any.

    Returns:
        The record, ready to be emitted.
    """
    return AuditRecord(
        user_id=user_id,
        operation=operation,
        backend=backend,
        status=status,
        execution_time=round(execution_time, 6),
        parameters=sanitize(parameters or {}),
        error_message=error_message,
    )
emit 
emit(record: AuditRecord) -> AuditRecord

Persist a record locally and schedule it for forwarding.

Parameters:

Name Type Description Default
record AuditRecord

The record to emit.

 required

Returns:

Type Description
AuditRecord

The same record, for convenience.
Source code in src/cryptography_manager/core/audit.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
def emit(self, record: AuditRecord) -> AuditRecord:
    """Persist a record locally and schedule it for forwarding.

    Args:
        record: The record to emit.

    Returns:
        The same record, for convenience.
    """
    self._write_local(record)
    if self.targets:
        self._pool.submit(self._forward, record)
    return record
record 
record(**kwargs: Any) -> AuditRecord

Build and emit a record in one call.

Source code in src/cryptography_manager/core/audit.py 
190
191
192
def record(self, **kwargs: Any) -> AuditRecord:
    """Build and emit a record in one call."""
    return self.emit(self.build(**kwargs))
shutdown 
shutdown() -> None

Wait for in-flight forwards to finish.

Source code in src/cryptography_manager/core/audit.py 
229
230
231
def shutdown(self) -> None:
    """Wait for in-flight forwards to finish."""
    self._pool.shutdown(wait=True)

sanitize 

sanitize(value: Any, _depth: int = 0) -> Any

Strip sensitive material from a parameter structure.

Parameters:

Name Type Description Default
value Any

Arbitrary parameter data.

 required
_depth int

Internal recursion guard.

 0

Returns:

Type Description
Any

A copy safe to persist and forward.
Source code in src/cryptography_manager/core/audit.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def sanitize(value: Any, _depth: int = 0) -> Any:
    """Strip sensitive material from a parameter structure.

    Args:
        value: Arbitrary parameter data.
        _depth: Internal recursion guard.

    Returns:
        A copy safe to persist and forward.
    """
    if _depth >= MAX_DEPTH:
        return "[TRUNCATED]"

    if isinstance(value, dict):
        clean: dict[str, Any] = {}
        for key, item in value.items():
            name = str(key)
            if _is_sensitive(name):
                clean[name] = REDACTED
            elif name.lower() in BULK_DATA_KEYS:
                clean[name] = _summarise(item)
            else:
                clean[name] = sanitize(item, _depth + 1)
        return clean

    if isinstance(value, (list, tuple)):
        # A bare sequence of values is data; record its shape, not its
        # contents, which could be the dataset itself.
        if len(value) > 10:
            return _summarise(value)
        return [sanitize(v, _depth + 1) for v in value]

    if isinstance(value, (str, int, float, bool)) or value is None:
        return value

    return str(type(value).__name__)

Workflow status

status

Workflow status codes and their HTTP mapping.

The CrM distinguishes between the workflow status -- the outcome of the internal auth -> pre-check -> validate -> execute -> audit pipeline -- and the HTTP status eventually returned to the caller. Keeping them separate matters because every workflow outcome, including the rejected ones, still produces an audit record.

WorkflowStatus

Bases: str, Enum

Outcome of a cryptographic workflow.

NOT_IMPLEMENTED covers operations that are accepted by the schema but have no adapter yet, so a caller can tell "you asked for something that does not exist" apart from "you asked for something not built yet".

is_success property 
is_success: bool

Whether this status represents a completed operation.

http_status_for 

http_status_for(status: WorkflowStatus) -> int

Map a workflow status onto its HTTP response code.

Parameters:

Name Type Description Default
status WorkflowStatus

The workflow outcome to translate.

 required

Returns:

Type Description
int

The HTTP status code to return to the caller.
Source code in src/cryptography_manager/core/status.py 
60
61
62
63
64
65
66
67
68
69
def http_status_for(status: WorkflowStatus) -> int:
    """Map a workflow status onto its HTTP response code.

    Args:
        status: The workflow outcome to translate.

    Returns:
        The HTTP status code to return to the caller.
    """
    return _HTTP_STATUS[status]