Hedronite · Dev Lesson · Polyglot-Dev / Python · Mon 2026-09-21

Python boto3 — EKS Pod Identity association census

List associations before you delete an IRSA annotation. Guessing which ServiceAccounts still bind IAM is how silent AccessDenied survives review.

Lesson Class: Dev (Python + boto3 + EKS)
Language: Python
Paired Ops: EKS Pod Identity versus IRSA
Paired Cert: CKS ServiceAccount hardening + Pod Identity
Grounding: AWS_ML EKS.md · KUR Ch.14 referenced
List
eks.list_pod_identity_associations (paginate nextToken).
Describe
status + createdAt per associationId.
Diff
Join against remaining eks.amazonaws.com/role-arn annotations.
Census associations first. Diff annotations second.

<!-- hal:authoritative:yaml -->

List associations before you delete an IRSA annotation. Guessing which ServiceAccounts still bind IAM is how silent AccessDenied survives review.

§I — Frame

Use boto3 eks to inventory Pod Identity associations for a cluster. Optionally use the official Python kubernetes client to list ServiceAccounts that still carry eks.amazonaws.com/role-arn. Pair with Ops Pod Identity versus IRSA. Do not reopen Gateway CustomObjects census (09-18) or PSA label walking (09-12). Do not treat this as a full reimplementation of the 09-06 IRSA annotation-only census.

§II — Client bootstrap

import boto3
from kubernetes import client, config

eks = boto3.client("eks")
CLUSTER = "prod"

config.load_kube_config()
v1 = client.CoreV1Api()
IRSA_ANN = "eks.amazonaws.com/role-arn"

Region comes from the usual boto3 chain (env, profile, instance role). The cluster name is an argument, not a guess from kube-context alone (contexts can lie across accounts).

§III — Association census rows

def list_all_associations(cluster_name: str):
    rows, token = [], None
    while True:
        kwargs = {"clusterName": cluster_name}
        if token:
            kwargs["nextToken"] = token
        resp = eks.list_pod_identity_associations(**kwargs)
        for a in resp.get("associations", []):
            rows.append({
                "associationId": a.get("associationId"),
                "namespace": a.get("namespace"),
                "serviceAccount": a.get("serviceAccount"),
                "roleArn": a.get("roleArn"),
                "ownerArn": a.get("ownerArn"),
            })
        token = resp.get("nextToken")
        if not token:
            break
    return rows

For status and create-time detail, call describe_pod_identity_association per id:

def enrich(cluster_name: str, rows: list[dict]) -> list[dict]:
    out = []
    for r in rows:
        d = eks.describe_pod_identity_association(
            clusterName=cluster_name,
            associationId=r["associationId"],
        )["association"]
        r = dict(r)
        r["status"] = (d.get("status") or {}).get("status") or d.get("status")
        r["createdAt"] = str(d.get("createdAt"))
        out.append(r)
    return out

Normalize status against the live response shape in your botocore version. Print a stable key for review: namespace/serviceAccount -> roleArn.

§IV — Remaining IRSA annotation pass

def irsa_annotated_sas(namespaces=None):
    hits = []
    if namespaces:
        sa_lists = [v1.list_namespaced_service_account(ns) for ns in namespaces]
    else:
        sa_lists = [v1.list_service_account_for_all_namespaces()]
    for listing in sa_lists:
        for sa in listing.items:
            ann = (sa.metadata.annotations or {})
            role = ann.get(IRSA_ANN)
            if not role:
                continue
            hits.append({
                "ns": sa.metadata.namespace,
                "name": sa.metadata.name,
                "roleArn": role,
            })
    return hits

Join in Python:

assoc = {(r["namespace"], r["serviceAccount"]): r for r in list_all_associations(CLUSTER)}
irsa = irsa_annotated_sas()

both, only_assoc, only_irsa = [], [], []
for h in irsa:
    key = (h["ns"], h["name"])
    if key in assoc:
        both.append({**h, "associationId": assoc[key]["associationId"]})
    else:
        only_irsa.append(h)
for key, r in assoc.items():
    if not any((h["ns"], h["name"]) == key for h in irsa):
        only_assoc.append(r)

Migration hygiene: both is the dual-bind risk set. only_irsa is the backlog. only_assoc is the Pod Identity-native set.

§V — Failure modes

SymptomLikely cause
list_pod_identity_associations AccessDeniedCaller IAM missing eks:ListPodIdentityAssociations (and Describe)
Empty list on a "migrated" clusterAssociations on another cluster name / account / region
Association row exists, app still AccessDeniedAgent not Ready; wrong SA on Pod; role permission policy too tight
Annotation still present after cutoverCensus not run; GitOps still applying old SA manifest

§VI — Minimal CLI smoke (same data)

aws eks list-pod-identity-associations --cluster-name prod --output table
kubectl get sa -A -o json \
  | jq -r '.items[] | select(.metadata.annotations["eks.amazonaws.com/role-arn"] != null)
    | [.metadata.namespace,.metadata.name,.metadata.annotations["eks.amazonaws.com/role-arn"]] | @tsv'

Use the Python join when you need a repeatable report artifact.

§VII — Closing

Census associations first. Diff against IRSA annotations second. Remove annotations only for ServiceAccounts that prove credentials on the Pod Identity path.

Related