Skip to main content

SDK Examples

Defakto provides SDKs for Python and TypeScript that let serverless workloads fetch SVIDs directly from the Trust Domain Server without a local SPIFFE agent.

The primary entry point is WorkloadAPIClient, a zero-argument client that reads its full configuration from environment variables. For cases requiring a custom transport or explicit programmatic setup, the lower-level AttestingWorkloadAPIClient is available.

Prerequisites​

Before using any of the examples below:

  • Serverless attestation must be enabled on the trust domain.
  • The appropriate attestation policy must be configured for your platform: AWS, GCP, or Azure.
  • A Defakto Trust Domain Server.

Finding your trust domain details​

The configuration values you need depend on whether your Trust Domain Server is self-hosted or managed by Defakto.

Self-hosted Trust Domain Server

Set DEFAKTO_SERVER_ADDRESS to the trust-domain server's hostname and port configured during deployment (e.g. my-server.internal:8443).

Defakto-hosted Trust Domain Server

Run spirlctl trust-domain list to find your trust domain name, then look up its ID:

spirlctl trust-domain info <trust-domain-name>

Example output:

ID                       td-m36ckrte4e
Name:                    example.org
Status:                  available
Self-Managed:            false

The ID (e.g. td-m36ckrte4e) is the value for DEFAKTO_TRUST_DOMAIN_ID.

Installation​

Requires Python 3.10+.

pip install spiffe-defakto

Install the extra for your cloud provider:

pip install 'spiffe-defakto[aws]'    # AWS
pip install 'spiffe-defakto[gcp]'    # GCP
pip install 'spiffe-defakto[azure]'  # Azure

Basic usage​

The recommended approach is to configure the SDK through environment variables and call WorkloadAPIClient() with no arguments. The client selects the right attestor and connects to the correct endpoint automatically, keeping your application code cloud-agnostic.

Set the following environment variables on your workload before it starts:

VariablePurpose
DEFAKTO_ATTESTORSA comma-separated list of attestor names to use for your platform. Possible values are aws_external_identity, gcp_service_account, or azure_managed_identity
DEFAKTO_SERVER_ADDRESSAddress of your Trust Domain Server in hostname or hostname:port format. Use for self-hosted trust domains
DEFAKTO_TRUST_DOMAIN_IDTrust domain ID (e.g. td-m36ckrte4e). Use for Defakto-hosted trust domains

If both DEFAKTO_SERVER_ADDRESS and DEFAKTO_TRUST_DOMAIN_ID are set, DEFAKTO_SERVER_ADDRESS takes precedence.

Example​

With the environment variables configured, use WorkloadAPIClient to fetch X.509 and JWT SVIDs:

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

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

asyncio.run(main())

Custom CA certificate​

When DEFAKTO_SERVER_ADDRESS points to a server with a certificate signed by a private CA, the CA root PEM must be made available to the SDK. This applies to both WorkloadAPIClient and AttestingWorkloadAPIClient.

Set GRPC_DEFAULT_SSL_ROOTS_FILE_PATH to the path of the CA root PEM file:

GRPC_DEFAULT_SSL_ROOTS_FILE_PATH=/path/to/ca.pem

Explicit configuration​

Use AttestingWorkloadAPIClient directly when you need to provide a custom transport or instantiate attestors explicitly rather than through environment variables.

AWS​

The AwsTokenAttestor fetches a signed JWT from AWS STS using the workload's attached IAM role.

Before deploying, configure an AWS Web Identity Token policy on the trust domain.

import asyncio
from spiffe_defakto import (
    AttestingClientOptions,
    AttestingWorkloadAPIClient,
    AwsTokenAttestor,
    TcpOptions,
)

async def main():
    options = AttestingClientOptions(
        attestors=[AwsTokenAttestor()],
        transport=TcpOptions(
            address="my-server.internal",
            port=8443,
        ),
    )
    async with AttestingWorkloadAPIClient(options) as client:
        svid = await client.x509.get_svid()
        print("SPIFFE ID:", svid.id)
        print("Expires at:", svid.expires_at.isoformat())

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

asyncio.run(main())

GCP​

The GcpIITAttestor fetches an Instance Identity Token from the GCP metadata service.

Before deploying, configure a GCP Instance Identity Token policy on the trust domain.

import asyncio
from spiffe_defakto import (
    AttestingClientOptions,
    AttestingWorkloadAPIClient,
    GcpIITAttestor,
    TcpOptions,
)

async def main():
    options = AttestingClientOptions(
        attestors=[GcpIITAttestor()],
        transport=TcpOptions(
            address="my-server.internal",
            port=8443,
        ),
    )
    async with AttestingWorkloadAPIClient(options) as client:
        svid = await client.x509.get_svid()
        print("SPIFFE ID:", svid.id)

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

asyncio.run(main())

Azure​

The AzureMSIAttestor fetches a Managed Identity token from the Azure Instance Metadata Service.

Before deploying, configure an Azure MSI policy on the trust domain.

import asyncio
from spiffe_defakto import (
    AttestingClientOptions,
    AttestingWorkloadAPIClient,
    AzureMSIAttestor,
    TcpOptions,
)

async def main():
    options = AttestingClientOptions(
        attestors=[AzureMSIAttestor()],
        transport=TcpOptions(
            address="my-server.internal",
            port=8443,
        ),
    )
    async with AttestingWorkloadAPIClient(options) as client:
        svid = await client.x509.get_svid()
        print("SPIFFE ID:", svid.id)

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

asyncio.run(main())