Add Custom Attributes with a Server Webhook

Workload attestation is the process of collecting evidence about a process requesting an SVID. The Defakto platform ships with attestors that can be used to collect common attributes from runtime environments, and can be extended for anything those attestors do not cover.

This guide adds custom attributes using the Server Workload Attestation Extension, a webhook the Trust Domain Server calls during issuance. To compare all other extension points, see Which extension do you need?.

What you'll build

A webhook that returns extra attributes for every workload, derived from the workload's namespace:

Namespace	Attributes returned
starts with `prod`	`environment=production`, `tier=critical`
starts with `staging`	`environment=staging`, `tier=standard`
anything else	`environment=development`, `tier=standard`

Namespace stands in for whatever attribute your real lookup would key off. The example hardcodes the result to keep the focus on the mechanics, in place of the business logic or external lookup an extension would perform.

Every attribute returned becomes available as custom.<key> that can be used to generate an SVID.

warning

This walkthrough uses a single pod and plain in-cluster HTTP, so the mechanics of the extension stay in focus. Before running an extension in production, see Availability and reliability for replica and timeout guidance, and Security for TLS and webhook authentication.

Prerequisites

A trust domain with Trust Domain Servers running on Kubernetes, installed with the spirl-server Helm chart
A Kubernetes cluster registered with Defakto, running the Agent and connected to the Trust Domain Servers above
kubectl, helm, and spirlctl, with a kubectl context for the cluster running your Trust Domain Servers
Your trust domain deployment ID, which is both the Helm release name and the namespace, and your cluster's ID. See Deploy Trust Domain Servers

Two terminals are useful. The webhook runs in the foreground in one, leaving the other free for the remaining commands.

Create the namespace and Service

The webhook must be reachable from the Trust Domain Server pods. This example installs it in the same Kubernetes cluster as the Trust Domain Server.

kubectl create namespace defakto-extension

cat > extension-webhook-service.yaml << 'EOF'
apiVersion: v1
kind: Service
metadata:
  name: extension-webhook
  namespace: defakto-extension
spec:
  selector:
    app: extension-webhook
  ports:
    - port: 8080
      targetPort: 8080
      protocol: TCP
      name: http
EOF

kubectl apply -f extension-webhook-service.yaml

The Service resolves to http://extension-webhook.defakto-extension.svc.cluster.local:8080 inside the cluster, which is the address the server will call.

Write and run the webhook

The webhook accepts POST /attest and returns a flat JSON object. Every key becomes a custom.* attribute. Two rules matter:

Values must be strings. The response is a flat map of string to string. Nested objects and non-string values are not attributes.
Always write a body. An empty body is not valid JSON, and the parse failure denies issuance rather than approving it. Return {} when the extension has nothing to add.

Go
Python
TypeScript

Start a pod using the Go image, labelled so the Service selects it:

extension-webhook-pod.yaml
apiVersion: v1
kind: Pod
metadata:
  name: extension-webhook
  namespace: defakto-extension
  labels:
    app: extension-webhook
spec:
  containers:
  - name: webhook
    image: golang:1.26
    command: ["sleep", "infinity"]

kubectl apply -f extension-webhook-pod.yaml
kubectl -n defakto-extension wait --for=condition=Ready pod/extension-webhook

Exec into the pod and set up the module:

kubectl -n defakto-extension exec -it extension-webhook -- bash

Inside the pod:

mkdir -p /webhook && cd /webhook
go mod init extension-webhook

Create the webhook. It uses only the standard library, so there is nothing to download:

cat > main.go << 'EOF'
package main

import (
	"encoding/json"
	"log"
	"maps"
	"net/http"
	"os"
	"slices"
	"strings"
)

func main() {
	port := os.Getenv("PORT")
	if port == "" {
		port = "8080"
	}

	http.HandleFunc("POST /attest", handleAttest)
	http.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) {
		w.WriteHeader(http.StatusOK)
	})

	log.Printf("listening on :%s", port)
	log.Fatal(http.ListenAndServe(":"+port, nil))
}

