Metadata-Version: 2.4
Name: spiffe-defakto
Version: 1.0.2
Summary: Python SDK for SPIFFE workload identity and Defakto attestation.
Project-URL: Homepage, https://defakto.security
Project-URL: Source, https://github.com/defakto-security/spiffe-defakto-python
Project-URL: Issues, https://github.com/defakto-security/spiffe-defakto-python/issues
Project-URL: Changelog, https://github.com/defakto-security/spiffe-defakto-python/blob/main/CHANGELOG.md
Author-email: Defakto <support@defakto.security>
License-Expression: Apache-2.0
License-File: LICENSE
Keywords: attestation,defakto,mtls,spiffe,svid,workload-identity,zero-trust
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Operating System :: MacOS
Classifier: Operating System :: POSIX
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Security
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: cryptography>=42.0
Requires-Dist: grpcio>=1.60
Requires-Dist: httpx>=0.27
Requires-Dist: protobuf>=4.25
Requires-Dist: pyjwt[crypto]>=2.8
Provides-Extra: all
Requires-Dist: aiohttp>=3.9; extra == 'all'
Requires-Dist: azure-core>=1.30; extra == 'all'
Requires-Dist: azure-identity>=1.17; extra == 'all'
Requires-Dist: boto3>=1.34; extra == 'all'
Requires-Dist: google-auth>=2.28; extra == 'all'
Provides-Extra: aws
Requires-Dist: boto3>=1.34; extra == 'aws'
Provides-Extra: azure
Requires-Dist: aiohttp>=3.9; extra == 'azure'
Requires-Dist: azure-core>=1.30; extra == 'azure'
Requires-Dist: azure-identity>=1.17; extra == 'azure'
Provides-Extra: gcp
Requires-Dist: google-auth>=2.28; extra == 'gcp'
Description-Content-Type: text/markdown

# spiffe-defakto

[![PyPI](https://img.shields.io/pypi/v/spiffe-defakto.svg)](https://pypi.org/project/spiffe-defakto/)
[![Python](https://img.shields.io/pypi/pyversions/spiffe-defakto.svg)](https://pypi.org/project/spiffe-defakto/)
[![License: Apache-2.0](https://img.shields.io/badge/License-Apache--2.0-blue.svg)](LICENSE)

> **Stable — 1.x.** The public API follows
> [Semantic Versioning](https://semver.org/spec/v2.0.0.html): no breaking
> changes within a major version. See the [changelog](CHANGELOG.md)
> before upgrading.

A Python SDK for [SPIFFE](https://spiffe.io/) 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

```sh
pip install spiffe-defakto
```

```python
# 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:

```python
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:

```python
# 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](docs/frameworks/asyncio.md) |
| FastAPI          | [docs/frameworks/fastapi.md](docs/frameworks/fastapi.md) |
| Starlette        | [docs/frameworks/starlette.md](docs/frameworks/starlette.md) |
| aiohttp          | [docs/frameworks/aiohttp.md](docs/frameworks/aiohttp.md) |
| Quart            | [docs/frameworks/quart.md](docs/frameworks/quart.md) |
| Flask            | [docs/frameworks/flask.md](docs/frameworks/flask.md) |
| Django           | [docs/frameworks/django.md](docs/frameworks/django.md) |
| Celery           | [docs/frameworks/celery.md](docs/frameworks/celery.md) |
| AnyIO            | [docs/frameworks/anyio.md](docs/frameworks/anyio.md) |
| Trio             | [docs/frameworks/trio.md](docs/frameworks/trio.md) |

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

## Installation

```sh
# 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

```sh
# 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`.
3. Otherwise → raises `SpiffeError` with code `SOCKET_PATH_NOT_CONFIGURED`.

```python
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_TRUST_DOMAIN_ID`
to resolve the Defakto endpoint.

| Name                     | Attestor class     | Environment            |
| ------------------------ | ------------------ | ---------------------- |
| `aws_external_identity`  | `AwsTokenAttestor` | AWS (Lambda, EC2, …)   |
| `azure_managed_identity` | `AzureMSIAttestor` | Azure Managed Identity |
| `gcp_service_account`    | `GcpIITAttestor`   | Google Cloud           |

```sh
# Serverless AWS Lambda — no code changes needed.
DEFAKTO_ATTESTORS=aws_external_identity
DEFAKTO_TRUST_DOMAIN_ID=td-0000000
```

```python
# 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](https://github.com/spiffe/spiffe/blob/main/standards/SPIFFE_Workload_API.md)
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`.

```python
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

```python
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

```python
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.

```python
# 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.

```python
# 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:

```python
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.

Setting `trust_domain_id` automatically resolves the endpoint to
`<trust_domain_id>.agent.spirl.com:443` over TLS. Override with an explicit
`transport` option (same shape as `LocalWorkloadAPIClient`).

### X.509 SVID

```python
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

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

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

---

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

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

```python
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.

```python
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`:

```python
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.

```python
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.             |

```python
from spiffe_defakto import GcpIITAttestor, GcpIITAttestorOptions

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

---

### Custom attestors

Implement the `Attestor` protocol to add any attestation source:

```python
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 provider credential integration

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

All three 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.

```python
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:

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

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

```python
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.

```python
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:

```python
credential = SPIFFECredential()
```

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

```python
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.

```python
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:

```python
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.                                                      |

---

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

```python
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`:

```python
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.                           |
| `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.      |
| `AwsTokenAttestor`            | class    | Built-in AWS IAM attestor (STS `GetWebIdentityToken`).     |
| `AzureMSIAttestor`            | class    | Built-in Azure Managed Identity 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`.                            |
| `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.      |

## License

Apache-2.0. See [LICENSE](LICENSE).
