Skip to main content

spiffe-defakto

PyPI Python License: Apache-2.0

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:

Framework Guide
asyncio (stdlib) docs/frameworks/asyncio.md
FastAPI docs/frameworks/fastapi.md
Starlette docs/frameworks/starlette.md
aiohttp docs/frameworks/aiohttp.md
Quart docs/frameworks/quart.md
Flask docs/frameworks/flask.md
Django docs/frameworks/django.md
Celery docs/frameworks/celery.md
AnyIO docs/frameworks/anyio.md
Trio docs/frameworks/trio.md

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).

Name Attestor class Environment
aws_external_identity AwsTokenAttestor AWS (Lambda, EC2, …)
azure_managed_identity AzureMSIAttestor Azure Managed Identity
gcp_service_account GcpIITAttestor Google Cloud
gcp_agent_identity GcpAgentIdentityAttestor GCP Vertex AI Agent Engine
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

# Serverless Realms (optional): scope issuance to a specific serverless cluster.
# Omit for trust-domain-scoped issuance.
# DEFAKTO_CLUSTER_ID=cl-0000000

To target a specific serverless cluster's policy set (Realms, RFD-194), set the DEFAKTO_CLUSTER_ID environment variable — the zero-config WorkloadAPIClient reads it automatically. To set it in code, pass cluster_id on AttestingClientOptions when constructing an AttestingWorkloadAPIClient directly (WorkloadAPIClient takes no options). An empty value selects the trust-domain-scoped policy set, so this is backward-compatible.

# Zero-config: WorkloadAPIClient reads DEFAKTO_CLUSTER_ID from the environment.
# Application code stays the same in every environment.
async with WorkloadAPIClient() as client:
    svid = await client.x509.get_svid()

# Or set it explicitly by constructing the attesting client directly:
async with AttestingWorkloadAPIClient(
    AttestingClientOptions(cluster_id="cl-0000000")
) 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

Option Type Default Description
audience tuple[str, ...] ("urn:defakto:security:server",) JWT audience (aud claim).
signing_algorithm Literal["RS256", "ES384"] "RS256" STS token signing algorithm.
duration_seconds int 60 Token validity in seconds (60–3600).
tags tuple[tuple[str, str], ...] () Custom STS session tags.
sts_config dict[str, object] {} Forwarded to boto3.client('sts', **sts_config).
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

Option Type Default Description
audience str "urn:defakto:security:server" Audience for the identity token.
service_account str "default" Service account to fetch the identity token for.
identity_token_host str "metadata.google.internal" Override the metadata service host.
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

Option Type Default Description
credentials_path str /var/run/secrets/workload-spiffe-credentials/credentials.json Path to the runtime credentials pointer file.
audience str "urn:defakto:security:server" Audience (aud claim) of the attestation token, not of any JWT-SVID later requested.
lifetime_seconds int 300 (5 minutes) JWT lifetime in seconds (1–900). Short-lived by design.
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

Option Type Default Description
source JwtSVIDSource WorkloadAPIClient().jwt (lazy) JWT SVID source; created lazily on first refresh if not provided.
role_arn str AWS_ROLE_ARN env var ARN of the IAM role to assume.
svid_audience str | tuple[str, ...] "urn:defakto:security:server" Audience(s) for the JWT SVID — must match the IAM identity provider.
role_session_name str "spiffe-session" Name for the assumed role session.
provider_id str | None None FQDN of the identity provider.
policy_arns tuple[dict[str, str], ...] () Managed session policies.
policy str | None None Inline session policy JSON.
duration_seconds int 3600 Role session duration in seconds.
client_config dict[str, object] {} Forwarded to boto3.client('sts', **client_config).

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

Option Type Default Description
source JwtSVIDSource WorkloadAPIClient().jwt JWT SVID source.
tenant_id str | None AZURE_TENANT_ID Azure AD tenant ID.
client_id str | None AZURE_CLIENT_ID Azure AD application (client) ID.
svid_audience str | tuple[str, ...] "api://AzureADTokenExchange" Audience(s) for the JWT SVID.
credential_kwargs dict[str, object] {} Extra keyword arguments for the inner ClientAssertionCredential.

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

