iam-cross-account-externalid-sourcearn

#!/usr/bin/env python3
"""Provision the cross-account trust chain in localstack.

Order matters because of inter-resource references:
  1. Generate two distinct ExternalId values (>=32 chars each)
  2. Create IAM users (agent-runner, attacker-runner)
  3. Create CMK + alias (default policy initially; we replace it after orch role exists)
  4. Create OrchestratorRole (trust = agent-runner user) so we know its ARN
  5. Replace KMS key policy: root admin (specific actions + key arn) + orchestrator decrypt via secretsmanager
  6. Create both ExternalId secrets, encrypted with the CMK
  7. Attach resource policies to both secrets naming OrchestratorRole as principal
  8. Create both member roles with strict trust (orch principal + ExternalId + SourceArn)
  9. Inline policies on member roles: scoped to that tenant's bucket
 10. Inline policies on OrchestratorRole: AssumeRole on the two member arns; GetSecretValue on the two secret arns; kms:Decrypt on the CMK via secretsmanager
 11. Create the two tenant buckets
 12. Publish SSM pointers under /harbor/...
"""
import boto3
import json
import os
import secrets

REGION = "us-east-1"
ENDPOINT = os.environ.get("AWS_ENDPOINT_URL", "http://localstack:4566")

iam = boto3.client("iam", region_name=REGION, endpoint_url=ENDPOINT)
kms = boto3.client("kms", region_name=REGION, endpoint_url=ENDPOINT)
sm = boto3.client("secretsmanager", region_name=REGION, endpoint_url=ENDPOINT)
ssm = boto3.client("ssm", region_name=REGION, endpoint_url=ENDPOINT)
s3 = boto3.client("s3", region_name=REGION, endpoint_url=ENDPOINT)
sts = boto3.client("sts", region_name=REGION, endpoint_url=ENDPOINT)

ACCOUNT = sts.get_caller_identity()["Account"]


def log(msg):
    print(f"[build] {msg}", flush=True)


# 1) ExternalIds , distinct, random, >=32 chars
ext_a = secrets.token_urlsafe(48)
ext_b = secrets.token_urlsafe(48)
assert ext_a != ext_b
assert len(ext_a) >= 32 and len(ext_b) >= 32

# 2) IAM users
for u in ["agent-runner", "attacker-runner"]:
    try:
        iam.create_user(UserName=u)
        log(f"created user {u}")
    except iam.exceptions.EntityAlreadyExistsException:
        log(f"user {u} already exists")

agent_runner_arn = f"arn:aws:iam::{ACCOUNT}:user/agent-runner"

# 3) KMS key + alias (default policy on creation; replaced once we know orchestrator arn)
existing_alias = next(
    (a for a in kms.list_aliases().get("Aliases", []) if a["AliasName"] == "alias/cross-account-cmk"),
    None,
)
if existing_alias:
    key_id = existing_alias["TargetKeyId"]
    log(f"reusing kms key {key_id}")
else:
    res = kms.create_key(Description="cross-account external-id encryption")
    key_id = res["KeyMetadata"]["KeyId"]
    kms.create_alias(AliasName="alias/cross-account-cmk", TargetKeyId=key_id)
    log(f"created kms key {key_id} alias alias/cross-account-cmk")

key_arn = f"arn:aws:kms:{REGION}:{ACCOUNT}:key/{key_id}"

# 4) OrchestratorRole , trust the agent-runner IAM user
orch_trust = {
    "Version": "2012-10-17",
    "Statement": [{
        "Effect": "Allow",
        "Principal": {"AWS": agent_runner_arn},
        "Action": "sts:AssumeRole",
    }],
}
try:
    iam.create_role(
        RoleName="OrchestratorRole",
        Path="/orchestrator/",
        AssumeRolePolicyDocument=json.dumps(orch_trust),
    )
    log("created OrchestratorRole")
