Python and the cloud SDKs, hands-on

Python for security automation

A security script answers a question about a whole estate, such as which access keys are stale or which firewall rules let in the whole internet, by calling the cloud provider's API (application programming interface) in a loop and printing what matched. In Python those calls go through the provider's SDK (software development kit): boto3 for AWS (Amazon Web Services), the Azure SDK for Python, and the Google Cloud client libraries. The learning path asks for one scripting language, Python by default, and sets the bar at reading and changing other people's scripts. This page is pitched at that bar. It is not a Python course: it explains each piece of Python as the scripts use it.

The Python is the easy half. The harder half is that the SDKs make it easy to ask a narrower question than you meant and get a confident answer to it. A list call returns one page and mentions the rest in a field nobody reads. A client talks to one Region. An except written for one expected error also swallows AccessDenied. The credentials come from somewhere you did not expect. An organization-wide loop skips the accounts it could not get into. None of these raises an error. Each one produces a shorter list, and a short list of findings looks like good news.

boto3 comes first because it has the most of these traps and because the lab can run it for free. Where Azure or Google Cloud has the same trap, a short paragraph says what it looks like there.

Every result has a scope

Read every script's output as a claim with five conditions attached: as this identity, on the pages it read, in these Regions, in these accounts, among the calls that worked. An empty list is good news only when all five are what you meant. Each step of the walk-through takes one condition, runs a script that gets it wrong without complaint, and then changes the script so that it either gets it right or says what it could not check.

First, the Python those scripts are made of. Every boto3 call returns the API's answer as plain Python data. This block needs nothing installed:

python3 - <<'PY'
response = {
    "SecurityGroups": [
        {"GroupName": "debug-ssh", "IpPermissions": [
            {"IpProtocol": "tcp", "FromPort": 22, "ToPort": 22, "IpRanges": [{"CidrIp": "0.0.0.0/0"}]}]},
        {"GroupName": "office-ssh", "IpPermissions": []},
    ],
    "NextToken": "page-2-starts-here",
}
for group in response["SecurityGroups"]:
    rules = group["IpPermissions"]
    print(f"{group['GroupName']}: {len(rules)} inbound rule(s)")
print("more pages?", "NextToken" in response)
print("Marker:", response.get("Marker"))
PY

debug-ssh: 1 inbound rule(s)
office-ssh: 0 inbound rule(s)
more pages? True
Marker: None

That is the shape of a real describe_security_groups answer, cut down. Curly braces make a dictionary, which maps names (keys) to values, and square brackets make a list. response["SecurityGroups"] reads a key and stops the script with a KeyError if it is missing; response.get("Marker") reads one and gives None, Python's "nothing", if it is missing. for runs the indented lines once per item, and in Python the indentation is what marks the lines that belong to the loop. An f"..." string fills in whatever is in {braces}, and in asks "is this key here?". The last two lines show the first silent failure: the answer says another page exists, and nothing makes you read it.

Walk-through: a pretend organization, five silent failures

The lab is an AWS organization running inside moto, an open-source AWS emulator for Python, on your computer. You are in the management account, 123456789012, and the organization has three member accounts: prod, dev and sandbox. The management account has 137 IAM (Identity and Access Management) roles, six users with access keys of various ages, security groups in four Regions (one of them an opt-in Region), and four S3 (Simple Storage Service) buckets, one of which your audit identity is not allowed to read. You audit it as a user called auditor.

1. Install boto3 and moto

Run python3 --version first. boto3 and moto need Python 3.10 or newer, and the python3 that comes with macOS is older. If yours is, install a newer Python from python.org or your package manager and use its name (for example python3.12) instead of python3 on the second line.

mkdir -p ~/sdk-lab && cd ~/sdk-lab
python3 -m venv .venv && . .venv/bin/activate
pip install --quiet 'boto3==1.43.123' 'moto[server]==5.2.5'
python3 -c "import boto3, moto; print(boto3.__version__, moto.__version__)"

1.43.123 5.2.5

The second line makes a private Python for the lab (a virtual environment) and switches to it, so nothing is installed system-wide. In a new terminal, cd ~/sdk-lab && . .venv/bin/activate switches back. Keep the version pins: boto3 releases several times a week, and moto's behaviour changes between releases. The quotes around moto[server] matter in zsh, which otherwise reads the square brackets as a file-name pattern and refuses the command.

2. Start the lab

Paste the whole block, including the final EOF line. It writes lab.py and starts it. You do not need to follow every line: it is scaffolding, and none of it appears in the scripts you write.

cat > lab.py <<'EOF'
"""lab.py: a pretend AWS organization on your own computer.

Starts moto, an AWS emulator, at http://localhost:8101, fills it with the
accounts, users, security groups and buckets this guide audits, then waits.
Ctrl-C stops it, and everything in it is forgotten. Nothing here talks to AWS.
"""
import datetime, json, logging, pathlib, time
import boto3
import moto.organizations.utils
from moto import settings
from moto.core.responses import ActionResult
from moto.core.utils import utcnow
from moto.ec2.models.availability_zones_and_regions import Region, RegionsAndZonesBackend
from moto.ec2.responses.availability_zones_and_regions import AvailabilityZonesAndRegions
from moto.iam.models import AccessKeyLastUsed, iam_backends
from moto.organizations.models import organizations_backends
from moto.server import ThreadedMotoServer

