Python IRSA census — ServiceAccount annotations and Pod joins

List ServiceAccounts that claim an IAM role. Join Pods. Fail CI on unused IRSA and default riders.

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

*List every ServiceAccount. Record who carries eks.amazonaws.com/role-arn. Then list Pods and catch the ones that never use that ServiceAccount.*

§I — Frame

Ops today wires IRSA: OIDC provider, annotation, IAM trust. Dev's job is the inventory that proves the wire-up survived GitOps and human hands.

09-03's Python Ingress census walked Ingress rules against Service backends. 08-31 walked SecurityContext fields. 08-04 taught in-cluster config and projected token refresh for calling the API as a Pod. Today reuses that client posture to answer a narrower question: which ServiceAccounts claim an IAM role, and which Pods actually select them?

The annotation key is exact: eks.amazonaws.com/role-arn. Misspell it and the webhook ignores you. The census treats spelling as data.

§II — Client bootstrap

Out-of-cluster (laptop / CI with kubeconfig):

from kubernetes import client, config

config.load_kube_config()
v1 = client.CoreV1Api()

In-cluster (a census Job using its own IRSA-or-RBAC SA):

config.load_incluster_config()
v1 = client.CoreV1Api()

08-04 covered projected token refresh. The census Job needs RBAC get,list on serviceaccounts and pods across the namespaces you care about. Prefer a dedicated irsa-census ServiceAccount with a Role per namespace (or a tight ClusterRole if the platform team owns cluster-wide inventory). Do not reuse invoice-reader's IRSA role for the census tool; that mixes data-plane AWS rights with control-plane read rights.

§III — ServiceAccount annotation pass

IRSA_KEY = "eks.amazonaws.com/role-arn"

def sa_rows(v1, namespaces):
    rows = []
    for ns in namespaces:
        for sa in v1.list_namespaced_service_account(ns).items:
            ann = sa.metadata.annotations or {}
            rows.append({
                "namespace": ns,
                "name": sa.metadata.name,
                "role_arn": ann.get(IRSA_KEY),
                "automount": sa.automount_service_account_token,
            })
    return rows

Print three buckets:

1. Annotated. role_arn is a non-empty string. Validate it looks like arn:aws:iam:::digits:role/.
2. Unannotated non-default. Workload SAs that should have been annotated but were not.
3. Default. Expect automount false in hardened namespaces (Kubestronaut / CKS posture). Flag default with a role ARN as a smell: shared identity.

import re
ARN_RE = re.compile(r"^arn:aws:iam::\d{12}:role/[\w+=,.@-]+$")

def classify(rows):
    annotated, bad_arn, plain, defaults = [], [], [], []
    for r in rows:
        if r["name"] == "default":
            defaults.append(r)
            continue
        if not r["role_arn"]:
            plain.append(r)
            continue
        if not ARN_RE.match(r["role_arn"]):
            bad_arn.append(r)
        else:
            annotated.append(r)
    return annotated, bad_arn, plain, defaults

Bad ARNs catch typos and partial pastes (role/invoice without account, or SSO permission-set ARNs that are not what the webhook expects).

§IV — Pod binding pass

An annotated ServiceAccount that no Pod selects is dead config. A Pod that selects default while a sibling SA holds the role is the live failure mode from Ops §IV.

def pod_sa_usage(v1, namespaces):
    usage = {}
    mismatches = []
    for ns in namespaces:
        for pod in v1.list_namespaced_pod(ns).items:
            if pod.status.phase in ("Succeeded", "Failed"):
                continue
            sa_name = pod.spec.service_account_name or "default"
            key = (ns, sa_name)
            usage[key] = usage.get(key, 0) + 1
            # optional: compare to expected chart label
            expected = (pod.metadata.labels or {}).get("app.kubernetes.io/service-account")
            if expected and expected != sa_name:
                mismatches.append({
                    "pod": pod.metadata.name,
                    "namespace": ns,
                    "uses": sa_name,
                    "label_expects": expected,
                })
    return usage, mismatches

Join annotated SAs to usage:

def unused_irsa(annotated, usage):
    return [
        r for r in annotated
        if usage.get((r["namespace"], r["name"]), 0) == 0
    ]