except iam.exceptions.EntityAlreadyExistsException:
    iam.update_assume_role_policy(
        RoleName="OrchestratorRole",
        PolicyDocument=json.dumps(orch_trust),
    )
    log("updated OrchestratorRole trust")

orch_arn = iam.get_role(RoleName="OrchestratorRole")["Role"]["Arn"]
log(f"orchestrator arn = {orch_arn}")

# 5) Replace KMS key policy with explicit principals + specific key arn
key_policy = {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "RootAccountAdmin",
            "Effect": "Allow",
            "Principal": {"AWS": f"arn:aws:iam::{ACCOUNT}:root"},
            "Action": [
                "kms:Encrypt",
                "kms:Decrypt",
                "kms:ReEncryptFrom",
                "kms:ReEncryptTo",
                "kms:GenerateDataKey",
                "kms:GenerateDataKeyWithoutPlaintext",
                "kms:DescribeKey",
                "kms:GetKeyPolicy",
                "kms:PutKeyPolicy",
                "kms:CreateAlias",
                "kms:DeleteAlias",
                "kms:UpdateAlias",
                "kms:CreateGrant",
                "kms:RetireGrant",
                "kms:RevokeGrant",
                "kms:ListGrants",
                "kms:EnableKey",
                "kms:DisableKey",
                "kms:ScheduleKeyDeletion",
                "kms:CancelKeyDeletion",
                "kms:TagResource",
                "kms:UntagResource",
            ],
            "Resource": key_arn,
        },
        {
            "Sid": "OrchestratorDecryptViaSecretsManager",
            "Effect": "Allow",
            "Principal": {"AWS": orch_arn},
            "Action": ["kms:Decrypt", "kms:DescribeKey"],
            "Resource": key_arn,
            "Condition": {
                "StringEquals": {
                    "kms:ViaService": f"secretsmanager.{REGION}.amazonaws.com"
                }
            },
        },
    ],
}
kms.put_key_policy(KeyId=key_id, PolicyName="default", Policy=json.dumps(key_policy))
log("kms key policy installed (specific principals, specific key arn, no wildcards)")


# 6) Secrets, encrypted with the CMK
def upsert_secret(name, value):
    try:
        res = sm.create_secret(Name=name, SecretString=value, KmsKeyId=key_id)
        return res["ARN"]
    except sm.exceptions.ResourceExistsException:
        sm.update_secret(SecretId=name, SecretString=value, KmsKeyId=key_id)
        return sm.describe_secret(SecretId=name)["ARN"]


secret_a_arn = upsert_secret("harbor/cross-account/external-id-tenant-a", ext_a)
secret_b_arn = upsert_secret("harbor/cross-account/external-id-tenant-b", ext_b)
log(f"secret A = {secret_a_arn}")
log(f"secret B = {secret_b_arn}")


# 7) Resource policies on secrets , orchestrator role admitted explicitly
def secret_resource_policy(secret_arn):
    return {
        "Version": "2012-10-17",
        "Statement": [{
            "Sid": "AllowOrchestratorRead",
            "Effect": "Allow",
            "Principal": {"AWS": orch_arn},
            "Action": "secretsmanager:GetSecretValue",
            "Resource": secret_arn,
        }],
    }


sm.put_resource_policy(
    SecretId="harbor/cross-account/external-id-tenant-a",
    ResourcePolicy=json.dumps(secret_resource_policy(secret_a_arn)),
)
sm.put_resource_policy(
    SecretId="harbor/cross-account/external-id-tenant-b",
    ResourcePolicy=json.dumps(secret_resource_policy(secret_b_arn)),
)
log("attached resource policies to both secrets")


# 8) Member role trust documents , both conditions on every trust
def member_trust(ext_id):
    return {
        "Version": "2012-10-17",
        "Statement": [{
            "Effect": "Allow",
            "Principal": {"AWS": orch_arn},
            "Action": "sts:AssumeRole",
            "Condition": {
                "StringEquals": {"sts:ExternalId": ext_id},
                "ArnEquals": {"aws:SourceArn": orch_arn},
            },
        }],
    }