Option Type Default Description
source JwtSVIDSource WorkloadAPIClient().jwt JWT SVID source.
audience str GOOGLE_CLOUD_WORKLOAD_IDENTITY_POOL Workload Identity Pool audience string.
svid_audience str | tuple[str, ...] value of audience Audience(s) for the JWT SVID.
service_account_impersonation_url str | None constructed from CLOUDSDK_AUTH_IMPERSONATE_SERVICE_ACCOUNT Optional service-account impersonation URL.
project_id str | None GOOGLE_CLOUD_PROJECT GCP project ID; required when running outside GCP.
universe_domain str | None "googleapis.com" Only needed for non-public GCP universes.
quota_project_id str | None GOOGLE_CLOUD_QUOTA_PROJECT Project used for quota and billing attribution.

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

Option Type Default Description
source JwtSVIDSource | None WorkloadAPIClient().jwt JWT SVID source; created lazily on first refresh if not provided.
svid_audience str | tuple[str, ...] "https://api.openai.com/v1" Audience(s) for the JWT SVID — must satisfy the provider's audience match.

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

Code Description
INVALID_SPIFFE_ID Malformed SPIFFE ID string.
INVALID_TRUST_DOMAIN Malformed trust domain.
BUNDLE_NOT_FOUND No bundle for the requested trust domain.
SVID_EXPIRED The SVID has expired.
SVID_INVALID The SVID is structurally invalid.
SVID_AUDIENCE_MISMATCH JWT audience does not match.
SOCKET_PATH_NOT_CONFIGURED No socket path was configured.
WORKLOAD_API_UNAVAILABLE Cannot reach the Workload API endpoint.
WORKLOAD_API_ERROR Workload API returned an unexpected error.
JWT_PARSE_ERROR Failed to parse a JWT.
JWT_SIGNATURE_INVALID JWT signature verification failed.
TOKEN_EXCHANGE_FAILED Token exchange failed due to network, HTTP, or protocol errors.
SERVER_ADDRESS_NOT_CONFIGURED No server address configured: DEFAKTO_SERVER_ADDRESS, DEFAKTO_TRUST_DOMAIN_ID, and transport are all unset.
INVALID_SERVER_ADDRESS DEFAKTO_SERVER_ADDRESS could not be parsed as a valid host or host:port.
NO_ATTESTORS_CONFIGURED AttestingWorkloadAPIClient requires at least one attestor.
ATTESTOR_COLLECTION_FAILED An attestor failed to collect evidence.
ATTESTATION_FAILED The endpoint rejected the attestation.

API reference

Main package (spiffe_defakto)

Export Kind Description
WorkloadAPIClient class Smart wrapper — auto-selects local or attesting client.
LocalWorkloadAPIClient class Standard SPIFFE Workload API client (gRPC to local agent).
AttestingWorkloadAPIClient class Defakto attesting client for serverless environments.
TokenExchangeClient class Token exchange client returning a JWT-SVID.
AwsTokenAttestor class Built-in AWS IAM attestor (STS GetWebIdentityToken).
AzureMSIAttestor class Built-in Azure Managed Identity attestor.
GcpAgentIdentityAttestor class Built-in GCP Vertex AI Agent Engine attestor.
GcpIITAttestor class Built-in GCP Instance Identity Token attestor.
SpiffeError class Exception type raised by the SDK.
SpiffeErrorCode enum All possible error codes.
Attestor protocol Implement to create a custom attestor.
AttestationEvidence class Evidence payload returned by an attestor.
ClientOptions / UnixOptions / TcpOptions types Transport options for LocalWorkloadAPIClient.
AttestingClientOptions class Options for AttestingWorkloadAPIClient.
AwsTokenAttestorOptions class Options for AwsTokenAttestor.
AzureMSIAttestorOptions class Options for AzureMSIAttestor.
GcpAgentIdentityAttestorOptions class Options for GcpAgentIdentityAttestor.
GcpIITAttestorOptions class Options for GcpIITAttestor.
X509Context class Snapshot of all X.509 SVIDs and bundles.
X509SVID / X509Bundle classes X.509 SVID and trust bundle types.
JwtSVID / JwtBundle classes JWT SVID and trust bundle types.
JwtClaims TypedDict Standard JWT-SVID claim shape for IDE autocomplete.
SpiffeID / TrustDomain classes Identity value objects.
marshal_x509_svid / parse_marshaled_x509_svid fns PEM (de)serialisation.
verify_x509_svid function Validate a peer's X.509 SVID chain against a bundle source.
VerifyX509SVIDOptions class Options for verify_x509_svid.
parse_and_validate_jwt_svid function Verify a JWT SVID against a bundle source.