URL = "http://localhost:8101"
MGMT, PROD, DEV, SANDBOX = "123456789012", "111111111111", "222222222222", "333333333333"
HERE = pathlib.Path(__file__).resolve().parent

# Moto lists every Region as usable. A real account can use the 17 Regions
# AWS enables by default plus any opt-in Region someone has turned on, so
# the lab copies that: me-central-1 (UAE) is opted in, other opt-in Regions are not.
DEFAULT_ON = {"us-east-1", "us-east-2", "us-west-1", "us-west-2", "ca-central-1",
              "sa-east-1", "eu-west-1", "eu-west-2", "eu-west-3", "eu-central-1",
              "eu-north-1", "ap-south-1", "ap-northeast-1", "ap-northeast-2",
              "ap-northeast-3", "ap-southeast-1", "ap-southeast-2"}
RegionsAndZonesBackend.regions = [
    Region(r, f"ec2.{r}.amazonaws.com",
           "opt-in-not-required" if r in DEFAULT_ON else
           "opted-in" if r == "me-central-1" else "not-opted-in")
    for r in boto3.session.Session().get_available_regions("ec2")]

def describe_regions(self):
    regions = self.ec2_backend.describe_regions(self._get_param("RegionNames", []))
    if not self._get_bool_param("AllRegions", False):
        regions = [r for r in regions if r.opt_in_status != "not-opted-in"]
    return ActionResult({"Regions": regions})
AvailabilityZonesAndRegions.describe_regions = describe_regions

logging.getLogger("werkzeug").setLevel(logging.ERROR)
ThreadedMotoServer(ip_address="127.0.0.1", port=8101, verbose=False).start()

def client(service, region="us-east-1", creds=None):
    creds = creds or {"aws_access_key_id": "lab", "aws_secret_access_key": "lab"}
    return boto3.client(service, region_name=region, endpoint_url=URL, **creds)

def into(account):
    c = client("sts").assume_role(RoleArn=f"arn:aws:iam::{account}:role/OrganizationAccountAccessRole",
                                  RoleSessionName="lab-setup")["Credentials"]
    return {"aws_access_key_id": c["AccessKeyId"], "aws_secret_access_key": c["SecretAccessKey"],
            "aws_session_token": c["SessionToken"]}

def policy(*statements):
    return json.dumps({"Version": "2012-10-17", "Statement": list(statements)})

READ = {"Effect": "Allow", "Resource": "*",
        "Action": ["iam:Get*", "iam:List*", "ec2:Describe*", "s3:Get*", "s3:List*",
                   "organizations:Describe*", "organizations:List*"]}

def open_group(name, port, region, creds=None, cidr="0.0.0.0/0"):
    ec2 = client("ec2", region, creds)
    gid = ec2.create_security_group(GroupName=name, Description=name)["GroupId"]
    rng = {"Ipv6Ranges": [{"CidrIpv6": cidr}]} if ":" in cidr else {"IpRanges": [{"CidrIp": cidr}]}
    perm = {"IpProtocol": "-1"} if port is None else {"IpProtocol": "tcp", "FromPort": port, "ToPort": port}
    ec2.authorize_security_group_ingress(GroupId=gid, IpPermissions=[{**perm, **rng}])

# The organization: you are in the management account, 123456789012.
org = client("organizations")
org.create_organization(FeatureSet="ALL")
organizations_backends[MGMT]["aws"].accounts[0].name = "management"
moto.organizations.utils.make_random_account_id = iter([PROD, DEV, SANDBOX]).__next__
for name in ["prod", "dev", "sandbox"]:
    org.create_account(Email=f"{name}@example.com", AccountName=name)

# An audit role in each member account, trusting the auditor user.
for account in [PROD, DEV, SANDBOX]:
    iam = client("iam", creds=into(account))
    iam.create_role(RoleName="audit-readonly", AssumeRolePolicyDocument=policy(
        {"Effect": "Allow", "Action": "sts:AssumeRole",
         "Principal": {"AWS": f"arn:aws:iam::{MGMT}:user/auditor"}}))
    iam.put_role_policy(RoleName="audit-readonly", PolicyName="read", PolicyDocument=policy(READ))
open_group("prod-bastion", 22, "us-west-2", into(PROD))
open_group("sandbox-test", None, "us-east-1", into(SANDBOX))

# IAM in the management account: 137 roles, so the list is longer than one page.
iam = client("iam")
for n in range(1, 138):
    iam.create_role(RoleName=f"workload-{n:03d}", AssumeRolePolicyDocument=policy(
        {"Effect": "Allow", "Action": "sts:AssumeRole", "Principal": {"Service": "ec2.amazonaws.com"}}))

# Users and their keys. Moto cannot create old keys, so the lab ages them:
# (user, days since the key was created, days since it was last used, active?)
KEYS = [("alice", 400, 2, True), ("bob", 300, 120, True), ("carol", 5, None, True),
        ("dave", 150, None, True), ("dave", 500, 400, False),
        ("build-bot", 200, 0, True), ("auditor", 0, None, True)]