def upsert_member_role(name, ext_id):
    trust = json.dumps(member_trust(ext_id))
    try:
        iam.create_role(
            RoleName=name, Path="/member/", AssumeRolePolicyDocument=trust
        )
        log(f"created {name}")
    except iam.exceptions.EntityAlreadyExistsException:
        iam.update_assume_role_policy(RoleName=name, PolicyDocument=trust)
        log(f"updated trust on {name}")


upsert_member_role("MemberRole-Tenant-A", ext_a)
upsert_member_role("MemberRole-Tenant-B", ext_b)

member_a_arn = iam.get_role(RoleName="MemberRole-Tenant-A")["Role"]["Arn"]
member_b_arn = iam.get_role(RoleName="MemberRole-Tenant-B")["Role"]["Arn"]


# 9) Member identity policies , scoped to that tenant's bucket only
def bucket_only_policy(bucket):
    return {
        "Version": "2012-10-17",
        "Statement": [{
            "Sid": "OwnTenantBucketAccess",
            "Effect": "Allow",
            "Action": ["s3:PutObject", "s3:GetObject"],
            "Resource": [f"arn:aws:s3:::{bucket}/*"],
        }],
    }


iam.put_role_policy(
    RoleName="MemberRole-Tenant-A",
    PolicyName="TenantABucketAccess",
    PolicyDocument=json.dumps(bucket_only_policy("tenant-a-bucket")),
)
iam.put_role_policy(
    RoleName="MemberRole-Tenant-B",
    PolicyName="TenantBBucketAccess",
    PolicyDocument=json.dumps(bucket_only_policy("tenant-b-bucket")),
)
log("attached member identity policies (per-tenant bucket only)")

# 10) OrchestratorRole inline policies , explicit member arns + explicit secret arns
orch_assume_policy = {
    "Version": "2012-10-17",
    "Statement": [{
        "Sid": "AssumeMemberRolesOnly",
        "Effect": "Allow",
        "Action": "sts:AssumeRole",
        "Resource": [member_a_arn, member_b_arn],
    }],
}
iam.put_role_policy(
    RoleName="OrchestratorRole",
    PolicyName="AssumeMemberRoles",
    PolicyDocument=json.dumps(orch_assume_policy),
)

orch_secret_policy = {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "ReadExternalIdSecrets",
            "Effect": "Allow",
            "Action": "secretsmanager:GetSecretValue",
            "Resource": [secret_a_arn, secret_b_arn],
        },
        {
            "Sid": "DecryptExternalIdSecretsViaSecretsManager",
            "Effect": "Allow",
            "Action": ["kms:Decrypt", "kms:DescribeKey"],
            "Resource": key_arn,
            "Condition": {
                "StringEquals": {
                    "kms:ViaService": f"secretsmanager.{REGION}.amazonaws.com"
                }
            },
        },
    ],
}
iam.put_role_policy(
    RoleName="OrchestratorRole",
    PolicyName="ReadExternalIdSecrets",
    PolicyDocument=json.dumps(orch_secret_policy),
)
log("attached orchestrator identity policies")

# 11) Tenant buckets
for b in ["tenant-a-bucket", "tenant-b-bucket"]:
    try:
        s3.create_bucket(Bucket=b)
        log(f"created bucket {b}")
    except s3.exceptions.BucketAlreadyOwnedByYou:
        log(f"bucket {b} already exists")
    except s3.exceptions.BucketAlreadyExists:
        log(f"bucket {b} already exists (other-owned)")

# 12) SSM pointers
ssm_params = [
    ("/harbor/orchestrator/role-arn", orch_arn),
    ("/harbor/external-id-secret-arn-tenant-a", secret_a_arn),
    ("/harbor/external-id-secret-arn-tenant-b", secret_b_arn),
    ("/harbor/member-role-arn-tenant-a", member_a_arn),
    ("/harbor/member-role-arn-tenant-b", member_b_arn),
]
for name, val in ssm_params:
    ssm.put_parameter(Name=name, Type="String", Value=val, Overwrite=True)