func handleAttest(w http.ResponseWriter, r *http.Request) {
	var req map[string]any
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		http.Error(w, `{"error":"invalid request body"}`, http.StatusBadRequest)
		return
	}

	// Log every attribute so you can see what is available to derive from.
	log.Printf("received request with attributes:")
	attrs := flatten("", req)
	for _, k := range slices.Sorted(maps.Keys(attrs)) {
		log.Printf("  %s=%s", k, attrs[k])
	}

	ns := attrs["kubernetes.pod.namespace"]
	custom := attributesFor(ns)

	log.Printf("ENRICH ns=%s environment=%s tier=%s",
		ns, custom["environment"], custom["tier"])

	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusOK)
	writeJSON(w, custom)
}

// attributesFor is the part you replace with a lookup against your own
// source of truth, such as a CMDB or service catalog. Cache the result:
// this runs on every SVID request.
func attributesFor(ns string) map[string]string {
	switch {
	case strings.HasPrefix(ns, "prod"):
		return map[string]string{"environment": "production", "tier": "critical"}
	case strings.HasPrefix(ns, "staging"):
		return map[string]string{"environment": "staging", "tier": "standard"}
	default:
		return map[string]string{"environment": "development", "tier": "standard"}
	}
}

// writeJSON always emits an object. An empty body fails to parse on the
// server and denies issuance.
func writeJSON(w http.ResponseWriter, body map[string]string) {
	if err := json.NewEncoder(w).Encode(body); err != nil {
		log.Printf("failed to write response: %v", err)
	}
}

// flatten turns the nested request into dotted keys such as
// kubernetes.pod.namespace.
func flatten(prefix string, m map[string]any) map[string]string {
	out := make(map[string]string)
	for k, v := range m {
		key := k
		if prefix != "" {
			key = prefix + "." + k
		}
		switch val := v.(type) {
		case map[string]any:
			maps.Copy(out, flatten(key, val))
		case string:
			out[key] = val
		}
	}
	return out
}
EOF

Run it in the foreground:
```
go run main.go
```
```
listening on :8080
```

Start a pod using the Python image, labelled so the Service selects it:

extension-webhook-pod.yaml
apiVersion: v1
kind: Pod
metadata:
  name: extension-webhook
  namespace: defakto-extension
  labels:
    app: extension-webhook
spec:
  containers:
  - name: webhook
    image: python:3-slim
    command: ["sleep", "infinity"]

kubectl apply -f extension-webhook-pod.yaml
kubectl -n defakto-extension wait --for=condition=Ready pod/extension-webhook

Exec into the pod:

kubectl -n defakto-extension exec -it extension-webhook -- bash

Inside the pod:

mkdir -p /webhook && cd /webhook

Create the webhook. It uses only the standard library, so there is nothing to install:

cat > main.py << 'EOF'
import json
import logging
import os
from http.server import BaseHTTPRequestHandler, HTTPServer

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
logger = logging.getLogger(__name__)


def flatten(prefix, m):
    """Turn the nested request into dotted keys such as kubernetes.pod.namespace."""
    out = {}
    for k, v in m.items():
        key = f"{prefix}.{k}" if prefix else k
        if isinstance(v, dict):
            out.update(flatten(key, v))
        elif isinstance(v, str):
            out[key] = v
    return out


def attributes_for(ns):
    """Replace this with a lookup against your own source of truth, such as
    a CMDB or service catalog. Cache the result: this runs on every SVID
    request."""
    if ns.startswith("prod"):
        return {"environment": "production", "tier": "critical"}
    if ns.startswith("staging"):
        return {"environment": "staging", "tier": "standard"}
    return {"environment": "development", "tier": "standard"}


class AttestHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        if self.path != "/attest":
            self.send_error(404)
            return

        length = int(self.headers.get("Content-Length", 0))
        try:
            req = json.loads(self.rfile.read(length))
        except json.JSONDecodeError:
            self._json(400, {"error": "invalid request body"})
            return

        # Log every attribute so you can see what is available to derive from.
        logger.info("received request with attributes:")
        attrs = flatten("", req)
        for k in sorted(attrs):
            logger.info("  %s=%s", k, attrs[k])

        ns = attrs.get("kubernetes.pod.namespace", "")
        custom = attributes_for(ns)

        logger.info(
            "ENRICH ns=%s environment=%s tier=%s",
            ns, custom["environment"], custom["tier"],
        )
        self._json(200, custom)

    def do_GET(self):
        if self.path == "/healthz":
            self.send_response(200)
            self.end_headers()
            return
        self.send_error(404)

    def log_message(self, *args):
        pass  # Suppress the default access log; attributes are logged instead.

    def _json(self, status, data):
        # Always write a body. An empty body fails to parse on the server
        # and denies issuance.
        body = json.dumps(data).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)


if __name__ == "__main__":
    port = int(os.environ.get("PORT", "8080"))
    logger.info("listening on :%d", port)
    HTTPServer(("", port), AttestHandler).serve_forever()
EOF

Run it in the foreground:
```
python -u main.py
```
```
listening on :8080
```

Start a pod using the Node image, labelled so the Service selects it:

extension-webhook-pod.yaml
apiVersion: v1
kind: Pod
metadata:
  name: extension-webhook
  namespace: defakto-extension
  labels:
    app: extension-webhook
spec:
  containers:
  - name: webhook
    image: node:26-slim
    command: ["sleep", "infinity"]

kubectl apply -f extension-webhook-pod.yaml
kubectl -n defakto-extension wait --for=condition=Ready pod/extension-webhook

Exec into the pod and install the TypeScript runner. Unlike the Go and Python versions, this step needs network access from the pod:
```
kubectl -n defakto-extension exec -it extension-webhook -- bash
```
Inside the pod:
```
mkdir -p /webhook && cd /webhook
npm install --no-save tsx @types/node
```

Create the webhook:

cat > main.ts << 'EOF'
import { createServer, IncomingMessage, ServerResponse } from "http";

const port = parseInt(process.env.PORT || "8080", 10);

// Turn the nested request into dotted keys such as kubernetes.pod.namespace.
function flatten(prefix: string, m: Record<string, unknown>): Record<string, string> {
  const out: Record<string, string> = {};
  for (const [k, v] of Object.entries(m)) {
    const key = prefix ? `${prefix}.${k}` : k;
    if (typeof v === "object" && v !== null && !Array.isArray(v)) {
      Object.assign(out, flatten(key, v as Record<string, unknown>));
    } else if (typeof v === "string") {
      out[key] = v;
    }
  }
  return out;
}

// Replace this with a lookup against your own source of truth, such as a
// CMDB or service catalog. Cache the result: this runs on every SVID request.
function attributesFor(ns: string): Record<string, string> {
  if (ns.startsWith("prod")) return { environment: "production", tier: "critical" };
  if (ns.startsWith("staging")) return { environment: "staging", tier: "standard" };
  return { environment: "development", tier: "standard" };
}

function readBody(req: IncomingMessage): Promise<string> {
  return new Promise((resolve, reject) => {
    const chunks: Buffer[] = [];
    req.on("data", (chunk) => chunks.push(chunk));
    req.on("end", () => resolve(Buffer.concat(chunks).toString()));
    req.on("error", reject);
  });
}

// Always write a body. An empty body fails to parse on the server and
// denies issuance.
function sendJSON(res: ServerResponse, status: number, body: unknown): void {
  res.writeHead(status, { "Content-Type": "application/json" });
  res.end(JSON.stringify(body));
}

