Service layer

The HTTP surface: the request and response envelope, bearer token verification, the single operation endpoint, and the ASGI application itself.

Request and response models

models

Pydantic data models for the Cryptography Manager API.

These models are the single source of truth for the request envelope, the response envelope and the audit record. They deliberately avoid importing any cryptographic backend so that the schema stays importable even when optional extras such as PyDP, Pyfhel or MPyC are not installed.

Operation

Bases: str, Enum

Cryptographic operation requested by the caller.

Backend

Bases: str, Enum

Cryptographic library used to service an operation.

OutputFormat

Bases: str, Enum

Desired response encoding.

JSON-LD is a plausible future addition, but its vocabulary would have to align with the Compliance Manager's DPV graphs, which are not pinned down yet. Only JSON is accepted for now; requesting anything else is a validation error rather than a silently ignored field.

DPQueryType

Bases: str, Enum

Aggregate queries supported by the differential privacy adapter.

Mirrors the PyDP Laplacian algorithms wired up in the DP adapter. Declared here rather than imported from the adapter so that importing the schema never requires PyDP to be installed.

CryptographyRequest

Bases: BaseModel

Request envelope for POST /cryptography.

The Authorization header is not modelled here; it is handled by the auth dependency before the body is ever parsed.

check_backend_compatible

check_backend_compatible() -> str | None

Verify the requested backend can service the operation.

Returns:

Type	Description
str | None	An error message if the pairing is invalid, otherwise None.

Source code in src/cryptography_manager/api/models.py

def check_backend_compatible(self) -> str | None:
    """Verify the requested backend can service the operation.

    Returns:
        An error message if the pairing is invalid, otherwise ``None``.
    """
    allowed = COMPATIBLE_BACKENDS[self.operation]
    if self.backend in allowed:
        return None
    names = ", ".join(sorted(b.value for b in allowed))
    return (
        f"Backend '{self.backend.value}' cannot service operation "
        f"'{self.operation.value}'; supported: {names}"
    )

DPParameters

Bases: BaseModel

Differential privacy parameters carried in parameters.

Validated separately from the envelope because the required fields depend on the operation. epsilon_cost is the privacy price of the query and is checked against the caller's remaining budget before execution.

CryptographyResponse

Bases: BaseModel

Response envelope for POST /cryptography.

AuditRecord

Bases: BaseModel

Structured audit entry produced for every request.

Sensitive material -- secret and private keys, plaintext values, secret shares and intermediate cryptographic state -- must never reach this model. Sanitisation happens in core.audit before construction.

Authentication

auth

Bearer token verification.

Verification sits behind a small interface with two implementations, because the identity provider this component will ultimately talk to is not settled. The OIDC verifier targets the standard discovery and JWKS endpoints rather than any one vendor's API, so it should work against any conforming provider and needs replacing only if the eventual choice is non-standard.

The development verifier exists so the service is runnable with no identity provider at all. It still rejects a missing or malformed token, so the rejection path is exercised the same way in both modes -- it simply does not verify signatures, and must never be used outside development.

AuthResult

AuthResult(
    user_id: str | None = None,
    claims: dict[str, Any] | None = None,
    error: str | None = None,
)

Outcome of verifying a token.

Source code in src/cryptography_manager/api/auth.py

def __init__(
    self,
    user_id: str | None = None,
    claims: dict[str, Any] | None = None,
    error: str | None = None,
) -> None:
    self.user_id = user_id
    self.claims = claims or {}
    self.error = error

ok property

ok: bool

Whether verification succeeded.

TokenVerifier

Bases: Protocol

Verifies a bearer token and identifies the caller.

verify

verify(token: str | None) -> AuthResult

Verify a token.

Parameters:

Name	Type	Description	Default
token	str | None	The bearer token, without its scheme prefix.	required

Returns:

Type	Description
AuthResult	The verification outcome. Failures are returned rather than
AuthResult	raised, because a rejected request still has to be audited.

Source code in src/cryptography_manager/api/auth.py

