iam-cross-account-externalid-sourcearn

#!/usr/bin/env bash
set -euo pipefail

# ---------- constants ----------
ACCOUNT_ID="000000000000"
REGION="us-east-1"

ORCH_USER="agent-runner"
ATTACKER_USER="attacker-runner"

ORCH_ROLE="OrchestratorRole"
ORCH_PATH="/orchestrator/"
ORCH_ROLE_ARN="arn:aws:iam::${ACCOUNT_ID}:role${ORCH_PATH}${ORCH_ROLE}"

MEMBER_PATH="/member/"
MEMBER_A="MemberRole-Tenant-A"
MEMBER_B="MemberRole-Tenant-B"
MEMBER_A_ARN="arn:aws:iam::${ACCOUNT_ID}:role${MEMBER_PATH}${MEMBER_A}"
MEMBER_B_ARN="arn:aws:iam::${ACCOUNT_ID}:role${MEMBER_PATH}${MEMBER_B}"

KMS_ALIAS="alias/cross-account-cmk"

SECRET_A_NAME="harbor/cross-account/external-id-tenant-a"
SECRET_B_NAME="harbor/cross-account/external-id-tenant-b"

BUCKET_A="tenant-a-bucket"
BUCKET_B="tenant-b-bucket"

WORK=/app/.state
mkdir -p "$WORK"

# ---------- helpers ----------
aws_() { aws --region "$REGION" "$@"; }

# ---------- 1. clean prior state (idempotent) ----------
echo ">> cleaning prior state"

# detach + delete role inline policies + roles
for role_path_pair in "${ORCH_PATH}${ORCH_ROLE}" "${MEMBER_PATH}${MEMBER_A}" "${MEMBER_PATH}${MEMBER_B}"; do
  role_name="${role_path_pair##*/}"
  if aws_ iam get-role --role-name "$role_name" >/dev/null 2>&1; then
    for p in $(aws_ iam list-role-policies --role-name "$role_name" --query 'PolicyNames[]' --output text); do
      aws_ iam delete-role-policy --role-name "$role_name" --policy-name "$p" || true
    done
    aws_ iam delete-role --role-name "$role_name" || true
  fi
done

# delete users (must remove access keys + inline policies first)
for u in "$ORCH_USER" "$ATTACKER_USER"; do
  if aws_ iam get-user --user-name "$u" >/dev/null 2>&1; then
    for k in $(aws_ iam list-access-keys --user-name "$u" --query 'AccessKeyMetadata[].AccessKeyId' --output text); do
      aws_ iam delete-access-key --user-name "$u" --access-key-id "$k" || true
    done
    for p in $(aws_ iam list-user-policies --user-name "$u" --query 'PolicyNames[]' --output text); do
      aws_ iam delete-user-policy --user-name "$u" --policy-name "$p" || true
    done
    aws_ iam delete-user --user-name "$u" || true
  fi
done

# delete secrets
for s in "$SECRET_A_NAME" "$SECRET_B_NAME"; do
  aws_ secretsmanager delete-secret --secret-id "$s" --force-delete-without-recovery >/dev/null 2>&1 || true
done

# buckets (empty + delete)
for b in "$BUCKET_A" "$BUCKET_B"; do
  aws_ s3 rm "s3://$b" --recursive >/dev/null 2>&1 || true
  aws_ s3api delete-bucket --bucket "$b" >/dev/null 2>&1 || true
done

# kms alias (delete; underlying key is left , localstack tolerates orphans)
aws_ kms delete-alias --alias-name "$KMS_ALIAS" >/dev/null 2>&1 || true

# ssm
for p in /harbor/orchestrator/role-arn \
         /harbor/external-id-secret-arn-tenant-a \
         /harbor/external-id-secret-arn-tenant-b \
         /harbor/member-role-arn-tenant-a \
         /harbor/member-role-arn-tenant-b; do
  aws_ ssm delete-parameter --name "$p" >/dev/null 2>&1 || true
done