const server = createServer(async (req: IncomingMessage, res: ServerResponse) => {
  if (req.method === "GET" && req.url === "/healthz") {
    res.writeHead(200);
    res.end();
    return;
  }

  if (req.method !== "POST" || req.url !== "/attest") {
    res.writeHead(404);
    res.end();
    return;
  }

  let parsed: Record<string, unknown>;
  try {
    parsed = JSON.parse(await readBody(req));
  } catch {
    sendJSON(res, 400, { error: "invalid request body" });
    return;
  }

  // Log every attribute so you can see what is available to derive from.
  console.log(`received request with attributes:`);
  const attrs = flatten("", parsed);
  for (const k of Object.keys(attrs).sort()) {
    console.log(`  ${k}=${attrs[k]}`);
  }

  const ns = attrs["kubernetes.pod.namespace"] ?? "";
  const custom = attributesFor(ns);

  console.log(`ENRICH ns=${ns} environment=${custom.environment} tier=${custom.tier}`);
  sendJSON(res, 200, custom);
});

server.listen(port, () => {
  console.log(`listening on :${port}`);
});
EOF

Run it in the foreground:
```
npx tsx main.ts
```
```
listening on :8080
```

note

The webhook is a child of your kubectl exec session, so it stops when that session ends. Leave this terminal running and use a second one for the remaining steps. The live log is the clearest view of each request as it happens.

Wire it into Defakto

In your second terminal, point the Trust Domain Server at the Service. The Helm release name and the namespace both equal your trust domain deployment ID.

Find it with spirlctl trust-domain deployment list. It is the tdd- prefixed value in the ID column:

spirlctl trust-domain deployment list

Name       ID              Configuration State  Last Configured
us-west-2  tdd-b80yo4kabl  Up to date           2026-08-12 13:55:11.823 +0000 UTC
us-east-1  tdd-e8m6d8tx1u  Up to date           2026-08-12 13:55:20.558 +0000 UTC

2 trust domain deployments found.

export DEPLOYMENT_ID=<your-trust-domain-deployment-id>

helm upgrade --install "$DEPLOYMENT_ID" \
  oci://ghcr.io/spirl/charts/spirl-server \
  --namespace "$DEPLOYMENT_ID" \
  --reuse-values \
  --set trustDomainDeployment.deployment.extensionWorkloadAttestation.webhookUrl="http://extension-webhook.defakto-extension.svc.cluster.local:8080/attest" \
  --set trustDomainDeployment.deployment.extensionWorkloadAttestation.timeout="10s"

kubectl -n "$DEPLOYMENT_ID" rollout status deployment -l app.kubernetes.io/name=spirl-server

--reuse-values preserves the rest of your server configuration.

warning

This modifies your existing Helm installation of the Trust Domain Server. If that installation is managed by automation or CI/CD, the next run may revert this change. Make the equivalent change in your CI/CD configuration if you want it to persist.

Use the attributes

The webhook now returns environment and tier on every request, available as custom.environment and custom.tier. Nothing observable changes yet. A custom attribute has no effect until something references it, which is the step most often missed.

There are three places to consume them. Pick whichever fits how your workloads authorize each other.

In the SPIFFE ID

Putting the attribute in the path makes it visible to every peer that validates the identity, and usable in SPIFFE-ID-based authorization policies:

cat > svid-issuance-policy.yaml << 'EOF'
section: SVIDIssuancePolicy
schema: v1
spec:
  policy:
    pathTemplate: "/{{custom.environment}}/ns/{{kubernetes.pod.namespace}}/sa/{{kubernetes.pod.service_account}}"
EOF

Find the cluster's ID with spirlctl cluster list. It is the c- prefixed value in the ID column:

spirlctl cluster list

Name                    ID            Trust Domain          Created
prod-us-east-1          c-ab49xuq9kh  example.com           2026-08-05T03:34:37Z
staging-us-east-1       c-4c4z9d3tjo  example.com           2026-07-23T18:11:37Z

2 clusters found.