def verify(self, token: str | None) -> AuthResult:
    """Verify a token.

    Args:
        token: The bearer token, without its scheme prefix.

    Returns:
        The verification outcome. Failures are returned rather than
        raised, because a rejected request still has to be audited.
    """
    ...

DevTokenVerifier

Accepts any well-formed token without verifying its signature.

Intended solely for development against no identity provider. It reads the claims of a JWT to identify the caller, and otherwise treats the token string itself as the identity.

verify

verify(token: str | None) -> AuthResult

Identify the caller without cryptographic verification.

Source code in src/cryptography_manager/api/auth.py

def verify(self, token: str | None) -> AuthResult:
    """Identify the caller without cryptographic verification."""
    if not token or not token.strip():
        return AuthResult(error="Missing authorization token")

    token = token.strip()
    try:
        claims = jwt.decode(
            token, options={"verify_signature": False, "verify_exp": False}
        )
    except Exception:
        # Not a JWT. Derive a stable identity from a digest of the token
        # rather than the token itself: the identifier reaches audit
        # records and budget state, and neither may ever hold a
        # credential that could be replayed.
        digest = hashlib.sha256(token.encode("utf-8")).hexdigest()
        return AuthResult(user_id=f"dev:{digest[:16]}")

    user_id = _extract_user_id(claims)
    if not user_id:
        return AuthResult(error="Token contains no usable subject claim")
    return AuthResult(user_id=user_id, claims=claims)

OIDCTokenVerifier

OIDCTokenVerifier(settings: Settings)

Verifies tokens against an OpenID Connect provider.

Keys are discovered through the provider's well-known configuration and cached by the JWKS client, so steady-state verification is local and does not add a network round trip per request.

Initialise the verifier.

Parameters:

Name	Type	Description	Default
settings	Settings	Runtime settings carrying the issuer and client details.	required

Raises:

Type	Description
ValueError	If no issuer is configured.

Source code in src/cryptography_manager/api/auth.py

def __init__(self, settings: Settings) -> None:
    """Initialise the verifier.

    Args:
        settings: Runtime settings carrying the issuer and client details.

    Raises:
        ValueError: If no issuer is configured.
    """
    issuer = settings.issuer_url
    if not issuer:
        raise ValueError("OIDC verification requires an issuer URL")
    self.issuer = issuer
    self.settings = settings
    self._jwks_client: PyJWKClient | None = None
    self._discovered: dict[str, Any] | None = None

verify

verify(token: str | None) -> AuthResult

Verify a token's signature, expiry and issuer.

Source code in src/cryptography_manager/api/auth.py

def verify(self, token: str | None) -> AuthResult:
    """Verify a token's signature, expiry and issuer."""
    if not token or not token.strip():
        return AuthResult(error="Missing authorization token")

    token = token.strip()
    try:
        signing_key = self._client().get_signing_key_from_jwt(token)
        audience = (
            self.settings.oidc_client_id
            if self.settings.oidc_verify_audience
            else None
        )
        claims = jwt.decode(
            token,
            signing_key.key,
            algorithms=ALLOWED_ALGORITHMS,
            issuer=self.issuer,
            audience=audience,
            options={
                "verify_signature": True,
                "verify_exp": True,
                "verify_iss": True,
                "verify_aud": bool(audience),
                "require": ["exp"],
            },
        )
    except jwt.ExpiredSignatureError:
        return AuthResult(error="Token has expired")
    except jwt.InvalidTokenError as exc:
        return AuthResult(error=f"Invalid token: {exc}")
    except Exception as exc:
        # A provider that is unreachable must not be reported as a valid
        # token; failing closed is the only safe direction here.
        logger.warning(f"Token verification could not complete: {exc}")
        return AuthResult(error="Token verification unavailable")

    user_id = _extract_user_id(claims)
    if not user_id:
        return AuthResult(error="Token contains no usable subject claim")
    return AuthResult(user_id=user_id, claims=claims)

build_verifier

build_verifier(settings: Settings) -> TokenVerifier

Construct the verifier selected by configuration.

Parameters:

Name	Type	Description	Default
settings	Settings	Runtime settings.	required