# ---------- 2. KMS customer-managed key + alias ----------
echo ">> creating KMS CMK"
KEY_POLICY=$(cat <<EOF
{
  "Version": "2012-10-17",
  "Id": "cross-account-cmk",
  "Statement": [
    {
      "Sid": "RootAdmin",
      "Effect": "Allow",
      "Principal": { "AWS": "arn:aws:iam::${ACCOUNT_ID}:root" },
      "Action": [
        "kms:Create*", "kms:Describe*", "kms:Enable*", "kms:List*",
        "kms:Put*", "kms:Update*", "kms:Revoke*", "kms:Disable*",
        "kms:Get*", "kms:Delete*", "kms:TagResource", "kms:UntagResource",
        "kms:ScheduleKeyDeletion", "kms:CancelKeyDeletion"
      ],
      "Resource": "*"
    },
    {
      "Sid": "AllowOrchestratorDecrypt",
      "Effect": "Allow",
      "Principal": { "AWS": "${ORCH_ROLE_ARN}" },
      "Action": [
        "kms:Decrypt",
        "kms:DescribeKey"
      ],
      "Resource": "*"
    },
    {
      "Sid": "AllowSecretsManagerUse",
      "Effect": "Allow",
      "Principal": { "Service": "secretsmanager.amazonaws.com" },
      "Action": [
        "kms:Encrypt",
        "kms:Decrypt",
        "kms:ReEncrypt",
        "kms:GenerateDataKey",
        "kms:DescribeKey"
      ],
      "Resource": "*"
    }
  ]
}
EOF
)
KEY_ID=$(aws_ kms create-key --description "cross-account-cmk" --policy "$KEY_POLICY" --query 'KeyMetadata.KeyId' --output text)
KEY_ARN="arn:aws:kms:${REGION}:${ACCOUNT_ID}:key/${KEY_ID}"
aws_ kms create-alias --alias-name "$KMS_ALIAS" --target-key-id "$KEY_ID"
echo "   key arn: $KEY_ARN"

# ---------- 3. ExternalIds ----------
echo ">> generating per-tenant ExternalIds"
EXT_A=$(python3 -c 'import secrets; print(secrets.token_urlsafe(36))')
EXT_B=$(python3 -c 'import secrets; print(secrets.token_urlsafe(36))')
# guarantee distinct + length
[ "$EXT_A" != "$EXT_B" ] || { echo "ExternalIds collided"; exit 1; }
[ ${#EXT_A} -ge 32 ] && [ ${#EXT_B} -ge 32 ] || { echo "ExternalId too short"; exit 1; }

# ---------- 4. Secrets Manager secrets (CMK-encrypted) ----------
echo ">> creating secrets"
SECRET_A_ARN=$(aws_ secretsmanager create-secret \
  --name "$SECRET_A_NAME" \
  --kms-key-id "$KEY_ARN" \
  --secret-string "$EXT_A" \
  --query 'ARN' --output text)
SECRET_B_ARN=$(aws_ secretsmanager create-secret \
  --name "$SECRET_B_NAME" \
  --kms-key-id "$KEY_ARN" \
  --secret-string "$EXT_B" \
  --query 'ARN' --output text)
echo "   secret A arn: $SECRET_A_ARN"
echo "   secret B arn: $SECRET_B_ARN"

# ---------- 5. IAM users ----------
echo ">> creating IAM users"
aws_ iam create-user --user-name "$ORCH_USER" >/dev/null
aws_ iam create-user --user-name "$ATTACKER_USER" >/dev/null

# access keys for the orchestrator caller
ORCH_KEYS_JSON=$(aws_ iam create-access-key --user-name "$ORCH_USER")
ORCH_AK=$(echo "$ORCH_KEYS_JSON" | jq -r '.AccessKey.AccessKeyId')
ORCH_SK=$(echo "$ORCH_KEYS_JSON" | jq -r '.AccessKey.SecretAccessKey')

# allow agent-runner user to assume OrchestratorRole
cat > "$WORK/agent-runner-policy.json" <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "sts:AssumeRole",
      "Resource": "${ORCH_ROLE_ARN}"
    }
  ]
}
EOF
aws_ iam put-user-policy --user-name "$ORCH_USER" \
  --policy-name AssumeOrchestrator \
  --policy-document "file://$WORK/agent-runner-policy.json"

