spiffe-defakto

Stable — 1.x. The public API follows Semantic Versioning: no breaking changes within a major version. See the changelog before upgrading.

A Python SDK for SPIFFE workload identity. It gives your Python workloads cryptographic identities that can be used for mTLS, JWT auth, and federated AWS / Azure / GCP credentials — with no long-lived secrets.

Quick start

pip install spiffe-defakto

# get_svid.py
import asyncio
from spiffe_defakto import WorkloadAPIClient

async def main():
    async with WorkloadAPIClient() as client:
        svid = await client.x509.get_svid()
        print("SPIFFE ID:", svid.id)
        print("Expires at:", svid.expires_at.isoformat())

asyncio.run(main())

Point SPIFFE_ENDPOINT_SOCKET at your Defakto agent (export SPIFFE_ENDPOINT_SOCKET=unix:///run/spirl/socket/agent.sock) and run it. That's the whole happy path.

Not using asyncio? Use the sync companion:

from spiffe_defakto.sync import SyncWorkloadAPIClient

with SyncWorkloadAPIClient() as client:
    svid = client.get_svid()
    print("SPIFFE ID:", svid.id)

Using AWS/Azure/GCP? Install the relevant extra and drop a SPIFFE-backed credential into any SDK client:

# pip install 'spiffe-defakto[aws]'
import boto3, botocore.session
from spiffe_defakto.aws import FromSpiffeOptions, from_spiffe

credentials = from_spiffe(FromSpiffeOptions(role_arn="arn:aws:iam::123:role/my-role"))
session = botocore.session.get_session()
session._credentials = credentials
s3 = boto3.Session(botocore_session=session).client("s3")

The rest of this README is reference material — read on when you want to know about serverless attestation, custom attestors, JWT SVIDs, or rotation streaming.

Framework guides

Per-framework integration guides with runnable examples:

See docs/frameworks/README.md for the full compatibility matrix and guidance on which client to use.

Installation

# pip
pip install spiffe-defakto
# uv
uv add spiffe-defakto
# poetry
poetry add spiffe-defakto
# pdm
pdm add spiffe-defakto

With optional cloud credential providers

# pip
pip install 'spiffe-defakto[aws]'
pip install 'spiffe-defakto[azure]'
pip install 'spiffe-defakto[gcp]'
pip install 'spiffe-defakto[all]'

# uv
uv add 'spiffe-defakto[aws]'
# poetry
poetry add 'spiffe-defakto[aws]'
# pdm
pdm add 'spiffe-defakto[aws]'

Python 3.10+. Works on Linux and macOS; TCP transport is available everywhere.

Clients

The SDK ships three clients that share the same surface area:

• WorkloadAPIClient — the default. Auto-selects the right underlying client based on environment variables. Works unchanged everywhere.
• LocalWorkloadAPIClient — connects to a local SPIFFE agent (Defakto or compatible) over a Unix socket or TCP. Use this in traditional server and container deployments.
• AttestingWorkloadAPIClient — attests the workload at request time and talks directly to Defakto's cloud endpoint. Use this in serverless or ephemeral environments (AWS Lambda, Cloud Run, Azure Container Apps, GitHub Actions, etc.) where no local agent is practical.

A SyncWorkloadAPIClient blocking facade mirrors the same API for scripts and non-async frameworks.

Smart wrapper (WorkloadAPIClient)

WorkloadAPIClient auto-selects the right underlying client. Use it as your default import — switch environments by flipping environment variables rather than application code.

Selection logic (priority order)

1. SPIFFE_ENDPOINT_SOCKET env var set → LocalWorkloadAPIClient.
2. DEFAKTO_ATTESTORS env var set → AttestingWorkloadAPIClient. The endpoint is resolved from DEFAKTO_SERVER_ADDRESS (required for self-hosted trust domains), falling back to DEFAKTO_TRUST_DOMAIN_ID (Defakto-hosted only). See endpoint resolution below.
3. Otherwise → raises SpiffeError with code SOCKET_PATH_NOT_CONFIGURED.

from spiffe_defakto import WorkloadAPIClient

async with WorkloadAPIClient() as client:
    svid = await client.x509.get_svid()
    print("SPIFFE ID:", svid.id)

Environment-driven attestation (DEFAKTO_ATTESTORS)

Set DEFAKTO_ATTESTORS to a comma-separated list of built-in attestor names to auto-instantiate them with default options. Pair with DEFAKTO_SERVER_ADDRESS (self-hosted trust domains) or DEFAKTO_TRUST_DOMAIN_ID (Defakto-hosted trust domains) to resolve the endpoint (see endpoint resolution below).

DEFAKTO_ATTESTORS=aws_external_identity

# Self-hosted trust domain — always set DEFAKTO_SERVER_ADDRESS:
DEFAKTO_SERVER_ADDRESS=my-server.example.com:443

# Defakto-hosted trust domain — set DEFAKTO_TRUST_DOMAIN_ID instead:
# DEFAKTO_TRUST_DOMAIN_ID=td-0000000

# Application code stays the same in every environment.
async with WorkloadAPIClient() as client:
    svid = await client.x509.get_svid()

Unrecognised names raise SpiffeError with code NO_ATTESTORS_CONFIGURED.

Standard SPIFFE (LocalWorkloadAPIClient)

LocalWorkloadAPIClient speaks the standard SPIFFE Workload API gRPC protocol. It works with any compliant SPIFFE agent, including Defakto.

Connection options

The client resolves its endpoint in the following order:

1. Explicit options passed to the constructor.
2. SPIFFE_ENDPOINT_SOCKET environment variable (supports unix:///path, unix://path, and unix:path prefixes).

If neither is provided the client raises SpiffeError with code SOCKET_PATH_NOT_CONFIGURED.

from spiffe_defakto import LocalWorkloadAPIClient, TcpOptions, UnixOptions

# Resolved from SPIFFE_ENDPOINT_SOCKET (raises if not set)
async with LocalWorkloadAPIClient() as client:
    ...

# Explicit Unix socket
async with LocalWorkloadAPIClient(UnixOptions(socket_path="/run/spirl/socket/agent.sock")) as client:
    ...

# Explicit TCP (e.g. remote Defakto agent)
async with LocalWorkloadAPIClient(
    TcpOptions(address="defakto-agent.internal", port=8081, insecure_skip_tls=False)
) as client:
    ...

X.509 SVID

svid = await client.x509.get_svid()

print("SPIFFE ID:", svid.id)                # spiffe://example.org/service
print("Expires at:", svid.expires_at.isoformat())
# svid.certificates — tuple of `cryptography.x509.Certificate`
# svid.private_key   — cryptography private key (RSA or EC)

JWT SVID

jwt = await client.jwt.fetch_svid(["https://api.example.org"])
print("SPIFFE ID:", jwt.id)
print("Token:", jwt.token)

Watching X.509 context rotations

The Workload API pushes a new context every time SVIDs are rotated. watch_x509_context() returns an async iterator that reconnects automatically when the stream ends cleanly.

# Full context (all SVIDs + trust bundles)
async for ctx in client.watch_x509_context():
    svid = ctx.default_svid()
    print("Rotated SVID:", svid.id)

    # ctx.svids   — every SVID issued to this workload
    # ctx.bundles — {trust_domain_name: X509Bundle}

# Or watch just the default SVID
async for svid in client.x509.watch_svid():
    print("New SVID expires:", svid.expires_at.isoformat())

Resource cleanup

All clients implement __aenter__/__aexit__, and expose an explicit close() coroutine.

# Recommended: automatic cleanup via `async with`
async with LocalWorkloadAPIClient() as client:
    svid = await client.x509.get_svid()
# client.close() is called automatically when the block exits

# Manual cleanup
client = LocalWorkloadAPIClient()
try:
    svid = await client.x509.get_svid()
finally:
    await client.close()

Sync facade (SyncWorkloadAPIClient)

Scripts and non-async frameworks can use the blocking companion. It drives the async client on a private background loop — streaming becomes a regular iterator:

from spiffe_defakto.sync import SyncWorkloadAPIClient

with SyncWorkloadAPIClient() as client:
    svid = client.get_svid()
    for ctx in client.watch_x509_context():
        ...

Prefer the async WorkloadAPIClient when you are already inside an event loop — the sync facade is a convenience, not a default.

Defakto Serverless (AttestingWorkloadAPIClient)

AttestingWorkloadAPIClient targets environments where a SPIFFE agent cannot run alongside the workload (AWS Lambda, Google Cloud Run, Azure Container Apps, GitHub Actions, etc.). Instead of relying on a pre-existing agent, it attests the workload inline — collecting cryptographic evidence from the environment on every request — and sends that evidence to Defakto's trust domain server endpoint to obtain SVIDs.

Endpoint resolution

The endpoint is resolved in the following priority order:

1. Explicit transport option (same shape as LocalWorkloadAPIClient) — overrides all environment-based resolution.
2. DEFAKTO_SERVER_ADDRESS environment variable — accepts host:port or a bare host (defaults to port 443). Always set this when your trust domain is self-hosted.
3. trust_domain_id option or DEFAKTO_TRUST_DOMAIN_ID environment variable — only applicable when the trust domain is hosted by Defakto. Automatically resolves to <trust_domain_id>.agent.spirl.com:443 over TLS.

If none of these are configured, AttestingWorkloadAPIClient raises SpiffeError with code SERVER_ADDRESS_NOT_CONFIGURED.

X.509 SVID

from spiffe_defakto import AttestingClientOptions, AttestingWorkloadAPIClient, AwsTokenAttestor

async with AttestingWorkloadAPIClient(
    AttestingClientOptions(
        trust_domain_id="td-0000000",
        attestors=(AwsTokenAttestor(),),
    )
) as client:
    svid = await client.x509.get_svid()

JWT SVID

jwt = await client.jwt.fetch_svid(["https://api.example.org"])
print("ID:", jwt.id)

Watching X.509 context rotations

The attesting client re-attests and re-fetches the SVID automatically before it expires (at 85 % of the remaining lifetime, with a 1-second floor).

async for ctx in client.watch_x509_context():
    svid = ctx.default_svid()
    print("Renewed SVID expires:", svid.expires_at.isoformat())

Token exchange (TokenExchangeClient)

TokenExchangeClient exchanges a user OIDC ID token for a Defakto JWT-SVID. Each exchange fetches a fresh agent JWT-SVID and sends it as the RFC 8693 client_assertion; the returned JWT-SVID is verified with the workload JWT bundle source before it is returned. Exchanged tokens are not cached by the SDK. The user token can come from any OIDC identity provider trusted by the token exchange endpoint. The client_assertion_audience argument is the token exchange service's OIDC issuer / authorization-server identifier; the SDK fetches the agent JWT-SVID with exactly that audience.

from spiffe_defakto.token_exchange import TokenExchangeClient

token_exchange = TokenExchangeClient(
    "https://tds.example.org/oauth/token",
    "https://tds.example.org",
)

svid = await token_exchange.exchange(user_id_token, "spiffe://example.org/recipient")

The returned value is a verified JwtSVID whose audience matches the target audience passed to exchange. Services that receive these JWT-SVIDs must still verify them against their normal JWT bundle policy. Exchange/network/protocol failures raise SpiffeError with code TOKEN_EXCHANGE_FAILED. Advanced integrations can pass source= when they need to provide their own JWT-SVID and JWT bundle source. Custom sources must fetch agent JWT-SVIDs accepted by the endpoint as jwt-spiffe client assertions, including exp and jti, and must resolve JWT bundles used to verify returned JWT-SVIDs. Pass http_client= to reuse or inject a custom httpx.AsyncClient.

A service that already holds a delegated JWT-SVID can exchange it for one targeting a new audience with re_exchange. The original user is carried through unchanged and the calling workload is stamped as the actor:

svid = await token_exchange.re_exchange(delegated_svid.token, "spiffe://example.org/next")

Built-in attestors

AWS (AwsTokenAttestor)

Attests using a signed JWT from AWS STS (GetWebIdentityToken). Works in any environment with an IAM role attached — Lambda, ECS, EC2, EKS (via IRSA), and so on.

from spiffe_defakto import AttestingClientOptions, AttestingWorkloadAPIClient, AwsTokenAttestor

async with AttestingWorkloadAPIClient(
    AttestingClientOptions(
        trust_domain_id="td-0000000",
        attestors=(AwsTokenAttestor(),),
    )
) as client:
    svid = await client.x509.get_svid()

Options

from spiffe_defakto import AwsTokenAttestor, AwsTokenAttestorOptions

attestor = AwsTokenAttestor(
    AwsTokenAttestorOptions(
        audience=("my-service",),
        signing_algorithm="ES384",
        duration_seconds=300,
        sts_config={"region_name": "eu-west-1"},
    )
)

Azure (AzureMSIAttestor)

Attests using an Azure Managed Identity token obtained from the Azure Instance Metadata Service. Works on any Azure compute resource with a managed identity — App Service, Container Apps, AKS, VM, and so on.

from spiffe_defakto import AttestingClientOptions, AttestingWorkloadAPIClient, AzureMSIAttestor

async with AttestingWorkloadAPIClient(
    AttestingClientOptions(
        trust_domain_id="td-0000000",
        attestors=(AzureMSIAttestor(),),
    )
) as client:
    svid = await client.x509.get_svid()

By default the attestor uses api://AzureADTokenExchange as the audience. If the managed identity is user-assigned, select it by client_id, principal_id, or resource_id:

from spiffe_defakto import AzureMSIAttestor, AzureMSIAttestorOptions

# System-assigned identity (default)
AzureMSIAttestor()

# User-assigned identity — select by client ID
AzureMSIAttestor(AzureMSIAttestorOptions(client_id="00000000-0000-0000-0000-000000000000"))

# User-assigned identity — select by object/principal ID
AzureMSIAttestor(AzureMSIAttestorOptions(principal_id="00000000-0000-0000-0000-000000000000"))

# Custom audience
AzureMSIAttestor(AzureMSIAttestorOptions(audience="api://my-tenant-id/defakto-server"))

At most one of client_id, principal_id, resource_id may be set.

GCP (GcpIITAttestor)

Attests using a GCP Instance Identity Token (IIT) fetched from the metadata service. The token is a signed JWT issued by Google that proves the workload is running on a specific GCE instance, Cloud Run service, GKE pod, and so on.

from spiffe_defakto import AttestingClientOptions, AttestingWorkloadAPIClient, GcpIITAttestor

async with AttestingWorkloadAPIClient(
    AttestingClientOptions(
        trust_domain_id="td-0000000",
        attestors=(GcpIITAttestor(),),
    )
) as client:
    svid = await client.x509.get_svid()

Options

from spiffe_defakto import GcpIITAttestor, GcpIITAttestorOptions

GcpIITAttestor(
    GcpIITAttestorOptions(
        audience="my-defakto-agent",
        service_account="my-sa@my-project.iam.gserviceaccount.com",
    )
)

GCP Vertex AI Agent Engine (GcpAgentIdentityAttestor)

Attests using SPIFFE credentials provisioned on disk by the Vertex AI Agent Engine runtime. The attestor reads the X.509 SVID (cert chain + EC P-256 private key) from the standard workload credential paths and mints a self-signed, X.509-bound proof-of-possession JWT signed by the leaf certificate (ES256). The JWT carries the certificate chain in the JOSE x5c header, proving possession of the leaf cert's private key. This attestation token is not a JWT-SVID — JWT-SVIDs are signed by the trust-domain authority, not by the workload. No optional dependency is required.

from spiffe_defakto import (
    AttestingClientOptions,
    AttestingWorkloadAPIClient,
    GcpAgentIdentityAttestor,
)

async with AttestingWorkloadAPIClient(
    AttestingClientOptions(
        trust_domain_id="td-0000000",
        attestors=(GcpAgentIdentityAttestor(),),
    )
) as client:
    svid = await client.x509.get_svid()

Options

from spiffe_defakto import GcpAgentIdentityAttestor, GcpAgentIdentityAttestorOptions

GcpAgentIdentityAttestor(
    GcpAgentIdentityAttestorOptions(
        audience="my-defakto-agent",
        lifetime_seconds=600,  # 10 minutes
    )
)

The attestor re-reads the credentials on each attestation, so SVID rotation performed by the runtime is picked up automatically. On every call it revalidates the leaf certificate (notBefore/notAfter, EC P-256, SPIFFE URI SAN) and verifies the private key matches the leaf's public key before signing. The minted JWT has iss and sub both set to the SPIFFE ID from the leaf certificate's URI SAN.

Custom attestors

Implement the Attestor protocol to add any attestation source:

from pathlib import Path
from spiffe_defakto import AttestationEvidence, AttestingClientOptions, AttestingWorkloadAPIClient


class KubernetesAttestor:
    plugin_name = "k8s_sat"         # must match what the endpoint expects
    plugin_version = "1.0.0"

    async def collect_evidence(self) -> AttestationEvidence:
        token = Path("/var/run/secrets/kubernetes.io/serviceaccount/token").read_bytes()
        return AttestationEvidence(
            plugin_name=self.plugin_name,
            plugin_version=self.plugin_version,
            payload=token,
        )


async with AttestingWorkloadAPIClient(
    AttestingClientOptions(trust_domain_id="td-0000000", attestors=(KubernetesAttestor(),))
) as client:
    svid = await client.x509.get_svid()

When multiple attestors are configured, evidence is collected concurrently with asyncio.gather and all results are sent together in a single request.

Cloud & model provider credential integration

Once you have a SPIFFE client, you can exchange its JWT SVIDs for native cloud or model-provider credentials — no long-lived secrets, no service account keys. Each integration lives in a sub-package so its heavy SDK dependency (cloud SDK or model SDK) is optional.

All integrations self-bootstrap when no source is supplied: they construct a WorkloadAPIClient automatically, reading the socket path or attestor config from environment variables.

AWS (from_spiffe)

from_spiffe returns a botocore.credentials.RefreshableCredentials object that assumes an IAM role using the SVID as a web identity token. Wire it into a botocore session to use it with any boto3 client.

import boto3
import botocore.session
from spiffe_defakto.aws import FromSpiffeOptions, from_spiffe

credentials = from_spiffe(
    FromSpiffeOptions(role_arn="arn:aws:iam::123456789012:role/my-role")
)

botocore_session = botocore.session.get_session()
botocore_session._credentials = credentials
session = boto3.Session(botocore_session=botocore_session)
s3 = session.client("s3", region_name="eu-west-1")

AWS credentials refresh automatically: each time the STS session expires, from_spiffe fetches a fresh SVID and re-exchanges it.

role_arn falls back to the AWS_ROLE_ARN environment variable — useful on ECS task definitions, EC2 instance env, or Kubernetes injection:

AWS_ROLE_ARN=arn:aws:iam::123456789012:role/my-role

To use an existing SPIFFE client, pass it via source:

from spiffe_defakto import LocalWorkloadAPIClient
from spiffe_defakto.aws import FromSpiffeOptions, from_spiffe

spiffe = LocalWorkloadAPIClient()
credentials = from_spiffe(
    FromSpiffeOptions(
        source=spiffe.jwt,
        role_arn="arn:aws:iam::123456789012:role/my-role",
    )
)

Options

Azure (SPIFFECredential)

SPIFFECredential is an Azure TokenCredential that authenticates using a SPIFFE JWT SVID as a client assertion. Enables workload identity federation — no client secrets or certificates.

from azure.storage.blob import BlobServiceClient
from spiffe_defakto.azure import SPIFFECredential, SPIFFECredentialOptions

credential = SPIFFECredential(
    SPIFFECredentialOptions(tenant_id="your-tenant-id", client_id="your-app-client-id")
)

blob_client = BlobServiceClient(
    "https://mystorageaccount.blob.core.windows.net",
    credential,
)

SPIFFECredential is a drop-in replacement for DefaultAzureCredential in any Azure SDK call path.

Environment variable–driven usage — in environments where AZURE_TENANT_ID and AZURE_CLIENT_ID are injected (AKS, Container Apps, VM env), you can omit options entirely:

credential = SPIFFECredential()

To use an existing SPIFFE client, pass it via source:

from spiffe_defakto import LocalWorkloadAPIClient
from spiffe_defakto.azure import SPIFFECredential, SPIFFECredentialOptions

spiffe = LocalWorkloadAPIClient()
credential = SPIFFECredential(
    SPIFFECredentialOptions(source=spiffe.jwt, tenant_id="...", client_id="...")
)

If AZURE_AUTHORITY_HOST is set (e.g. for sovereign clouds), it is used as the authority host unless explicitly overridden via credential_kwargs.

Options

GCP (SPIFFEIdentityPoolClient)

SPIFFEIdentityPoolClient exchanges a SPIFFE JWT SVID for a GCP access token via Workload Identity Federation. It returns a google.auth.Credentials instance that can be passed as credentials= to any Google Cloud client library.

from google.cloud import storage
from spiffe_defakto.gcp import SPIFFEIdentityPoolClient, SPIFFEIdentityPoolClientOptions

credentials = SPIFFEIdentityPoolClient(
    SPIFFEIdentityPoolClientOptions(
        audience=(
            "//iam.googleapis.com/projects/123456789/locations/global/"
            "workloadIdentityPools/my-pool/providers/my-provider"
        ),
    )
)

client = storage.Client(credentials=credentials, project="my-project")

For setups that use service-account impersonation:

credentials = SPIFFEIdentityPoolClient(
    SPIFFEIdentityPoolClientOptions(
        audience="//iam.googleapis.com/projects/.../providers/my-provider",
        service_account_impersonation_url=(
            "https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/"
            "my-sa@my-project.iam.gserviceaccount.com:generateAccessToken"
        ),
    )
)

Environment-variable-driven usage — set GOOGLE_CLOUD_WORKLOAD_IDENTITY_POOL and optionally GOOGLE_CLOUD_PROJECT and you can omit options entirely. If CLOUDSDK_AUTH_IMPERSONATE_SERVICE_ACCOUNT is set, the impersonation URL is constructed automatically.

Options

OpenAI (subject_token_from_spiffe)

subject_token_from_spiffe returns an OpenAI workload identity subject token provider — the object passed as workload_identity["provider"] when constructing the OpenAI client. On each refresh the OpenAI SDK calls get_token() to obtain a short-lived JWT SVID, which it then exchanges for an OpenAI access token via OIDC Workload Identity Federation. No OPENAI_API_KEY is needed and there are no token files on disk.

from openai import AsyncOpenAI
from spiffe_defakto.openai import subject_token_from_spiffe

# Zero-arg: WorkloadAPIClient created lazily on first token refresh.
client = AsyncOpenAI(
    workload_identity={
        "identity_provider_id": "idp-...",
        "service_account_id": "sa-...",
        "provider": subject_token_from_spiffe(),
    },
)

response = await client.responses.create(
    model="gpt-5.5",
    input="Hello!",
)

To use an existing SPIFFE client, pass it via source:

from spiffe_defakto import LocalWorkloadAPIClient
from spiffe_defakto.openai import (
    SubjectTokenFromSpiffeOptions,
    subject_token_from_spiffe,
)

spiffe = LocalWorkloadAPIClient()
provider = subject_token_from_spiffe(
    SubjectTokenFromSpiffeOptions(
        source=spiffe.jwt,
        svid_audience="https://api.openai.com/v1",
    ),
)

Options

Verifying peer X.509 SVIDs

When your workload accepts mTLS connections from other SPIFFE workloads, use verify_x509_svid to validate the peer's certificate chain against a trust bundle you fetched yourself. The function enforces every rule from the SPIFFE spec: expiry on each certificate, the leaf-SVID constraints (non-root path, non-CA leaf, no keyCertSign / cRLSign key usage), and chain-signature verification back to one of the bundle's trust authorities.

from spiffe_defakto import WorkloadAPIClient, verify_x509_svid

async with WorkloadAPIClient() as client:
    # `peer_der_chain` is usually the list of DER certificates your TLS
    # transport hands you in a peer-verification callback.
    peer_id = await verify_x509_svid(peer_der_chain, client.x509)
    print("Peer identity:", peer_id)

The bundle_source argument is anything that implements :class:X509BundleSource — the x509 sub-object on any of the clients satisfies that protocol, so you can reuse the same client you're already using for your own SVIDs.

Error handling

All errors raised by this SDK are instances of SpiffeError:

from spiffe_defakto import SpiffeError, SpiffeErrorCode

try:
    svid = await client.x509.get_svid()
except SpiffeError as err:
    print(err.code, err)
    if err.code is SpiffeErrorCode.WORKLOAD_API_UNAVAILABLE:
        ...

Error codes

API reference

Main package (spiffe_defakto)

Sub-packages

License

Apache-2.0. See LICENSE.