Apply the policy to that cluster:

export CLUSTER_ID=<your-cluster-id>
spirlctl config set cluster --id "$CLUSTER_ID" svid-issuance-policy.yaml

Managed Configuration will take up to a minute to apply the changes.

warning

When a path template references an attribute, every workload covered by that policy must have the attribute or its SVID cannot be built. The policy above applies to every workload in the cluster, so any workload missing the attribute stops receiving SVIDs. The webhook above always returns a value, defaulting to development, which is what makes this safe. If you replace that with a lookup in an external system, decide what it should return when there is no matching result.

See SPIFFE ID templates for the full syntax.

In the X.509 certificate

Attributes can populate certificate subject fields and SANs, which suits peers that authorize on certificate contents rather than on the SPIFFE ID. See X.509-SVID customization.

As JWT claims

Attributes can be added as claims on JWT-SVIDs, which suits federating to an external system that reads claims. See JWT-SVID customization.

Verify

On the Kubernetes cluster running your agents, create a namespace that will be classified as production:

kubectl create namespace prod-payments

kubectl run svid-check -n prod-payments -l "k8s.spirl.com/spiffe-csi=enabled" --restart=Never --image ghcr.io/spirl/spirldbg:latest --rm -i -- spirldbg svid-x509

The webhook terminal shows the attributes it received and what it returned:

received request with attributes:
  cluster.cluster_id=c-xxxxxxxxxx
  kubernetes.pod.name=svid-checker
  kubernetes.pod.namespace=prod-payments
  kubernetes.pod.service_account=default
  ...
ENRICH ns=prod-payments environment=production tier=critical

If you applied the path template above, the attribute appears in the identity itself:

Successfully received x509 SVID
SPIFFE ID: spiffe://your-trust-domain/production/ns/prod-payments/sa/default

Without a template referencing it, the SVID is issued normally and the attribute simply goes unused. The attribute appears in the audit logs shown in the Defakto Console.

If an attribute you expect is missing, check that the webhook returned it as a string at the top level of the response, and that Attribute Redaction is not filtering custom.*.

Clean up

Restore the cluster's default path template first, if you changed it:

cat > svid-issuance-policy.yaml << 'EOF'
section: SVIDIssuancePolicy
schema: v1
spec:
  policy:
    pathTemplate: "/{{cluster.name}}/ns/{{kubernetes.pod.namespace}}/sa/{{kubernetes.pod.service_account}}"
EOF
spirlctl config set cluster --id "$CLUSTER_ID" svid-issuance-policy.yaml

Then remove the extension and the webhook:

helm upgrade "$DEPLOYMENT_ID" \
  oci://ghcr.io/spirl/charts/spirl-server \
  --namespace "$DEPLOYMENT_ID" \
  --reuse-values \
  --set trustDomainDeployment.deployment.extensionWorkloadAttestation.webhookUrl=""
kubectl -n "$DEPLOYMENT_ID" rollout status deployment -l app.kubernetes.io/name=spirl-server

On the cluster with the agents:

kubectl delete namespace prod-payments

On the cluster with the servers:

kubectl delete namespace defakto-extension

Next steps

Server Workload Attestation Extension — Configuration fields, retry behavior, filtering, and troubleshooting
Deny SVID Issuance — The same surface used to refuse an identity rather than to enrich one
Registration Entries — Assign each workload's SPIFFE ID path from a registration list
Attribute Redaction — Control which attributes reach the server

What you'll build​

Prerequisites​

Create the namespace and Service​

Write and run the webhook​

Wire it into Defakto​

Use the attributes​

In the SPIFFE ID​

In the X.509 certificate​

As JWT claims​

Verify​

Clean up​

Next steps​

What you'll build

Prerequisites

Create the namespace and Service

Write and run the webhook

Wire it into Defakto

Use the attributes

In the SPIFFE ID

In the X.509 certificate

As JWT claims

Verify

Clean up

Next steps