log("ssm pointers published under /harbor/")

print(json.dumps({
    "account": ACCOUNT,
    "orchestrator_role_arn": orch_arn,
    "member_a_arn": member_a_arn,
    "member_b_arn": member_b_arn,
    "secret_a_arn": secret_a_arn,
    "secret_b_arn": secret_b_arn,
    "kms_key_arn": key_arn,
    "agent_runner_arn": agent_runner_arn,
}, indent=2))

#!/usr/bin/env python3
"""End-to-end happy path. Discovers everything via SSM.

Chain: agent-runner creds -> assume OrchestratorRole -> read tenant-A
ExternalId from Secrets Manager -> assume MemberRole-Tenant-A with
--external-id <secret> -> PutObject into tenant-a-bucket -> GetObject
roundtrip.
"""
import boto3
import os
import sys

REGION = "us-east-1"
ENDPOINT = os.environ.get("AWS_ENDPOINT_URL", "http://localstack:4566")

ssm = boto3.client("ssm", region_name=REGION, endpoint_url=ENDPOINT)
sts = boto3.client("sts", region_name=REGION, endpoint_url=ENDPOINT)


def get_param(name):
    return ssm.get_parameter(Name=name)["Parameter"]["Value"]


orch_arn = get_param("/harbor/orchestrator/role-arn")
secret_a_arn = get_param("/harbor/external-id-secret-arn-tenant-a")
member_a_arn = get_param("/harbor/member-role-arn-tenant-a")

print(f"[happy] orchestrator    = {orch_arn}")
print(f"[happy] member tenant-A = {member_a_arn}")
print(f"[happy] secret  tenant-A= {secret_a_arn}")

# Step 1: assume OrchestratorRole (as agent-runner would)
orch_creds = sts.assume_role(
    RoleArn=orch_arn, RoleSessionName="orchestrator-session"
)["Credentials"]
print("[happy] step 1: assumed OrchestratorRole")

# Step 2: orchestrator reads the tenant-A external id from Secrets Manager
sm = boto3.client(
    "secretsmanager",
    region_name=REGION,
    endpoint_url=ENDPOINT,
    aws_access_key_id=orch_creds["AccessKeyId"],
    aws_secret_access_key=orch_creds["SecretAccessKey"],
    aws_session_token=orch_creds["SessionToken"],
)
ext_id = sm.get_secret_value(SecretId=secret_a_arn)["SecretString"]
print(f"[happy] step 2: read tenant-A ExternalId ({len(ext_id)} chars)")

# Step 3: orchestrator -> assume MemberRole-Tenant-A with --external-id
sts_orch = boto3.client(
    "sts",
    region_name=REGION,
    endpoint_url=ENDPOINT,
    aws_access_key_id=orch_creds["AccessKeyId"],
    aws_secret_access_key=orch_creds["SecretAccessKey"],
    aws_session_token=orch_creds["SessionToken"],
)
member_creds = sts_orch.assume_role(
    RoleArn=member_a_arn,
    RoleSessionName="member-a-session",
    ExternalId=ext_id,
)["Credentials"]
print("[happy] step 3: assumed MemberRole-Tenant-A with ExternalId")

# Step 4: PutObject into tenant-a-bucket using member-role creds
s3 = boto3.client(
    "s3",
    region_name=REGION,
    endpoint_url=ENDPOINT,
    aws_access_key_id=member_creds["AccessKeyId"],
    aws_secret_access_key=member_creds["SecretAccessKey"],
    aws_session_token=member_creds["SessionToken"],
)
s3.put_object(Bucket="tenant-a-bucket", Key="hello.txt", Body=b"happy path proof")
print("[happy] step 4: PutObject -> s3://tenant-a-bucket/hello.txt")