# ---------- 6. Orchestrator role ----------
echo ">> creating OrchestratorRole"
cat > "$WORK/orch-trust.json" <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": { "AWS": "arn:aws:iam::${ACCOUNT_ID}:user/${ORCH_USER}" },
      "Action": "sts:AssumeRole"
    }
  ]
}
EOF
aws_ iam create-role \
  --role-name "$ORCH_ROLE" \
  --path "$ORCH_PATH" \
  --assume-role-policy-document "file://$WORK/orch-trust.json" >/dev/null

# orchestrator identity policy: explicit member arns + scoped secret arns (no wildcards)
cat > "$WORK/orch-identity.json" <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AssumeMemberRoles",
      "Effect": "Allow",
      "Action": "sts:AssumeRole",
      "Resource": [
        "${MEMBER_A_ARN}",
        "${MEMBER_B_ARN}"
      ]
    },
    {
      "Sid": "ReadExternalIdSecrets",
      "Effect": "Allow",
      "Action": "secretsmanager:GetSecretValue",
      "Resource": [
        "${SECRET_A_ARN}",
        "${SECRET_B_ARN}"
      ]
    },
    {
      "Sid": "DecryptExternalIdSecrets",
      "Effect": "Allow",
      "Action": "kms:Decrypt",
      "Resource": "${KEY_ARN}"
    }
  ]
}
EOF
aws_ iam put-role-policy \
  --role-name "$ORCH_ROLE" \
  --policy-name OrchestratorPolicy \
  --policy-document "file://$WORK/orch-identity.json"

# ---------- 7. Member role builder ----------
build_member_role() {
  local role_name="$1" external_id="$2" bucket="$3"

  cat > "$WORK/${role_name}-trust.json" <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": { "AWS": "${ORCH_ROLE_ARN}" },
      "Action": "sts:AssumeRole",
      "Condition": {
        "StringEquals": {
          "sts:ExternalId": "${external_id}"
        },
        "ArnEquals": {
          "aws:SourceArn": "${ORCH_ROLE_ARN}"
        }
      }
    }
  ]
}
EOF

  aws_ iam create-role \
    --role-name "$role_name" \
    --path "$MEMBER_PATH" \
    --assume-role-policy-document "file://$WORK/${role_name}-trust.json" >/dev/null

  cat > "$WORK/${role_name}-identity.json" <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "s3:PutObject",
        "s3:GetObject",
        "s3:ListBucket"
      ],
      "Resource": [
        "arn:aws:s3:::${bucket}",
        "arn:aws:s3:::${bucket}/*"
      ]
    }
  ]
}
EOF
  aws_ iam put-role-policy \
    --role-name "$role_name" \
    --policy-name TenantBucketAccess \
    --policy-document "file://$WORK/${role_name}-identity.json"
}

echo ">> creating MemberRole-Tenant-A"
build_member_role "$MEMBER_A" "$EXT_A" "$BUCKET_A"
echo ">> creating MemberRole-Tenant-B"
build_member_role "$MEMBER_B" "$EXT_B" "$BUCKET_B"

# ---------- 8. Secrets Manager resource policies ----------
echo ">> attaching resource policies on secrets"
RES_POLICY=$(cat <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "OrchestratorOnly",
      "Effect": "Allow",
      "Principal": { "AWS": "${ORCH_ROLE_ARN}" },
      "Action": "secretsmanager:GetSecretValue",
      "Resource": "*"
    }
  ]
}
EOF
)
aws_ secretsmanager put-resource-policy --secret-id "$SECRET_A_NAME" --resource-policy "$RES_POLICY" >/dev/null
aws_ secretsmanager put-resource-policy --secret-id "$SECRET_B_NAME" --resource-policy "$RES_POLICY" >/dev/null

# ---------- 9. S3 buckets ----------
echo ">> creating tenant buckets"
aws_ s3api create-bucket --bucket "$BUCKET_A" >/dev/null
aws_ s3api create-bucket --bucket "$BUCKET_B" >/dev/null

# ---------- 10. SSM pointers ----------
echo ">> publishing SSM pointers"
aws_ ssm put-parameter --name /harbor/orchestrator/role-arn          --type String --overwrite --value "$ORCH_ROLE_ARN"   >/dev/null
aws_ ssm put-parameter --name /harbor/external-id-secret-arn-tenant-a --type String --overwrite --value "$SECRET_A_ARN"  >/dev/null
aws_ ssm put-parameter --name /harbor/external-id-secret-arn-tenant-b --type String --overwrite --value "$SECRET_B_ARN"  >/dev/null
aws_ ssm put-parameter --name /harbor/member-role-arn-tenant-a       --type String --overwrite --value "$MEMBER_A_ARN"  >/dev/null
aws_ ssm put-parameter --name /harbor/member-role-arn-tenant-b       --type String --overwrite --value "$MEMBER_B_ARN"  >/dev/null