Sub-packages

Import Description
spiffe_defakto.sync → SyncWorkloadAPIClient Blocking facade over the async API for scripts and non-async frameworks.
spiffe_defakto.aws → from_spiffe Botocore-compatible credentials provider using a SPIFFE SVID as a web identity token.
spiffe_defakto.azure → SPIFFECredential Azure TokenCredential backed by a SPIFFE JWT SVID (workload identity federation).
spiffe_defakto.gcp → SPIFFEIdentityPoolClient google-auth credentials using a SPIFFE JWT SVID for GCP Workload Identity Federation.
spiffe_defakto.openai → subject_token_from_spiffe OpenAI workload-identity subject token provider backed by a SPIFFE JWT SVID.
spiffe_defakto.token_exchange → TokenExchangeClient RFC 8693 token exchange client returning a JWT-SVID.

License

Apache-2.0. See LICENSE.

Metadata

Release files for spiffe-defakto 1.1.0

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for spiffe-defakto 1.1.0
File Size Uploaded Provenance
spiffe_defakto-1.1.0.tar.gz 78.3 kB With provenance Details

Built distribution (wheel)

Table of built distributions (wheels) for spiffe-defakto 1.1.0
File Interpreter ABI Platform
spiffe_defakto-1.1.0-py3-none-any.whl Python 3 none any Details

Total release size: 161.0 kB

Release files / spiffe_defakto-1.1.0.tar.gz

Download URL spiffe_defakto-1.1.0.tar.gz
Size 78.3 kB
Tags Source
SHA-256 checksum
How to use checksums
68983bb23e0bbd279fab887d5688a0faf89d0e3f104bc53b27ca0640dbf7c40f
BLAKE2b-256 checksum
How to use checksums
e68db3b3d536c62d59d873a941e1caaffb4325d098041cb5dc3bbc38adb24ad2
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
Yes
Uploaded via twine/7.0.0 CPython/3.13.14

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

PyPI Publish Attestation

PyPI verified that this artifact, at this checksum, originated from the publisher listed below.

Signed by GitHub Actions, verified by PyPI on Aug 25, 2026.

Transparency log

Release files / spiffe_defakto-1.1.0-py3-none-any.whl

Download URL spiffe_defakto-1.1.0-py3-none-any.whl
Size 82.7 kB
Tags Python 3
SHA-256 checksum
How to use checksums
620a5eca6bb6a7752558b17ef67e76411f5a52624e706740d996c456bb1fd41c
BLAKE2b-256 checksum
How to use checksums
42fb2ee98b5b7f8381582f707b8494a7d6b6c124e7934ba99a208f550fbabb93
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
Yes
Uploaded via twine/7.0.0 CPython/3.13.14

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

PyPI Publish Attestation

PyPI verified that this artifact, at this checksum, originated from the publisher listed below.

Signed by GitHub Actions, verified by PyPI on Aug 25, 2026.

Transparency log

Security for spiffe-defakto 1.1.0

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

Full provenance

All 2 files in this release have been uploaded with provenance metadata.

Select from 2 files

Attestation for spiffe_defakto-1.1.0.tar.gz

PyPI Publish Attestation

PyPI verified that this artifact, at this checksum, originated from the publisher listed below.

Signed by GitHub Actions, verified by PyPI on Aug 25, 2026.

Transparency log

A note on package security

PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.

To learn more about protecting yourself from malicious actors, visit our package security documentation.

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page