Returns:

Type	Description
TokenVerifier	The configured verifier.

Source code in src/cryptography_manager/api/auth.py

def build_verifier(settings: Settings) -> TokenVerifier:
    """Construct the verifier selected by configuration.

    Args:
        settings: Runtime settings.

    Returns:
        The configured verifier.
    """
    if settings.auth_mode == "oidc":
        logger.info(f"Auth mode: OIDC against {settings.issuer_url}")
        return OIDCTokenVerifier(settings)
    logger.warning(
        "Auth mode: development. Tokens are NOT verified -- "
        "set CRM_AUTH_MODE=oidc before any real deployment."
    )
    return DevTokenVerifier()

extract_bearer

extract_bearer(header: str | None) -> str | None

Pull the token out of an Authorization header.

Parameters:

Name	Type	Description	Default
header	str | None	Raw header value, if present.	required

Returns:

Type	Description
str | None	The token, or None if the header is absent or not a bearer token.

Source code in src/cryptography_manager/api/auth.py

def extract_bearer(header: str | None) -> str | None:
    """Pull the token out of an Authorization header.

    Args:
        header: Raw header value, if present.

    Returns:
        The token, or ``None`` if the header is absent or not a bearer token.
    """
    if not header:
        return None
    parts = header.split(None, 1)
    if len(parts) != 2 or parts[0].lower() != "bearer":
        return None
    return parts[1].strip() or None

Routes

routes

HTTP routes.

The request body is read and validated inside the handler rather than through a typed signature. That is deliberate: the framework would otherwise consume and parse the body before the handler runs, which would bypass the audit trail, answer an unauthenticated caller with a schema complaint instead of a refusal, and buffer an arbitrarily large payload before anyone had a chance to object. Authorization is decided first, the body is then read under a size ceiling, and every rejection still produces an audit record.

health async

health(request: Request) -> dict[str, Any]

Report service liveness and which capabilities are available.

Source code in src/cryptography_manager/api/routes.py

@router.get("/health", tags=["service"])
async def health(request: Request) -> dict[str, Any]:
    """Report service liveness and which capabilities are available."""
    state = request.app.state
    return {
        "status": "ok",
        "auth_mode": state.settings.auth_mode,
        "operations": {
            "differential_privacy": "available",
            "encryption": "available",
            "key_management": "available",
            "homomorphic_encryption": "not_implemented",
            "smpc": "not_implemented",
        },
    }

cryptography async

cryptography(
    request: Request,
    authorization: str | None = Header(default=None),
) -> JSONResponse

Execute one cryptographic operation.

Parameters:

Name	Type	Description	Default
request	Request	The incoming request, carrying application state.	required
authorization	str | None	Bearer token issued by the identity provider.	Header(default=None)

Returns:

Type	Description
JSONResponse	The response envelope, with a status code reflecting the outcome.

Source code in src/cryptography_manager/api/routes.py