# ---------- 11. dump local artefacts the runner needs ----------
cat > "$WORK/orch-creds.env" <<EOF
ORCH_AK=${ORCH_AK}
ORCH_SK=${ORCH_SK}
ORCH_ROLE_ARN=${ORCH_ROLE_ARN}
MEMBER_A_ARN=${MEMBER_A_ARN}
MEMBER_B_ARN=${MEMBER_B_ARN}
SECRET_A_NAME=${SECRET_A_NAME}
SECRET_B_NAME=${SECRET_B_NAME}
BUCKET_A=${BUCKET_A}
BUCKET_B=${BUCKET_B}
EOF

echo ">> setup complete"

#!/usr/bin/env python3
"""End-to-end happy path: agent-runner -> OrchestratorRole -> MemberRole-Tenant-A -> S3 PutObject."""
import io
import os
import sys

import boto3

ENDPOINT = os.environ["AWS_ENDPOINT_URL"]
REGION = os.environ.get("AWS_DEFAULT_REGION", "us-east-1")


def client(service: str, **creds):
    return boto3.client(service, endpoint_url=ENDPOINT, region_name=REGION, **creds)


def load_state() -> dict:
    state = {}
    with open("/app/.state/orch-creds.env") as f:
        for line in f:
            k, _, v = line.strip().partition("=")
            state[k] = v
    return state


def main() -> int:
    s = load_state()

    # SSM pointers , verifier-style discovery
    ssm = client("ssm")
    ssm_orch_arn = ssm.get_parameter(Name="/harbor/orchestrator/role-arn")["Parameter"]["Value"]
    ssm_secret_a = ssm.get_parameter(Name="/harbor/external-id-secret-arn-tenant-a")["Parameter"]["Value"]
    ssm_member_a = ssm.get_parameter(Name="/harbor/member-role-arn-tenant-a")["Parameter"]["Value"]
    assert ssm_orch_arn == s["ORCH_ROLE_ARN"], "SSM mismatch"
    assert ssm_member_a == s["MEMBER_A_ARN"], "SSM mismatch"

    # Step 1 , agent-runner user credentials assume OrchestratorRole
    sts_user = client("sts", aws_access_key_id=s["ORCH_AK"], aws_secret_access_key=s["ORCH_SK"])
    orch = sts_user.assume_role(
        RoleArn=s["ORCH_ROLE_ARN"],
        RoleSessionName="orch-session",
    )["Credentials"]
    print(f"  [1] agent-runner assumed OrchestratorRole -> {orch['AccessKeyId']}")

    orch_creds = dict(
        aws_access_key_id=orch["AccessKeyId"],
        aws_secret_access_key=orch["SecretAccessKey"],
        aws_session_token=orch["SessionToken"],
    )

    # Step 2 , fetch tenant-A ExternalId from Secrets Manager
    sm = client("secretsmanager", **orch_creds)
    ext_a = sm.get_secret_value(SecretId=ssm_secret_a)["SecretString"]
    print(f"  [2] orchestrator fetched ExternalId for tenant A (len={len(ext_a)})")

    # Step 3 , assume MemberRole-Tenant-A with --external-id
    sts_orch = client("sts", **orch_creds)
    member = sts_orch.assume_role(
        RoleArn=ssm_member_a,
        RoleSessionName="tenant-a-session",
        ExternalId=ext_a,
    )["Credentials"]
    print(f"  [3] orchestrator assumed MemberRole-Tenant-A -> {member['AccessKeyId']}")

    # Step 4 , put an object into tenant-a-bucket
    s3 = client(
        "s3",
        aws_access_key_id=member["AccessKeyId"],
        aws_secret_access_key=member["SecretAccessKey"],
        aws_session_token=member["SessionToken"],
    )
    payload = b"hello from tenant A"
    s3.put_object(Bucket=s["BUCKET_A"], Key="hello.txt", Body=io.BytesIO(payload))
    body = s3.get_object(Bucket=s["BUCKET_A"], Key="hello.txt")["Body"].read()
    assert body == payload, "round-trip mismatch"
    print(f"  [4] put+get on s3://{s['BUCKET_A']}/hello.txt OK")

    print("HAPPY PATH OK")
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""
Audit verifier , inspects policy documents the way an external auditor would.