def pods_on_default_in_irsa_ns(v1, annotated_namespaces):
    offenders = []
    for ns in sorted(set(annotated_namespaces)):
        for pod in v1.list_namespaced_pod(ns).items:
            if pod.status.phase in ("Succeeded", "Failed"):
                continue
            sa_name = pod.spec.service_account_name or "default"
            if sa_name == "default":
                offenders.append(f"{ns}/{pod.metadata.name}")
    return offenders

This is the census that closes the Ops loop: annotation present, Pod selected, caller identity then verifiable with aws sts get-caller-identity inside one sample Pod.

§IV.B — Pagination and large clusters

list_namespaced_service_account is fine for a handful of namespaces. Cluster-wide inventory should use the list continue token:

def list_all_sa(v1, namespace):
    items = []
    _continue = None
    while True:
        kwargs = {"limit": 100}
        if _continue:
            kwargs["_continue"] = _continue
        resp = v1.list_namespaced_service_account(namespace, **kwargs)
        items.extend(resp.items)
        _continue = resp.metadata._continue
        if not _continue:
            break
    return items

Same pattern for Pods. Watch API is the wrong tool for a point-in-time census; list+continue is enough and cheaper on the apiserver than an open watch.

When the platform runs hundreds of namespaces, drive the outer loop from a namespace allowlist file committed next to the tool, not from list_namespace without filters. IRSA census for kube-system addon ServiceAccounts is usually noise unless you are auditing the CNI or cluster-autoscaler roles on purpose.

§IV.C — Cross-check annotation against IAM (optional second hop)

If the census host has sts:GetCallerIdentity and iam:GetRole read, you can verify the role exists and that its trust policy contains the expected sub:

import boto3, json

def trust_subs(role_name):
    iam = boto3.client("iam")
    role = iam.get_role(RoleName=role_name)
    doc = role["Role"]["AssumeRolePolicyDocument"]
    if isinstance(doc, str):
        doc = json.loads(doc)
    subs = []
    for stmt in doc.get("Statement", []):
        cond = stmt.get("Condition", {}).get("StringEquals", {})
        for k, v in cond.items():
            if k.endswith(":sub"):
                subs.append(v if isinstance(v, str) else tuple(v))
    return subs

Parse the role name from the ARN (arn.split("/")[-1]). Compare system:serviceaccount:{ns}:{name} to trust sub. This hop is optional because many CI identities must not read IAM. When available, it catches the Ops failure mode where Kubernetes is correct and the trust policy still names last week's ServiceAccount.

Never grant the census role permission to update trust policies. Read-only inventory stays read-only.

§V.B — Report schema for Maghrib and CI

Stable JSON keys so Maghrib quiz authors and CI dashboards do not scrape prose:

{
  "cluster": "prod-use1",
  "namespaces": ["billing", "payments"],
  "annotated": [
    {"namespace": "billing", "name": "invoice-reader", "role_arn": "arn:aws:iam::111122223333:role/invoice-reader-irsa", "pod_count": 2}
  ],
  "bad_arn": [],
  "unused_irsa": [],
  "default_pods": ["payments/worker-7d9f8"],
  "defaults_automount_true": ["payments/default"]
}

defaults_automount_true is the CKS-flavored companion signal: Kubestronaut wants default automount off. The census can flag it without fixing it.

§V.C — Relation to the 08-04 authorization lesson

08-04 used SelfSubjectAccessReview to preflight whether the running identity could perform a verb. Today's census assumes RBAC was already granted to the census SA. If list calls return 403, fail with exit 2 and print the namespace+resource. Do not catch ApiException and continue with empty lists; silent empty looks like a clean cluster.

Projected token refresh from 08-04 still matters when the census Job runs longer than the token lifetime. For a one-shot Job under ten minutes, the default projected mount is enough. For a long-lived Deployment that rescans hourly, use the kubernetes client's refresh behavior on in-cluster config and keep the Job schedule instead of a forever loop when you can.

§V — Optional: read the projected token audience

When the census Job itself runs in-cluster, you can show what IRSA injection looks like without calling AWS:

from pathlib import Path

def token_paths():
    root = Path("/var/run/secrets")
    hits = []
    if not root.exists():
        return hits
    for path in root.rglob("*"):
        if path.is_file() and "token" in path.name:
            hits.append(str(path))
    return hits

Do not print token contents to logs. Presence and path layout are enough for a hygiene report. Bound token shape and expirationSeconds belong to the Cert lesson and Bootcamp question 30.

§VI — CLI shape for operators