got = s3.get_object(Bucket="tenant-a-bucket", Key="hello.txt")["Body"].read()
if got != b"happy path proof":
    print(f"[happy] roundtrip mismatch: {got!r}")
    sys.exit(1)
print("[happy] step 5: GetObject roundtrip confirmed")
print("[happy] OK")

#!/usr/bin/env python3
"""Auditor-style verifier , checks policy DOCUMENT shape, not runtime behavior.

Reads SSM pointers, then asserts:
  * each member role's trust contains BOTH StringEquals[sts:ExternalId]
    matching the per-tenant secret value AND ArnEquals/ArnLike
    [aws:SourceArn] equal to the orchestrator role arn
  * trust principal is the orchestrator role arn (not '*')
  * trust action is exactly sts:AssumeRole
  * the two ExternalIds differ between tenants and are >=32 chars
  * orchestrator identity policy enumerates the two member arns and
    the two secret arns explicitly (no wildcards)
  * each member identity policy references only its own tenant bucket
  * each secret has a resource policy admitting the orchestrator role
  * no policy has Action:'*', s3:'*', or kms:'*'; no Resource:'*' for
    s3- or kms-actioned statements
"""
import boto3
import json
import os
import sys

REGION = "us-east-1"
ENDPOINT = os.environ.get("AWS_ENDPOINT_URL", "http://localstack:4566")

iam = boto3.client("iam", region_name=REGION, endpoint_url=ENDPOINT)
ssm = boto3.client("ssm", region_name=REGION, endpoint_url=ENDPOINT)
sm = boto3.client("secretsmanager", region_name=REGION, endpoint_url=ENDPOINT)
kms = boto3.client("kms", region_name=REGION, endpoint_url=ENDPOINT)

failures = []


def check(cond, msg):
    if cond:
        print(f"  PASS {msg}")
    else:
        print(f"  FAIL {msg}")
        failures.append(msg)


def get_param(name):
    return ssm.get_parameter(Name=name)["Parameter"]["Value"]


def stmts(policy):
    s = policy.get("Statement", [])
    return s if isinstance(s, list) else [s]


def as_list(x):
    if x is None:
        return []
    return x if isinstance(x, list) else [x]


def actions_of(s):
    return as_list(s.get("Action"))


def resources_of(s):
    return as_list(s.get("Resource"))


def principal_aws_of(s):
    p = s.get("Principal", {})
    if not isinstance(p, dict):
        return []
    return as_list(p.get("AWS"))


print("== ssm pointers ==")
orch_arn = get_param("/harbor/orchestrator/role-arn")
member_a_arn = get_param("/harbor/member-role-arn-tenant-a")
member_b_arn = get_param("/harbor/member-role-arn-tenant-b")
secret_a_arn = get_param("/harbor/external-id-secret-arn-tenant-a")
secret_b_arn = get_param("/harbor/external-id-secret-arn-tenant-b")
print(f"  orchestrator: {orch_arn}")
print(f"  member-A:     {member_a_arn}")
print(f"  member-B:     {member_b_arn}")
print(f"  secret-A:     {secret_a_arn}")
print(f"  secret-B:     {secret_b_arn}")

print("\n== role enumeration via path ==")
orch_roles = iam.list_roles(PathPrefix="/orchestrator/")["Roles"]
check(
    [r["RoleName"] for r in orch_roles] == ["OrchestratorRole"],
    "/orchestrator/ contains exactly OrchestratorRole",
)
member_roles = iam.list_roles(PathPrefix="/member/")["Roles"]
member_names = sorted(r["RoleName"] for r in member_roles)
check(
    member_names == ["MemberRole-Tenant-A", "MemberRole-Tenant-B"],
    f"/member/ contains exactly the two member roles (got {member_names})",
)