Asserts (per the spec):
  trust shape:
    - both tenants' trust docs contain Condition.StringEquals[sts:ExternalId] matching the per-tenant secret value
    - both contain Condition.{ArnLike|ArnEquals}[aws:SourceArn] equal to the orchestrator role arn
    - principal is the orchestrator role arn, NOT "*"
    - action is exactly "sts:AssumeRole"
  ExternalId hygiene:
    - each ExternalId is >= 32 chars
    - the two ExternalIds differ
  least privilege:
    - orchestrator identity policy enumerates the two member role arns (no wildcard assume)
    - orchestrator secret-get scoped to the two known secret arns (no wildcard)
    - each member role identity policy references only its own bucket arn (no "*", no other tenant's bucket)
    - no identity policy has Action:"*" or Resource:"*" for s3 or kms
  secret resource policy:
    - both secrets carry a resource policy naming OrchestratorRole arn as Principal.AWS for secretsmanager:GetSecretValue
"""
from __future__ import annotations

import json
import os
import sys
from typing import Any

import boto3

ENDPOINT = os.environ["AWS_ENDPOINT_URL"]
REGION = os.environ.get("AWS_DEFAULT_REGION", "us-east-1")

ORCH_PATH = "/orchestrator/"
MEMBER_PATH = "/member/"
ORCH_ROLE_NAME = "OrchestratorRole"
MEMBER_A_NAME = "MemberRole-Tenant-A"
MEMBER_B_NAME = "MemberRole-Tenant-B"
BUCKET_A = "tenant-a-bucket"
BUCKET_B = "tenant-b-bucket"


def client(svc: str):
    return boto3.client(svc, endpoint_url=ENDPOINT, region_name=REGION)


# ---------- assertion harness ----------
class V:
    def __init__(self) -> None:
        self.checks: list[tuple[bool, str]] = []

    def ok(self, cond: bool, msg: str) -> None:
        self.checks.append((bool(cond), msg))

    def report(self) -> int:
        passed = sum(1 for c, _ in self.checks if c)
        total = len(self.checks)
        for c, m in self.checks:
            print(f"  [{'PASS' if c else 'FAIL'}] {m}")
        print(f"\n{passed}/{total} checks passed")
        return 0 if passed == total else 1


def as_list(x: Any) -> list:
    return x if isinstance(x, list) else [x]


# ---------- helpers ----------
def get_role(iam, name: str) -> dict:
    return iam.get_role(RoleName=name)["Role"]


def get_inline_policy(iam, role: str, policy: str) -> dict:
    return iam.get_role_policy(RoleName=role, PolicyName=policy)["PolicyDocument"]


def list_role_policies(iam, role: str) -> list[str]:
    return iam.list_role_policies(RoleName=role)["PolicyNames"]


def find_assume_role_statement(doc: dict) -> dict | None:
    for st in as_list(doc.get("Statement", [])):
        actions = as_list(st.get("Action", []))
        if "sts:AssumeRole" in actions and st.get("Effect") == "Allow":
            return st
    return None


def has_wildcard_for_service(doc: dict, services: tuple[str, ...]) -> str | None:
    for st in as_list(doc.get("Statement", [])):
        if st.get("Effect") != "Allow":
            continue
        actions = as_list(st.get("Action", []))
        resources = as_list(st.get("Resource", []))
        # detect if this statement governs any of the given services
        relevant = any(
            a == "*"
            or any(a.lower().startswith(p + ":") for p in services)
            for a in actions
        )
        if not relevant:
            continue
        if "*" in actions:
            return f"Action: '*' present in statement {st.get('Sid', '?')}"
        for a in actions:
            for p in services:
                if a.lower() == f"{p}:*":
                    return f"Action '{a}' is full-service wildcard"
        if "*" in resources:
            return f"Resource: '*' present in statement {st.get('Sid', '?')}"
    return None


# ---------- main ----------
def main() -> int:
    v = V()
    iam = client("iam")
    sm = client("secretsmanager")
    ssm = client("ssm")

    # discover via SSM (verifier doesn't guess names)
    orch_arn = ssm.get_parameter(Name="/harbor/orchestrator/role-arn")["Parameter"]["Value"]
    secret_a_arn = ssm.get_parameter(Name="/harbor/external-id-secret-arn-tenant-a")["Parameter"]["Value"]
    secret_b_arn = ssm.get_parameter(Name="/harbor/external-id-secret-arn-tenant-b")["Parameter"]["Value"]
    member_a_arn = ssm.get_parameter(Name="/harbor/member-role-arn-tenant-a")["Parameter"]["Value"]
    member_b_arn = ssm.get_parameter(Name="/harbor/member-role-arn-tenant-b")["Parameter"]["Value"]
    print(f"orchestrator: {orch_arn}")
    print(f"member A:     {member_a_arn}")
    print(f"member B:     {member_b_arn}\n")

    # enumerate roles by path (the verifier uses paths)
    orch_under_path = iam.list_roles(PathPrefix=ORCH_PATH)["Roles"]
    members_under_path = iam.list_roles(PathPrefix=MEMBER_PATH)["Roles"]
    v.ok(len(orch_under_path) == 1 and orch_under_path[0]["RoleName"] == ORCH_ROLE_NAME,
         f"exactly one role under {ORCH_PATH} (OrchestratorRole)")
    v.ok({r["RoleName"] for r in members_under_path} == {MEMBER_A_NAME, MEMBER_B_NAME},
         f"exactly two roles under {MEMBER_PATH} (Tenant-A, Tenant-B)")

    # pull ExternalId values from secrets
    ext_a = sm.get_secret_value(SecretId=secret_a_arn)["SecretString"]
    ext_b = sm.get_secret_value(SecretId=secret_b_arn)["SecretString"]
    v.ok(len(ext_a) >= 32, f"tenant-A ExternalId length >= 32 (got {len(ext_a)})")
    v.ok(len(ext_b) >= 32, f"tenant-B ExternalId length >= 32 (got {len(ext_b)})")
    v.ok(ext_a != ext_b, "tenant-A and tenant-B ExternalIds differ")

    # ---------- per-tenant trust docs ----------
    for tenant_label, role_name, expected_ext, expected_arn in (
        ("A", MEMBER_A_NAME, ext_a, member_a_arn),
        ("B", MEMBER_B_NAME, ext_b, member_b_arn),
    ):
        role = get_role(iam, role_name)
        v.ok(role["Arn"] == expected_arn, f"{role_name} arn matches SSM pointer")

        trust = role["AssumeRolePolicyDocument"]
        st = find_assume_role_statement(trust)
        v.ok(st is not None, f"{role_name} trust has Allow sts:AssumeRole statement")
        if not st:
            continue

        actions = as_list(st["Action"])
        v.ok(actions == ["sts:AssumeRole"], f"{role_name} trust action is exactly sts:AssumeRole (got {actions})")

        principal = st.get("Principal", {})
        principal_aws = as_list(principal.get("AWS", []))
        v.ok(principal_aws == [orch_arn],
             f"{role_name} trust Principal.AWS is the orchestrator role arn (no '*')")
        v.ok("*" not in principal_aws and principal != "*" and principal.get("AWS") != "*",
             f"{role_name} trust Principal is not '*'")

        cond = st.get("Condition", {})
        # ExternalId
        ext_block = cond.get("StringEquals", {})
        ext_val = ext_block.get("sts:ExternalId")
        v.ok(ext_val == expected_ext,
             f"{role_name} StringEquals['sts:ExternalId'] matches secret value for tenant {tenant_label}")

        # SourceArn , accept ArnEquals or ArnLike
        src_arn_val = (
            cond.get("ArnEquals", {}).get("aws:SourceArn")
            or cond.get("ArnLike", {}).get("aws:SourceArn")
        )
        v.ok(src_arn_val == orch_arn,
             f"{role_name} ArnEquals/ArnLike['aws:SourceArn'] equals orchestrator role arn")

    # ---------- orchestrator identity policy ----------
    orch_policies = list_role_policies(iam, ORCH_ROLE_NAME)
    v.ok(len(orch_policies) >= 1, "orchestrator has at least one inline policy")
    # merge all inline policies for analysis
    orch_doc_statements: list[dict] = []
    for p in orch_policies:
        orch_doc_statements.extend(as_list(get_inline_policy(iam, ORCH_ROLE_NAME, p).get("Statement", [])))

    # find the assume statement
    assume_st = next(
        (s for s in orch_doc_statements if "sts:AssumeRole" in as_list(s.get("Action", []))),
        None,
    )
    v.ok(assume_st is not None, "orchestrator identity policy contains an sts:AssumeRole statement")
    if assume_st:
        res = as_list(assume_st.get("Resource", []))
        v.ok(set(res) == {member_a_arn, member_b_arn},
             "orchestrator AssumeRole Resource enumerates the two member role arns exactly")
        v.ok("*" not in res, "orchestrator AssumeRole has no Resource:'*'")

    # secrets-get statement
    secret_st = next(
        (s for s in orch_doc_statements if "secretsmanager:GetSecretValue" in as_list(s.get("Action", []))),
        None,
    )
    v.ok(secret_st is not None, "orchestrator identity policy contains GetSecretValue statement")
    if secret_st:
        res = as_list(secret_st.get("Resource", []))
        v.ok(set(res) == {secret_a_arn, secret_b_arn},
             "orchestrator GetSecretValue Resource enumerates the two ExternalId secret arns exactly")
        v.ok("*" not in res, "orchestrator GetSecretValue has no Resource:'*'")

    # no s3/kms wildcards
    issue = has_wildcard_for_service({"Statement": orch_doc_statements}, ("s3", "kms"))
    v.ok(issue is None, f"orchestrator identity policy has no s3/kms wildcards ({issue or 'clean'})")

    # ---------- member role identity policies ----------
    for role_name, own_bucket, other_bucket in (
        (MEMBER_A_NAME, BUCKET_A, BUCKET_B),
        (MEMBER_B_NAME, BUCKET_B, BUCKET_A),
    ):
        names = list_role_policies(iam, role_name)
        v.ok(len(names) >= 1, f"{role_name} has at least one inline policy")
        statements: list[dict] = []
        for n in names:
            statements.extend(as_list(get_inline_policy(iam, role_name, n).get("Statement", [])))

        own_arns = {f"arn:aws:s3:::{own_bucket}", f"arn:aws:s3:::{own_bucket}/*"}
        all_resources: set[str] = set()
        for st in statements:
            for r in as_list(st.get("Resource", [])):
                all_resources.add(r)

        v.ok("*" not in all_resources, f"{role_name} has no Resource:'*'")
        v.ok(all_resources == own_arns,
             f"{role_name} resources are exactly {sorted(own_arns)} (got {sorted(all_resources)})")
        v.ok(not any(other_bucket in r for r in all_resources),
             f"{role_name} contains no reference to other tenant's bucket '{other_bucket}'")

        issue = has_wildcard_for_service({"Statement": statements}, ("s3", "kms"))
        v.ok(issue is None, f"{role_name} has no s3/kms wildcards ({issue or 'clean'})")

    # ---------- secret resource policies ----------
    for label, secret_arn in (("A", secret_a_arn), ("B", secret_b_arn)):
        rp = sm.get_resource_policy(SecretId=secret_arn).get("ResourcePolicy")
        v.ok(rp is not None, f"tenant-{label} secret has a resource policy")
        if rp:
            doc = json.loads(rp)
            allowed = False
            for st in as_list(doc.get("Statement", [])):
                if st.get("Effect") != "Allow":
                    continue
                actions = as_list(st.get("Action", []))
                if "secretsmanager:GetSecretValue" not in actions and "*" not in actions:
                    continue
                principal = st.get("Principal", {})
                if isinstance(principal, dict):
                    aws_p = as_list(principal.get("AWS", []))
                else:
                    aws_p = [principal]
                if orch_arn in aws_p and "*" not in aws_p:
                    allowed = True
                    break
            v.ok(allowed,
                 f"tenant-{label} secret resource policy admits OrchestratorRole arn for GetSecretValue (and not '*')")

    return v.report()


if __name__ == "__main__":
    sys.exit(main())