creds = {}
for user, created, used, active in KEYS:
    if user not in iam_backends[MGMT]["global"].users:
        iam.create_user(UserName=user)
    key = iam.create_access_key(UserName=user)["AccessKey"]
    creds[user] = (key["AccessKeyId"], key["SecretAccessKey"])
    if not active:
        iam.update_access_key(UserName=user, AccessKeyId=key["AccessKeyId"], Status="Inactive")
    stored = iam_backends[MGMT]["global"].users[user].access_keys[-1]
    stored.create_date = utcnow() - datetime.timedelta(days=created)
    if used is not None:
        stored.last_used = AccessKeyLastUsed(utcnow() - datetime.timedelta(days=used), "iam", "us-east-1")
iam.put_user_policy(UserName="auditor", PolicyName="audit", PolicyDocument=policy(
    READ,
    # The auditor may assume the audit role in prod and dev. sandbox joined the
    # organization after this list was written, and nobody added it.
    {"Effect": "Allow", "Action": "sts:AssumeRole",
     "Resource": [f"arn:aws:iam::{PROD}:role/audit-readonly", f"arn:aws:iam::{DEV}:role/audit-readonly"]},
    {"Effect": "Deny", "Action": "s3:*", "Resource": ["arn:aws:s3:::csoh-lab-payroll-exports",
                                                      "arn:aws:s3:::csoh-lab-payroll-exports/*"]}))
iam.put_user_policy(UserName="build-bot", PolicyName="ci", PolicyDocument=policy(
    {"Effect": "Allow", "Action": "s3:PutObject", "Resource": "arn:aws:s3:::csoh-lab-build-artifacts/*"}))

# Security groups in four Regions, one of them opt-in.
open_group("web-public", 443, "us-east-1")
open_group("debug-ssh", 22, "us-east-1")
open_group("legacy-rdp", 3389, "eu-west-1")
open_group("ipv6-everything", None, "ap-southeast-2", cidr="::/0")
open_group("temp-postgres", 5432, "me-central-1")
client("ec2").create_security_group(GroupName="office-ssh", Description="office-ssh")

# Buckets: Block Public Access on, off, partly on, and one the auditor may not read.
s3 = client("s3")
ALL_ON = dict.fromkeys(["BlockPublicAcls", "IgnorePublicAcls", "BlockPublicPolicy", "RestrictPublicBuckets"], True)
for bucket, bpa in [("csoh-lab-cloudtrail", ALL_ON), ("csoh-lab-web-assets", None),
                    ("csoh-lab-build-artifacts", {**ALL_ON, "BlockPublicPolicy": False}),
                    ("csoh-lab-payroll-exports", ALL_ON)]:
    s3.create_bucket(Bucket=bucket)
    if bpa:
        s3.put_public_access_block(Bucket=bucket, PublicAccessBlockConfiguration=bpa)
s3.put_bucket_policy(Bucket="csoh-lab-cloudtrail", Policy=policy(
    {"Effect": "Allow", "Principal": {"Service": "cloudtrail.amazonaws.com"},
     "Action": "s3:PutObject", "Resource": "arn:aws:s3:::csoh-lab-cloudtrail/*"}))

# From here on every request is checked against the caller's IAM policies.
settings.INITIAL_NO_AUTH_ACTION_COUNT = 0

(HERE / "aws").mkdir(exist_ok=True)
(HERE / "aws" / "config").write_text(
    "[default]\nregion = us-east-1\n\n[profile auditor]\nregion = us-east-1\n")
(HERE / "aws" / "credentials").write_text("".join(
    f"[{profile}]\naws_access_key_id = {creds[user][0]}\naws_secret_access_key = {creds[user][1]}\n\n"
    for profile, user in [("default", "alice"), ("auditor", "auditor")]))
(HERE / "build-bot.env").write_text(
    f"export AWS_ACCESS_KEY_ID={creds['build-bot'][0]}\nexport AWS_SECRET_ACCESS_KEY={creds['build-bot'][1]}\n")
(HERE / "lab.env").write_text(
    "unset AWS_PROFILE AWS_ACCESS_KEY_ID AWS_SECRET_ACCESS_KEY AWS_SESSION_TOKEN AWS_REGION AWS_DEFAULT_REGION\n"
    f"export AWS_ENDPOINT_URL={URL}\n"
    f"export AWS_CONFIG_FILE={HERE}/aws/config\n"
    f"export AWS_SHARED_CREDENTIALS_FILE={HERE}/aws/credentials\n")
print(f"Lab running at {URL}. Leave this terminal open; Ctrl-C stops it.", flush=True)
try:
    while True:
        time.sleep(3600)
except KeyboardInterrupt:
    print("Lab stopped.")
EOF
python3 lab.py

Lab running at http://localhost:8101. Leave this terminal open; Ctrl-C stops it.