def check_member_trust(role_name, ext_id_value, label):
    print(f"\n== {role_name} trust ==")
    role = iam.get_role(RoleName=role_name)["Role"]
    doc = role["AssumeRolePolicyDocument"]
    if isinstance(doc, str):
        doc = json.loads(doc)
    sts_list = stmts(doc)
    check(len(sts_list) >= 1, f"{label}: trust has at least one Statement")
    s = sts_list[0]

    check(s.get("Effect") == "Allow", f"{label}: Effect == Allow")

    principals = principal_aws_of(s)
    check("*" not in principals, f"{label}: Principal.AWS is NOT '*'")
    check(orch_arn in principals, f"{label}: Principal.AWS includes orchestrator role arn")

    acts = actions_of(s)
    check(acts == ["sts:AssumeRole"], f"{label}: Action is exactly ['sts:AssumeRole'] (got {acts})")

    cond = s.get("Condition", {})
    str_eq_ext = cond.get("StringEquals", {}).get("sts:ExternalId")
    check(
        str_eq_ext == ext_id_value,
        f"{label}: Condition.StringEquals.sts:ExternalId matches the secret VALUE",
    )

    src_arn = None
    cond_op_used = None
    for op in ("ArnEquals", "ArnLike"):
        v = cond.get(op, {}).get("aws:SourceArn")
        if v is not None:
            src_arn = v
            cond_op_used = op
            break
    check(
        src_arn == orch_arn,
        f"{label}: Condition.{{ArnEquals|ArnLike}}.aws:SourceArn == orchestrator arn"
        + (f" (via {cond_op_used})" if cond_op_used else ""),
    )


ext_a = sm.get_secret_value(SecretId=secret_a_arn)["SecretString"]
ext_b = sm.get_secret_value(SecretId=secret_b_arn)["SecretString"]
check_member_trust("MemberRole-Tenant-A", ext_a, "tenant-A")
check_member_trust("MemberRole-Tenant-B", ext_b, "tenant-B")

print("\n== external id properties ==")
check(len(ext_a) >= 32, f"tenant-A ExternalId length >= 32 (got {len(ext_a)})")
check(len(ext_b) >= 32, f"tenant-B ExternalId length >= 32 (got {len(ext_b)})")
check(ext_a != ext_b, "tenant-A and tenant-B ExternalIds differ")


def collect_inline_statements(role_name):
    out = []
    for pn in iam.list_role_policies(RoleName=role_name)["PolicyNames"]:
        doc = iam.get_role_policy(RoleName=role_name, PolicyName=pn)["PolicyDocument"]
        if isinstance(doc, str):
            doc = json.loads(doc)
        for s in stmts(doc):
            out.append((pn, s))
    return out


print("\n== orchestrator identity policies ==")
orch_pairs = collect_inline_statements("OrchestratorRole")
assume_resources = []
for pn, s in orch_pairs:
    if s.get("Effect") != "Allow":
        continue
    if "sts:AssumeRole" in actions_of(s):
        assume_resources.extend(resources_of(s))
check(
    set(assume_resources) == {member_a_arn, member_b_arn},
    f"orchestrator sts:AssumeRole resources are exactly the two member arns (got {assume_resources})",
)
check(
    "*" not in assume_resources,
    "orchestrator sts:AssumeRole resource list contains no '*'",
)

sec_resources = []
for pn, s in orch_pairs:
    if s.get("Effect") != "Allow":
        continue
    if "secretsmanager:GetSecretValue" in actions_of(s):
        sec_resources.extend(resources_of(s))
check(
    set(sec_resources) == {secret_a_arn, secret_b_arn},
    f"orchestrator GetSecretValue resources are exactly the two secret arns (got {sec_resources})",
)
check("*" not in sec_resources, "orchestrator GetSecretValue resource list contains no '*'")