@router.post(
    "/cryptography",
    tags=["cryptography"],
    summary="Execute a cryptographic operation",
    response_model=CryptographyResponse,
    responses={
        200: {"description": "Operation completed successfully"},
        400: {"description": "Configuration or parameter validation failed"},
        403: {"description": "Authorization token missing or invalid"},
        429: {"description": "Privacy budget exhausted"},
        500: {"description": "Unexpected execution error"},
        413: {"description": "Request body exceeds the configured limit"},
        501: {"description": "Operation recognised but not implemented"},
    },
    openapi_extra={
        "requestBody": {
            "required": True,
            "content": {
                "application/json": {
                    "schema": CryptographyRequest.model_json_schema()
                }
            },
        }
    },
)
async def cryptography(
    request: Request,
    authorization: str | None = Header(default=None),
) -> JSONResponse:
    """Execute one cryptographic operation.

    Args:
        request: The incoming request, carrying application state.
        authorization: Bearer token issued by the identity provider.

    Returns:
        The response envelope, with a status code reflecting the outcome.
    """
    state = request.app.state
    workflow = state.workflow

    auth = state.verifier.verify(extract_bearer(authorization))
    if not auth.ok:
        outcome = workflow.run(None, None, auth.error)
        return _respond(outcome.status, outcome.response)

    limit = state.settings.max_request_bytes
    body, oversized = await _read_capped(request, limit)
    if oversized:
        outcome = workflow.reject(
            WorkflowStatus.PAYLOAD_TOO_LARGE,
            auth.user_id,
            f"Request body exceeds the {limit} byte limit",
        )
        return _respond(outcome.status, outcome.response)

    try:
        payload = json.loads(body) if body.strip() else {}
    except json.JSONDecodeError as exc:
        outcome = workflow.reject(
            WorkflowStatus.VALIDATION_ERROR,
            auth.user_id,
            f"Request body is not valid JSON: {exc}",
        )
        return _respond(outcome.status, outcome.response)

    if not isinstance(payload, dict):
        outcome = workflow.reject(
            WorkflowStatus.VALIDATION_ERROR,
            auth.user_id,
            "Request body must be a JSON object",
        )
        return _respond(outcome.status, outcome.response)

    try:
        parsed = CryptographyRequest.model_validate(payload)
    except ValidationError as exc:
        outcome = workflow.reject(
            WorkflowStatus.VALIDATION_ERROR, auth.user_id, _render(exc)
        )
        return _respond(outcome.status, outcome.response)

    outcome = workflow.run(parsed, auth.user_id)
    return _respond(outcome.status, outcome.response)

Application entry point

main

Service entry point.

Builds the FastAPI application and wires together the pieces a request needs: configuration, the manager that dispatches to backends, per-user budget accounting, audit logging and token verification. Everything is assembled once at startup and shared, because rebuilding adapters or re-reading budget state per request would be both slow and, in the budget's case, wrong.

create_app

create_app(settings: Settings | None = None) -> FastAPI

Build the application.

Parameters:

Name	Type	Description	Default
settings	Settings | None	Runtime settings; read from the environment when omitted.	None

Returns:

Type	Description
FastAPI	The configured application.

Source code in src/cryptography_manager/main.py

def create_app(settings: Settings | None = None) -> FastAPI:
    """Build the application.

    Args:
        settings: Runtime settings; read from the environment when omitted.

    Returns:
        The configured application.
    """
    settings = settings or get_settings()
    _configure_logging(settings.log_level)

    config = Config()
    config_path = Path(settings.config_path)
    if config_path.exists():
        config.load_from_file(config_path)
        logger.info(f"Loaded configuration from {config_path}")
    else:
        logger.warning(
            f"No configuration file at {config_path}; using defaults"
        )

    manager = CryptographyManager(config=config)
    audit = AuditLogger(
        log_path=settings.audit_log_path,
        targets=settings.audit_targets,
        timeout=settings.forward_timeout_seconds,
    )
    budget_store = _build_budget_store(settings, config)

    @asynccontextmanager
    async def lifespan(_: FastAPI) -> AsyncIterator[None]:
        """Let in-flight audit forwards finish before the process exits."""
        yield
        audit.shutdown()

    app = FastAPI(
        title=TITLE,
        version=VERSION,
        description=DESCRIPTION,
        lifespan=lifespan,
    )

    app.state.settings = settings
    app.state.config = config
    app.state.manager = manager
    app.state.audit = audit
    app.state.budget_store = budget_store
    app.state.verifier = build_verifier(settings)
    app.state.workflow = CryptographyWorkflow(manager, audit, budget_store)

    app.include_router(router)

    targets = ", ".join(settings.audit_targets) or "local file only"
    logger.info(f"{TITLE} ready; audit targets: {targets}")
    return app

main

main() -> None

Console script entry point: serve the API with uvicorn.

Source code in src/cryptography_manager/main.py

def main() -> None:
    """Console script entry point: serve the API with uvicorn."""
    import uvicorn

    settings = get_settings()
    uvicorn.run(
        "cryptography_manager.main:app",
        host="0.0.0.0",
        port=8000,
        log_level=settings.log_level.lower(),
    )