python irsa_census.py --namespaces billing,payments --json

Exit codes:

• 0 — every annotated SA has ≥1 running/pending Pod; no bad ARNs; no running Pods on default in those namespaces.
• 1 — unused IRSA SAs or default-Pod offenders (policy miss).
• 2 — API/list errors (RBAC or connectivity).

Emit JSON for CI and a short human summary for chat:

IRSA census billing,payments
  annotated: 4  bad_arn: 0  unused: 1  default_pods: 2
  unused: billing/legacy-reader
  default_pods: payments/worker-7d9f8, payments/worker-7d9f8-debug

§VI.B — Full script skeleton

#!/usr/bin/env python3
import argparse, json, re, sys
from kubernetes import client, config
from kubernetes.client.rest import ApiException

IRSA_KEY = "eks.amazonaws.com/role-arn"
ARN_RE = re.compile(r"^arn:aws:iam::\d{12}:role/[\w+=,.@-]+$")

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--namespaces", required=True)
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--in-cluster", action="store_true")
    args = ap.parse_args()
    if args.in_cluster:
        config.load_incluster_config()
    else:
        config.load_kube_config()
    v1 = client.CoreV1Api()
    namespaces = [n.strip() for n in args.namespaces.split(",") if n.strip()]
    try:
        rows = []
        usage = {}
        default_pods = []
        for ns in namespaces:
            for sa in v1.list_namespaced_service_account(ns).items:
                ann = sa.metadata.annotations or {}
                rows.append({
                    "namespace": ns,
                    "name": sa.metadata.name,
                    "role_arn": ann.get(IRSA_KEY),
                    "automount": sa.automount_service_account_token,
                })
            for pod in v1.list_namespaced_pod(ns).items:
                if pod.status.phase in ("Succeeded", "Failed"):
                    continue
                sa_name = pod.spec.service_account_name or "default"
                usage[(ns, sa_name)] = usage.get((ns, sa_name), 0) + 1
                if sa_name == "default":
                    default_pods.append(f"{ns}/{pod.metadata.name}")
    except ApiException as e:
        print(f"api_error status={e.status} reason={e.reason}", file=sys.stderr)
        sys.exit(2)

    annotated, bad_arn, unused = [], [], []
    defaults_automount_true = []
    for r in rows:
        if r["name"] == "default":
            if r["automount"] is not False:
                defaults_automount_true.append(f"{r['namespace']}/default")
            continue
        if not r["role_arn"]:
            continue
        if not ARN_RE.match(r["role_arn"]):
            bad_arn.append(r)
            continue
        r = dict(r)
        r["pod_count"] = usage.get((r["namespace"], r["name"]), 0)
        annotated.append(r)
        if r["pod_count"] == 0:
            unused.append(r)

    report = {
        "namespaces": namespaces,
        "annotated": annotated,
        "bad_arn": bad_arn,
        "unused_irsa": unused,
        "default_pods": default_pods,
        "defaults_automount_true": defaults_automount_true,
    }
    if args.json:
        print(json.dumps(report, indent=2))
    else:
        print(
            f"annotated={len(annotated)} bad_arn={len(bad_arn)} "
            f"unused={len(unused)} default_pods={len(default_pods)}"
        )
    if bad_arn or unused or default_pods:
        sys.exit(1)
    sys.exit(0)

if __name__ == "__main__":
    main()

Ship this as a Job manifest in the same directory as the chart that owns IRSA ServiceAccounts. Schedule it after sync waves so annotation and Pod creation have both landed.

§VII — Tests without a cluster

Stub CoreV1Api with simple namespaces objects. Feed two ServiceAccounts (one annotated, one not) and three Pods (one correct, one default, one Succeeded). Assert classify buckets and unused list. Keep fixtures in-repo so the census does not depend on a live EKS control plane for unit proof.

§VIII — Close

The Python client does not assume the IAM role for you. It proves the Kubernetes objects line up so the AWS SDK inside the workload Pod can. Census annotations, join to Pod serviceAccountName, fail CI on unused IRSA SAs and on workloads still riding default.

Examine the join before you chase STS errors in CloudTrail.

Related

• EKS IRSA Ops lesson
• Python in-cluster config and projected tokens
• Cross-References/Archmagus-Stack — Polyglot-Dev Python hub
• Grounding: KUR Ch.14 Service Account Management; Bootcamp EKS.md IRSA