The lab listens only on your own computer, at localhost port 8101 (moto's usual port, 5000, is often taken on a Mac by macOS's AirPlay Receiver). It builds the organization, then writes four files beside itself: an AWS config file and credentials file under aws/, build-bot.env, which step 3 uses, and lab.env, which points every AWS tool in a terminal at the lab. Three parts of it change moto rather than fill it: moto lists every Region as usable and ignores the difference between default and opt-in Regions, it cannot create an access key that is already old, and it gives new accounts random IDs. The lab corrects all three so the walk-through behaves like a real account and your output matches this page. The last setting before the files makes moto check every request against the caller's IAM policies, so an AccessDenied in the lab is a real refusal.

3. Find out who you are (identity)

Leave the lab running and open a second terminal. Everything from here on runs there. The last command on the first line loads lab.env, which clears any AWS settings already in that terminal and points boto3 at the lab. Skip it and these scripts read whatever AWS account your own credentials point at.

cd ~/sdk-lab && . .venv/bin/activate && . ./lab.env
cat > whoami.py <<'EOF'
import boto3

session = boto3.Session()
me = session.client("sts").get_caller_identity()
print("account: ", me["Account"])
print("identity:", me["Arn"])
print("found in:", session.get_credentials().method)
print("region:  ", session.region_name)
EOF
python3 whoami.py

account:  123456789012
identity: arn:aws:iam::123456789012:user/alice
found in: shared-credentials-file
region:   us-east-1

get_caller_identity asks AWS STS (Security Token Service) which identity signed the request. It needs no permissions, so it works for any valid credentials, and every script that reports on an account should call it first and print the answer. This one ran as alice, the [default] profile in the credentials file, and not as the auditor. found in is where boto3 found the credentials, and region is the Region its clients will talk to unless told otherwise. Ask for the auditor's profile:

AWS_PROFILE=auditor python3 whoami.py

account:  123456789012
identity: arn:aws:iam::123456789012:user/auditor
found in: shared-credentials-file
region:   us-east-1

That is the identity the rest of the page needs. Now repeat a common afternoon: earlier you were debugging the CI (continuous integration) pipeline and exported its access key into this terminal, and now you run the audit with the auditor's profile.

. ./build-bot.env
AWS_PROFILE=auditor python3 whoami.py

account:  123456789012
identity: arn:aws:iam::123456789012:user/build-bot
found in: env
region:   us-east-1

You asked for the auditor and got build-bot. boto3 tries a fixed list of places for credentials and stops at the first that has some. Environment variables (AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY) come third in boto3's credential order, ahead of the profiles in the credentials file, and AWS_PROFILE does not override them. Passing the profile in code, boto3.Session(profile_name="auditor"), does override them: botocore removes the environment-variable step when a profile is given that way. The AWS CLI (command-line interface) behaves the same way, and --profile on its command line does override environment keys. In this lab the mix-up shows at once, because build-bot may only upload to one bucket, so any audit that lists anything fails. A build identity with read access everywhere would produce a complete, plausible report about the wrong permissions. Clear the leftover keys and set the auditor for the rest of the page:

unset AWS_ACCESS_KEY_ID AWS_SECRET_ACCESS_KEY
export AWS_PROFILE=auditor
python3 whoami.py

account:  123456789012
identity: arn:aws:iam::123456789012:user/auditor
found in: shared-credentials-file
region:   us-east-1

A script that should only ever run against one account can go further and stop when the account ID is not the one it expects.

On Azure, DefaultAzureCredential from the azure-identity library has the same shape of chain: a service principal set in environment variables comes first, then workload identity, then managed identity, and your Azure CLI login comes well after those, so leftover service-principal variables such as AZURE_CLIENT_SECRET win over the account you think you logged in with, and so does a virtual machine's managed identity when the script runs on one. At INFO log level it says which one it used, in a line that reads DefaultAzureCredential acquired a token from ....

On Google Cloud, Application Default Credentials (ADC) take the credentials file named by GOOGLE_APPLICATION_CREDENTIALS before the credentials that gcloud auth application-default login saved, and the account you signed in with through plain gcloud auth login is not one of the places ADC looks. google.auth.default() returns the credentials it chose and their project, so print them.

4. Read every page

A list call returns one page of results. IAM's ListRoles returns up to 100 roles per call unless asked for a different number, says whether more exist in IsTruncated, and hands back a Marker to continue from. Count the roles three ways:

cat > roles.py <<'EOF'
import boto3

iam = boto3.client("iam")

first = iam.list_roles()
print("one call:         ", len(first["Roles"]), "roles, IsTruncated =", first["IsTruncated"])

roles, extra = [], {}
while True:
    page = iam.list_roles(**extra)
    roles.extend(page["Roles"])
    if "NextToken" not in page:
        break
    extra = {"NextToken": page["NextToken"]}
print("hand-written loop:", len(roles), "roles")

roles = []
for page in iam.get_paginator("list_roles").paginate():
    roles.extend(page["Roles"])
print("paginator:        ", len(roles), "roles")
EOF
python3 roles.py

one call:          100 roles, IsTruncated = True
hand-written loop: 100 roles
paginator:         137 roles