def check_member_identity(role_name, own_bucket, other_bucket, label):
    print(f"\n== {role_name} identity policy ==")
    own_prefix = f"arn:aws:s3:::{own_bucket}"
    other_prefix = f"arn:aws:s3:::{other_bucket}"
    pairs = collect_inline_statements(role_name)
    all_resources, all_actions = [], []
    for pn, s in pairs:
        if s.get("Effect") != "Allow":
            continue
        all_resources.extend(resources_of(s))
        all_actions.extend(actions_of(s))
    check(
        len(all_resources) > 0 and all(r.startswith(own_prefix) for r in all_resources),
        f"{label}: every resource references {own_bucket} (got {all_resources})",
    )
    check(
        not any(r.startswith(other_prefix) for r in all_resources),
        f"{label}: no resource references {other_bucket}",
    )
    check("*" not in all_resources, f"{label}: no Resource '*'")
    check("*" not in all_actions, f"{label}: no Action '*'")
    check("s3:*" not in all_actions, f"{label}: no Action 's3:*'")


check_member_identity("MemberRole-Tenant-A", "tenant-a-bucket", "tenant-b-bucket", "tenant-A")
check_member_identity("MemberRole-Tenant-B", "tenant-b-bucket", "tenant-a-bucket", "tenant-B")


def check_secret_resource_policy(secret_id, label):
    print(f"\n== {label} secret resource policy ==")
    rp = sm.get_resource_policy(SecretId=secret_id)
    rp_str = rp.get("ResourcePolicy")
    check(bool(rp_str), f"{label}: resource policy attached")
    if not rp_str:
        return
    pol = json.loads(rp_str)
    found = False
    for s in stmts(pol):
        if s.get("Effect") != "Allow":
            continue
        if "secretsmanager:GetSecretValue" not in actions_of(s):
            continue
        if orch_arn in principal_aws_of(s):
            found = True
            break
    check(found, f"{label}: resource policy admits orchestrator role for GetSecretValue")


check_secret_resource_policy("harbor/cross-account/external-id-tenant-a", "tenant-A")
check_secret_resource_policy("harbor/cross-account/external-id-tenant-b", "tenant-B")

print("\n== wildcard scan (s3 / kms) over all role inline policies ==")
WILDCARD_ACTIONS = {"*", "s3:*", "kms:*"}
for role_name in ["OrchestratorRole", "MemberRole-Tenant-A", "MemberRole-Tenant-B"]:
    for pn, s in collect_inline_statements(role_name):
        if s.get("Effect") != "Allow":
            continue
        acts = actions_of(s)
        rsrcs = resources_of(s)
        for a in acts:
            if a in WILDCARD_ACTIONS:
                msg = f"{role_name}/{pn}: forbidden wildcard action {a!r}"
                print(f"  FAIL {msg}")
                failures.append(msg)
        if "*" in rsrcs:
            for a in acts:
                svc = a.split(":")[0] if ":" in a else None
                if svc in ("s3", "kms"):
                    msg = f"{role_name}/{pn}: Resource '*' on {svc} action {a!r}"
                    print(f"  FAIL {msg}")
                    failures.append(msg)
print("  (scan done)")

print("\n== kms key policy ==")
key_id = kms.describe_key(KeyId="alias/cross-account-cmk")["KeyMetadata"]["KeyId"]
kp_str = kms.get_key_policy(KeyId=key_id, PolicyName="default")["Policy"]
kp = json.loads(kp_str)
for s in stmts(kp):
    if s.get("Effect") != "Allow":
        continue
    principals = principal_aws_of(s)
    check(
        "*" not in principals,
        f"kms key policy stmt {s.get('Sid', '?')!r}: Principal.AWS is not '*'",
    )
    acts = actions_of(s)
    check(
        "*" not in acts and "kms:*" not in acts,
        f"kms key policy stmt {s.get('Sid', '?')!r}: no wildcard kms action",
    )
    rsrcs = resources_of(s)
    check(
        "*" not in rsrcs,
        f"kms key policy stmt {s.get('Sid', '?')!r}: Resource is not '*'",
    )

print()
if failures:
    print(f"== VERIFY FAILED ({len(failures)} issues) ==")
    for f in failures:
        print(f"  - {f}")
    sys.exit(1)
print("== VERIFY OK ==")