The single call stops at 100 and says so in a field the script never checks. The hand-written loop is the kind people copy from an EC2 (Elastic Compute Cloud) script: EC2 and many other services call their continuation field NextToken, IAM calls it Marker, so the loop sees no NextToken after the first page and stops, as if it had finished. (while True: repeats until break, and **extra passes the dictionary's contents as named arguments.) A paginator knows each operation's own field names and keeps calling until the last page, so get_paginator(...).paginate() is the habit: when client.can_paginate("operation_name") is True, use it.

"No more pages" is a signal only the API can give, and two details make guessing it worse. IAM's documentation says it may return fewer than the page size even when more results exist, so a short page is not a last page. AWS Organizations goes further: its list calls can occasionally return an empty page while more results remain, and only a missing NextToken means you are done. A loop that stops on an empty page is wrong there, and a paginator is not.

The Azure SDK and Google's google-cloud-* libraries page for you: the ItemPaged object an Azure list method returns follows nextLink as you loop over it, and Google's pagers follow next_page_token. The one-page bug comes back when a script skips the SDK, calls the REST (representational state transfer) API with requests and ignores nextLink or nextPageToken, or uses Google's older googleapiclient discovery client, where each execute() returns one page.

5. Cover every enabled Region

A boto3 client talks to one Region: the region_name you pass, or else the AWS_DEFAULT_REGION environment variable, or else the profile's region setting. boto3 ignores AWS_REGION, which the AWS CLI does read, so in a terminal where only AWS_REGION is set, the CLI and your script can be looking at different Regions. Security groups, like most resources, live in a Region. This script lists the non-default groups in the client's Region, then in every Region the account has enabled:

cat > regions.py <<'EOF'
import boto3

def group_names(ec2):
    names = []
    for page in ec2.get_paginator("describe_security_groups").paginate():
        for group in page["SecurityGroups"]:
            if group["GroupName"] != "default":
                names.append(group["GroupName"])
    return names

ec2 = boto3.client("ec2")
print(f"{ec2.meta.region_name} only: {group_names(ec2)}")

enabled = [r["RegionName"] for r in ec2.describe_regions()["Regions"]]
print(len(enabled), "Regions enabled for this account:")
for region in sorted(enabled):
    names = group_names(boto3.client("ec2", region_name=region))
    if names:
        print(f"  {region}: {names}")
EOF
python3 regions.py

us-east-1 only: ['web-public', 'debug-ssh', 'office-ssh']
18 Regions enabled for this account:
  ap-southeast-2: ['ipv6-everything']
  eu-west-1: ['legacy-rdp']
  me-central-1: ['temp-postgres']
  us-east-1: ['web-public', 'debug-ssh', 'office-ssh']

def defines a function, a named block that runs when called and hands back what return gives it. Here it lets the same code run once per Region. The first line is what a script that never names a Region sees: three groups, and nothing about the three Regions it did not look at. describe_regions returns the Regions enabled for the account, and AllRegions=True adds the disabled ones, each with an OptInStatus.

One of the 18 needs explaining. AWS enables 17 Regions in every commercial account, the ones launched before 20 March 2019. Newer Regions are opt-in: disabled until someone turns them on, and then as usable as any other. The lab account has opted in to me-central-1 (United Arab Emirates), and a database port is open to the internet there. A script with a hard-coded list of the usual Regions never looks there, and neither does a reviewer who knows the account "only uses us-east-1". Ask the account, every run.

Not everything is regional. IAM and Organizations are global, so one client covers them, and S3's list_buckets returns every bucket whatever its Region. Azure draws the line elsewhere: a subscription is the unit, and calls such as network_security_groups.list_all() in azure-mgmt-network return a subscription's resources in every Azure region at once. On Google Cloud, VPC (virtual private cloud) firewall rules are global within a project, though organization, folder and network firewall policies also decide what gets in and have list calls of their own, and for regional and zonal resources the Compute library's aggregated_list covers every region and zone of one project. On both, the boundary a script forgets is the subscription or the project, which step 7 comes back to.

6. Let errors say what they could not check

When a call fails, boto3 raises ClientError, and the error code AWS sent is in err.response["Error"]["Code"]. Some codes are answers: NoSuchBucketPolicy means the bucket has no bucket policy. Others mean the script did not find out, such as AccessDenied or a throttling code after the retries run out. This script treats them alike:

cat > policies_naive.py <<'EOF'
import boto3

s3 = boto3.client("s3")
for bucket in s3.list_buckets()["Buckets"]:
    try:
        s3.get_bucket_policy(Bucket=bucket["Name"])
        print(bucket["Name"], "has a bucket policy")
    except Exception:
        print(bucket["Name"], "has no bucket policy")
EOF
python3 policies_naive.py

csoh-lab-cloudtrail has a bucket policy
csoh-lab-web-assets has no bucket policy
csoh-lab-build-artifacts has no bucket policy
csoh-lab-payroll-exports has no bucket policy

The last line is false. The auditor is not allowed to read anything about csoh-lab-payroll-exports, so the call was refused, and except Exception: turned the refusal into an answer. try: runs the indented lines, and when one of them raises an error Python jumps to the except block that matches it; Exception matches nearly everything, so no failure can get past it. Catch only ClientError, act on the codes that are answers, and keep a list of everything else:

cat > policies.py <<'EOF'
import sys
import boto3
from botocore.config import Config
from botocore.exceptions import ClientError

s3 = boto3.client("s3", config=Config(retries={"mode": "standard", "total_max_attempts": 10}))
has_policy, no_policy, not_checked = [], [], []

for page in s3.get_paginator("list_buckets").paginate():
    for bucket in page["Buckets"]:
        name = bucket["Name"]
        try:
            s3.get_bucket_policy(Bucket=name)
            has_policy.append(name)
        except ClientError as err:
            code = err.response["Error"]["Code"]
            if code == "NoSuchBucketPolicy":
                no_policy.append(name)
            else:
                not_checked.append(f"{name} ({code})")

print("has a policy:", has_policy)
print("no policy:   ", no_policy)
print("NOT CHECKED: ", not_checked)
sys.exit(1 if not_checked else 0)
EOF
python3 policies.py; echo "exit code: $?"

has a policy: ['csoh-lab-cloudtrail']
no policy:    ['csoh-lab-web-assets', 'csoh-lab-build-artifacts']
NOT CHECKED:  ['csoh-lab-payroll-exports (AccessDenied)']
exit code: 1

The bucket is now reported as not checked, with the reason, and the script exits with code 1, so a scheduler or CI job that runs it sees an incomplete result as a failure. sys.exit(1 if not_checked else 0) is a one-line if-else: 1 when the list has anything in it, 0 when it is empty. The bucket listing now uses a paginator too: AWS strongly recommends paginated ListBuckets requests and rejects unpaginated ones from accounts allowed more than 10,000 buckets.

The Config line sets the retry behaviour instead of leaving it to the default. In the version pinned here, boto3's default is legacy retry mode, which makes at most five attempts in all and retries a shorter list of throttling codes than standard mode does. In this botocore release a new environment variable can already switch the default, so setting the mode yourself keeps the behaviour from changing under you. Use total_max_attempts, which counts every attempt, rather than max_attempts, which in a Config counts only the retries. When the attempts run out, the throttling code (such as Throttling, RequestLimitExceeded or SlowDown) lands in the NOT CHECKED list with everything else.

Not every failure needs catching. Run the same script as alice, who has no permissions, and the very first call fails outside the try:

AWS_PROFILE=default python3 policies.py 2>&1 | tail -n 1

botocore.errorfactory.AccessDenied: An error occurred (AccessDenied) when calling the ListBuckets operation: Access Denied

A crash with a traceback is a loud failure, which is fine. The failure to fear is the quiet one. A network failure is loud in the same way: botocore raises it as a BotoCoreError, such as EndpointConnectionError, not as a ClientError, so this script stops on one instead of listing it. (2>&1 | tail -n 1 sends the error output through tail, which keeps only the last line of the traceback.) Azure has the same split: the SDK raises HttpResponseError from azure.core.exceptions with a status_code, 403 for a refusal and 404 for something that does not exist, while Google's client libraries raise exceptions from google.api_core.exceptions, such as Forbidden and PermissionDenied for a refusal and NotFound for a missing resource. A broad except turns all of them into "nothing found".

7. Cover every account in the organization

One identity normally sees one account. To audit an organization you list its accounts, then in each one assume a role that trusts your audit identity, and use the temporary credentials that assume_role returns. This script counts the IAM roles in every account:

cat > org_roles.py <<'EOF'
import sys
import boto3
from botocore.exceptions import ClientError

ROLE = "audit-readonly"
home = boto3.Session()
my_account = home.client("sts").get_caller_identity()["Account"]

def session_in(account_id):
    if account_id == my_account:
        return home
    creds = home.client("sts").assume_role(
        RoleArn=f"arn:aws:iam::{account_id}:role/{ROLE}",
        RoleSessionName="security-audit")["Credentials"]
    return boto3.Session(aws_access_key_id=creds["AccessKeyId"],
                         aws_secret_access_key=creds["SecretAccessKey"],
                         aws_session_token=creds["SessionToken"],
                         region_name=home.region_name)

def count_roles(session):
    iam = session.client("iam")
    return sum(len(page["Roles"]) for page in iam.get_paginator("list_roles").paginate())

not_checked = []
for page in home.client("organizations").get_paginator("list_accounts").paginate():
    for account in page["Accounts"]:
        label = f"{account['Name']} ({account['Id']})"
        if account["State"] != "ACTIVE":
            print(f"{label}: skipped, state is {account['State']}")
            continue
        try:
            print(f"{label}: {count_roles(session_in(account['Id']))} roles")
        except ClientError as err:
            not_checked.append(label)
            print(f"{label}: NOT CHECKED, {err.response['Error']['Code']}")

sys.exit(1 if not_checked else 0)
EOF
python3 org_roles.py; echo "exit code: $?"

management (123456789012): 137 roles
prod (111111111111): 2 roles
dev (222222222222): 2 roles
sandbox (333333333333): NOT CHECKED, AccessDenied
exit code: 1

session_in() returns a boto3 session for any account: the one you are in already, or a new session built from the role's temporary credentials. The auditor's own policy lets it assume the audit role in prod and dev only. sandbox joined later, nobody added it, so AWS refused the role and the script says so. AssumeRole is also refused when the role does not exist under that exact name (role names are case sensitive when you assume one), when its trust policy does not name you, or when you fail one of its conditions, and IAM's troubleshooting guide lists what to check. Write that loop with except Exception: continue and it prints three accounts and exits 0.

Three details from the AWS documentation shape this loop. First, read State and not Status: AWS said it would retire the account Status field from the Organizations API on 9 September 2026. A script that tests account.get("Status") != "ACTIVE" gets None once the field is gone, and its "skip inactive accounts" line then skips every account. Second, list_accounts works only from the management account or from a member account that is a delegated administrator, so an audit that runs from a separate security account needs that account registered as one. Third, accounts created through Organizations come with a role named OrganizationAccountAccessRole by default, but it grants full administrator access, and accounts invited into the organization do not get it unless someone creates it. Many organizations deploy their own read-only audit role to every account instead, as the lab does. Then every account that joins later needs the role, and in the lab it also needs adding to the auditor's policy.

On Azure, the subscription is the boundary: SubscriptionClient(credential).subscriptions.list() lists the subscriptions your credential can see, and the rest of a script runs per subscription. In azure-mgmt-resource 25.0.0 that client moved into its own package, azure-mgmt-resource-subscriptions, which azure-mgmt-resource does not install for you. Older scripts that import it from azure.mgmt.resource fail until you install that package and import from azure.mgmt.resource.subscriptions. On Google Cloud the project is the boundary, and ProjectsClient().search_projects() in google-cloud-resource-manager lists the projects you can see. On both clouds a subscription or project your identity cannot see is missing from that list, with no error at all, so compare the count with what the organization says it has.

Hands-on exercises

These are the questions people script most often. Write each one in ~/sdk-lab with the lab still running, as the auditor, and compare your output with the answer.

cat > stale_keys.py <<'EOF'
import datetime
import boto3

LIMIT = datetime.timedelta(days=90)
now = datetime.datetime.now(datetime.timezone.utc)
iam = boto3.client("iam")

for page in iam.get_paginator("list_users").paginate():
    for user in page["Users"]:
        name = user["UserName"]
        for key_page in iam.get_paginator("list_access_keys").paginate(UserName=name):
            for key in key_page["AccessKeyMetadata"]:
                if key["Status"] != "Active":
                    continue
                used = iam.get_access_key_last_used(AccessKeyId=key["AccessKeyId"])["AccessKeyLastUsed"]
                if "LastUsedDate" in used:
                    idle, how = now - used["LastUsedDate"], "last used"
                else:
                    idle, how = now - key["CreateDate"], "never used, created"
                if idle > LIMIT:
                    print(f"{name}: {how} {idle.days} days ago")
EOF
python3 stale_keys.py

bob: last used 120 days ago
dave: never used, created 150 days ago

cat > open_sgs.py <<'EOF'
import boto3

WORLD = {"0.0.0.0/0", "::/0"}

def describe(rule):
    if rule["IpProtocol"] == "-1":
        return "all traffic"
    low, high = rule.get("FromPort"), rule.get("ToPort")
    return f"{rule['IpProtocol']} {low if low == high else f'{low}-{high}'}"

def open_groups(session):
    found = []
    ec2 = session.client("ec2")
    for region in sorted(r["RegionName"] for r in ec2.describe_regions()["Regions"]):
        regional = session.client("ec2", region_name=region)
        for page in regional.get_paginator("describe_security_groups").paginate():
            for group in page["SecurityGroups"]:
                for rule in group["IpPermissions"]:
                    sources = [r["CidrIp"] for r in rule.get("IpRanges", [])]
                    sources += [r["CidrIpv6"] for r in rule.get("Ipv6Ranges", [])]
                    for source in WORLD.intersection(sources):
                        found.append(f"{region} {group['GroupName']} ({group['GroupId']}): "
                                     f"{describe(rule)} from {source}")
    return sorted(found)

if __name__ == "__main__":
    for line in open_groups(boto3.Session()):
        print(line)
EOF
python3 open_sgs.py

ap-southeast-2 ipv6-everything (sg-7febd3ce0102d6546): all traffic from ::/0
eu-west-1 legacy-rdp (sg-ddfad65ca6f049189): tcp 3389 from 0.0.0.0/0
me-central-1 temp-postgres (sg-be8767ea2416195ef): tcp 5432 from 0.0.0.0/0
us-east-1 debug-ssh (sg-c225ead533fca7440): tcp 22 from 0.0.0.0/0
us-east-1 web-public (sg-f3b1db538cdbb826c): tcp 443 from 0.0.0.0/0

cat > bpa.py <<'EOF'
import sys
import boto3
from botocore.exceptions import ClientError

SETTINGS = ["BlockPublicAcls", "IgnorePublicAcls", "BlockPublicPolicy", "RestrictPublicBuckets"]
session = boto3.Session()
account_id = session.client("sts").get_caller_identity()["Account"]

def missing(config):
    return [name for name in SETTINGS if not config.get(name)]

try:
    account = session.client("s3control").get_public_access_block(AccountId=account_id)
    account_off = missing(account["PublicAccessBlockConfiguration"])
except ClientError as err:
    if err.response["Error"]["Code"] != "NoSuchPublicAccessBlockConfiguration":
        raise
    account_off = SETTINGS
print("account level, off:", ", ".join(account_off) or "none")

s3 = session.client("s3")
not_checked = []
for page in s3.get_paginator("list_buckets").paginate():
    for bucket in page["Buckets"]:
        name = bucket["Name"]
        try:
            config = s3.get_public_access_block(Bucket=name)["PublicAccessBlockConfiguration"]
            bucket_off = missing(config)
        except ClientError as err:
            code = err.response["Error"]["Code"]
            if code != "NoSuchPublicAccessBlockConfiguration":
                not_checked.append(f"{name} ({code})")
                continue
            bucket_off = SETTINGS
        off_at_both = [setting for setting in bucket_off if setting in account_off]
        if off_at_both:
            print(f"{name}: off at both levels: {', '.join(off_at_both)}")

print("NOT CHECKED:", not_checked)
sys.exit(1 if not_checked else 0)
EOF
python3 bpa.py; echo "exit code: $?"

account level, off: BlockPublicAcls, IgnorePublicAcls, BlockPublicPolicy, RestrictPublicBuckets
csoh-lab-web-assets: off at both levels: BlockPublicAcls, IgnorePublicAcls, BlockPublicPolicy, RestrictPublicBuckets
csoh-lab-build-artifacts: off at both levels: BlockPublicPolicy
NOT CHECKED: ['csoh-lab-payroll-exports (AccessDenied)']
exit code: 1

cat > org_open_sgs.py <<'EOF'
import sys
import boto3
from botocore.exceptions import ClientError
from open_sgs import open_groups

home = boto3.Session()
my_account = home.client("sts").get_caller_identity()["Account"]
not_checked = []

for page in home.client("organizations").get_paginator("list_accounts").paginate():
    for account in page["Accounts"]:
        if account["State"] != "ACTIVE":
            continue
        try:
            session = home
            if account["Id"] != my_account:
                creds = home.client("sts").assume_role(
                    RoleArn=f"arn:aws:iam::{account['Id']}:role/audit-readonly",
                    RoleSessionName="security-audit")["Credentials"]
                session = boto3.Session(aws_access_key_id=creds["AccessKeyId"],
                                        aws_secret_access_key=creds["SecretAccessKey"],
                                        aws_session_token=creds["SessionToken"],
                                        region_name=home.region_name)
            for line in open_groups(session):
                print(account["Name"], line)
        except ClientError as err:
            not_checked.append(f"{account['Name']} ({err.response['Error']['Code']})")

print("NOT CHECKED:", not_checked)
sys.exit(1 if not_checked else 0)
EOF
python3 org_open_sgs.py; echo "exit code: $?"

management ap-southeast-2 ipv6-everything (sg-7febd3ce0102d6546): all traffic from ::/0
management eu-west-1 legacy-rdp (sg-ddfad65ca6f049189): tcp 3389 from 0.0.0.0/0
management me-central-1 temp-postgres (sg-be8767ea2416195ef): tcp 5432 from 0.0.0.0/0
management us-east-1 debug-ssh (sg-c225ead533fca7440): tcp 22 from 0.0.0.0/0
management us-east-1 web-public (sg-f3b1db538cdbb826c): tcp 443 from 0.0.0.0/0
prod us-west-2 prod-bastion (sg-efdb8f9beeb14faa8): tcp 22 from 0.0.0.0/0
NOT CHECKED: ['sandbox (AccessDenied)']
exit code: 1

Common mistakes

• Calling a list operation once. IAM's list calls page at 100, Organizations' ListAccounts at 20 at most, and other services at sizes of their own. Use a paginator wherever can_paginate() says one exists.
• Hand-writing a pagination loop. Each service names its continuation field: Marker in IAM and NextToken in EC2 and Organizations, and S3 varies by operation. S3 uses ContinuationToken for ListBuckets but NextContinuationToken for ListObjectsV2. A loop that checks the wrong one stops after the first page, and one that stops on an empty or short page stops early.
• Trusting AWS_PROFILE with keys in the environment. Environment variables win over it. Print get_caller_identity() at the top of the script, or pass profile_name in code.
• Looking in one Region, or in a fixed list. Ask describe_regions() every run, so opt-in Regions someone enabled last month are included.
• Setting AWS_REGION for a boto3 script. boto3 reads AWS_DEFAULT_REGION and ignores AWS_REGION, so the script quietly uses the profile's Region instead.
• except Exception: pass, or continue. It turns AccessDenied, throttling and network failures into "nothing found". Catch ClientError, handle the codes that are answers, and report the rest.
• Reading a field that might be missing with .get() and treating None as "fine". A never-used access key has no LastUsedDate, a rule for all traffic has no FromPort, and AWS said it would retire the account Status field. Decide what a missing value means before you write the comparison.
• Exiting 0 from an incomplete audit. A scheduled job can only see the exit code. Exit non-zero whenever anything was not checked.
• Expecting a list call to return everything about each item. list_roles leaves out each role's last-used date, tags and permissions boundary; get_role returns them, one role per call.