iam-cross-account-externalid-sourcearn

"""Build the cross-account trust topology end-to-end on LocalStack.

Resources created (in dependency order):
  - IAM users: agent-runner (orchestrator caller), attacker-runner (impersonator stand-in)
  - KMS CMK + alias/cross-account-cmk
  - Two ExternalId secrets (KMS-encrypted) with resource policies
  - OrchestratorRole (trusted by agent-runner) under path /orchestrator/
  - MemberRole-Tenant-A and MemberRole-Tenant-B under path /member/
      trust = orchestrator role arn AS PRINCIPAL
            + sts:ExternalId == per-tenant secret value
            + aws:SourceArn  == orchestrator role arn
  - Two tenant S3 buckets
  - SSM /harbor/... pointers so the verifier can find things by name
"""

import json
import secrets
import string

import boto3

ENDPOINT = "http://localstack:4566"
REGION = "us-east-1"
ACCOUNT_ID = "000000000000"

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


def gen_external_id() -> str:
    alphabet = string.ascii_letters + string.digits + "_-"
    return "".join(secrets.choice(alphabet) for _ in range(48))


def main() -> None:
    ext_id_a = gen_external_id()
    ext_id_b = gen_external_id()
    assert ext_id_a != ext_id_b, "ExternalIds must differ between tenants"
    assert len(ext_id_a) >= 32 and len(ext_id_b) >= 32

    # ── pre-compute ARNs we'll cross-reference ───────────────────────────────
    orch_arn = f"arn:aws:iam::{ACCOUNT_ID}:role/orchestrator/OrchestratorRole"
    mem_a_arn = f"arn:aws:iam::{ACCOUNT_ID}:role/member/MemberRole-Tenant-A"
    mem_b_arn = f"arn:aws:iam::{ACCOUNT_ID}:role/member/MemberRole-Tenant-B"
    agent_arn = f"arn:aws:iam::{ACCOUNT_ID}:user/agent-runner"

    # ── IAM users ────────────────────────────────────────────────────────────
    iam.create_user(UserName="agent-runner")
    iam.create_user(UserName="attacker-runner")

    # ── KMS key for ExternalId secret encryption ─────────────────────────────
    key = kms.create_key(Description="cross-account ExternalId CMK")
    key_id = key["KeyMetadata"]["KeyId"]
    key_arn = key["KeyMetadata"]["Arn"]
    kms.create_alias(AliasName="alias/cross-account-cmk", TargetKeyId=key_id)

    # ── Secrets (KMS-encrypted) holding the per-tenant ExternalId values ────
    sec_a = sm.create_secret(
        Name="harbor/cross-account/external-id-tenant-a",
        KmsKeyId=key_arn,
        SecretString=ext_id_a,
    )
    sec_b = sm.create_secret(
        Name="harbor/cross-account/external-id-tenant-b",
        KmsKeyId=key_arn,
        SecretString=ext_id_b,
    )
    sec_a_arn = sec_a["ARN"]
    sec_b_arn = sec_b["ARN"]

    # ── OrchestratorRole , trusted by the agent-runner IAM user ─────────────
    orch_trust = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Effect": "Allow",
                "Principal": {"AWS": agent_arn},
                "Action": "sts:AssumeRole",
            }
        ],
    }
    iam.create_role(
        RoleName="OrchestratorRole",
        Path="/orchestrator/",
        AssumeRolePolicyDocument=json.dumps(orch_trust),
    )

    # ── Member roles , trust requires ExternalId AND SourceArn ──────────────
    def member_trust(ext_id: str) -> dict:
        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},
                    },
                }
            ],
        }

    iam.create_role(
        RoleName="MemberRole-Tenant-A",
        Path="/member/",
        AssumeRolePolicyDocument=json.dumps(member_trust(ext_id_a)),
    )
    iam.create_role(
        RoleName="MemberRole-Tenant-B",
        Path="/member/",
        AssumeRolePolicyDocument=json.dumps(member_trust(ext_id_b)),
    )

    # ── Resource policies on each secret , orchestrator must be admitted ────
    def secret_resource_policy(secret_arn: str) -> dict:
        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=sec_a_arn,
        ResourcePolicy=json.dumps(secret_resource_policy(sec_a_arn)),
    )
    sm.put_resource_policy(
        SecretId=sec_b_arn,
        ResourcePolicy=json.dumps(secret_resource_policy(sec_b_arn)),
    )

    # ── Tenant buckets ──────────────────────────────────────────────────────
    s3.create_bucket(Bucket="tenant-a-bucket")
    s3.create_bucket(Bucket="tenant-b-bucket")

    # ── Identity policies (least privilege) ─────────────────────────────────
    orch_identity = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "AssumeMemberRoles",
                "Effect": "Allow",
                "Action": "sts:AssumeRole",
                "Resource": [mem_a_arn, mem_b_arn],
            },
            {
                "Sid": "ReadExternalIdSecrets",
                "Effect": "Allow",
                "Action": "secretsmanager:GetSecretValue",
                "Resource": [sec_a_arn, sec_b_arn],
            },
            {
                "Sid": "DecryptExternalIdSecrets",
                "Effect": "Allow",
                "Action": "kms:Decrypt",
                "Resource": key_arn,
            },
        ],
    }
    iam.put_role_policy(
        RoleName="OrchestratorRole",
        PolicyName="OrchestratorIdentityPolicy",
        PolicyDocument=json.dumps(orch_identity),
    )

    def member_identity(bucket: str) -> dict:
        return {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Sid": "TenantBucketObjectRW",
                    "Effect": "Allow",
                    "Action": ["s3:PutObject", "s3:GetObject", "s3:DeleteObject"],
                    "Resource": f"arn:aws:s3:::{bucket}/*",
                },
                {
                    "Sid": "TenantBucketList",
                    "Effect": "Allow",
                    "Action": "s3:ListBucket",
                    "Resource": f"arn:aws:s3:::{bucket}",
                },
            ],
        }

    iam.put_role_policy(
        RoleName="MemberRole-Tenant-A",
        PolicyName="TenantABucketPolicy",
        PolicyDocument=json.dumps(member_identity("tenant-a-bucket")),
    )
    iam.put_role_policy(
        RoleName="MemberRole-Tenant-B",
        PolicyName="TenantBBucketPolicy",
        PolicyDocument=json.dumps(member_identity("tenant-b-bucket")),
    )

    # agent-runner needs identity-side allow to assume OrchestratorRole
    iam.put_user_policy(
        UserName="agent-runner",
        PolicyName="AssumeOrchestrator",
        PolicyDocument=json.dumps(
            {
                "Version": "2012-10-17",
                "Statement": [
                    {
                        "Effect": "Allow",
                        "Action": "sts:AssumeRole",
                        "Resource": orch_arn,
                    }
                ],
            }
        ),
    )

    # ── SSM pointers ────────────────────────────────────────────────────────
    for name, val in [
        ("/harbor/orchestrator/role-arn", orch_arn),
        ("/harbor/external-id-secret-arn-tenant-a", sec_a_arn),
        ("/harbor/external-id-secret-arn-tenant-b", sec_b_arn),
        ("/harbor/member-role-arn-tenant-a", mem_a_arn),
        ("/harbor/member-role-arn-tenant-b", mem_b_arn),
    ]:
        ssm.put_parameter(Name=name, Value=val, Type="String", Overwrite=True)

    print("setup complete")
    print(f"  OrchestratorRole       = {orch_arn}")
    print(f"  MemberRole-Tenant-A    = {mem_a_arn}")
    print(f"  MemberRole-Tenant-B    = {mem_b_arn}")
    print(f"  external-id secret A   = {sec_a_arn}")
    print(f"  external-id secret B   = {sec_b_arn}")
    print(f"  KMS CMK                = {key_arn}  (alias/cross-account-cmk)")


if __name__ == "__main__":
    main()

"""End-to-end verifier for the cross-account trust topology.

Two phases:
  1. happy path  , assume OrchestratorRole, fetch tenant-A ExternalId from
                   Secrets Manager, assume MemberRole-Tenant-A with that
                   ExternalId, then put an object into tenant-a-bucket.
  2. shape audit , read the trust + identity policy documents directly
                   and assert the SOC2 conditions an auditor would check.
"""

import json
import sys

import boto3

ENDPOINT = "http://localstack:4566"
REGION = "us-east-1"
ACCOUNT_ID = "000000000000"


def base_client(service: str, **overrides):
    kwargs = {"endpoint_url": ENDPOINT, "region_name": REGION, **overrides}
    return boto3.client(service, **kwargs)


def assumed_clients(creds: dict, services):
    out = {}
    for svc in services:
        out[svc] = base_client(
            svc,
            aws_access_key_id=creds["AccessKeyId"],
            aws_secret_access_key=creds["SecretAccessKey"],
            aws_session_token=creds["SessionToken"],
        )
    return out


# ── helpers for shape checks ─────────────────────────────────────────────────
def fail(msg: str):
    print(f"  FAIL: {msg}")
    fail.count += 1


fail.count = 0


def ok(msg: str):
    print(f"  ok:   {msg}")


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


def has_wildcard(value) -> bool:
    if isinstance(value, str):
        return value == "*"
    if isinstance(value, list):
        return any(v == "*" for v in value)
    return False


# ── phase 1: happy path ──────────────────────────────────────────────────────
def happy_path(ssm, iam):
    print("== phase 1: happy path ==")

    orch_arn = ssm.get_parameter(Name="/harbor/orchestrator/role-arn")["Parameter"]["Value"]
    sec_a_arn = ssm.get_parameter(Name="/harbor/external-id-secret-arn-tenant-a")["Parameter"]["Value"]
    mem_a_arn = ssm.get_parameter(Name="/harbor/member-role-arn-tenant-a")["Parameter"]["Value"]

    # Create access keys for agent-runner (the orchestrator's runtime identity).
    key = iam.create_access_key(UserName="agent-runner")["AccessKey"]
    user_sts = base_client(
        "sts",
        aws_access_key_id=key["AccessKeyId"],
        aws_secret_access_key=key["SecretAccessKey"],
    )

    # 1) agent-runner assumes OrchestratorRole.
    orch_creds = user_sts.assume_role(
        RoleArn=orch_arn, RoleSessionName="orchestrator-session"
    )["Credentials"]
    print(f"  assumed OrchestratorRole → AKID {orch_creds['AccessKeyId'][:8]}…")

    # 2) Orchestrator pulls the tenant-A ExternalId from Secrets Manager.
    orch = assumed_clients(orch_creds, ["secretsmanager", "sts"])
    ext_id = orch["secretsmanager"].get_secret_value(SecretId=sec_a_arn)["SecretString"]
    print(f"  fetched tenant-A ExternalId  (len={len(ext_id)})")

    # 3) Orchestrator assumes MemberRole-Tenant-A with the ExternalId.
    member_creds = orch["sts"].assume_role(
        RoleArn=mem_a_arn,
        RoleSessionName="tenant-a-session",
        ExternalId=ext_id,
    )["Credentials"]
    print(f"  assumed MemberRole-Tenant-A → AKID {member_creds['AccessKeyId'][:8]}…")

    # 4) Member writes an object to tenant-a-bucket.
    member = assumed_clients(member_creds, ["s3"])
    member["s3"].put_object(
        Bucket="tenant-a-bucket",
        Key="proof-of-life.txt",
        Body=b"happy-path-ok",
    )
    body = member["s3"].get_object(Bucket="tenant-a-bucket", Key="proof-of-life.txt")["Body"].read()
    assert body == b"happy-path-ok", f"unexpected body: {body!r}"
    print("  put + get on tenant-a-bucket → ok")
    print()


# ── phase 2: shape audit ─────────────────────────────────────────────────────
def shape_audit(iam, sm):
    print("== phase 2: policy-shape audit ==")

    orch_role = iam.get_role(RoleName="OrchestratorRole")["Role"]
    orch_arn = orch_role["Arn"]
    mem_a = iam.get_role(RoleName="MemberRole-Tenant-A")["Role"]
    mem_b = iam.get_role(RoleName="MemberRole-Tenant-B")["Role"]
    mem_a_arn = mem_a["Arn"]
    mem_b_arn = mem_b["Arn"]

    sec_a = sm.describe_secret(SecretId="harbor/cross-account/external-id-tenant-a")
    sec_b = sm.describe_secret(SecretId="harbor/cross-account/external-id-tenant-b")
    sec_a_arn = sec_a["ARN"]
    sec_b_arn = sec_b["ARN"]
    ext_a = sm.get_secret_value(SecretId=sec_a_arn)["SecretString"]
    ext_b = sm.get_secret_value(SecretId=sec_b_arn)["SecretString"]

    # ── path scoping ────────────────────────────────────────────────────────
    if orch_role["Path"] == "/orchestrator/":
        ok("OrchestratorRole sits under /orchestrator/")
    else:
        fail(f"OrchestratorRole path is {orch_role['Path']!r}, expected /orchestrator/")
    for r, name in ((mem_a, "MemberRole-Tenant-A"), (mem_b, "MemberRole-Tenant-B")):
        if r["Path"] == "/member/":
            ok(f"{name} sits under /member/")
        else:
            fail(f"{name} path is {r['Path']!r}, expected /member/")

    # ── ExternalIds differ ──────────────────────────────────────────────────
    if ext_a != ext_b:
        ok("tenant-A and tenant-B ExternalIds differ")
    else:
        fail("ExternalIds are identical across tenants , defeats the purpose")
    if len(ext_a) >= 32 and len(ext_b) >= 32:
        ok(f"both ExternalIds ≥ 32 chars (a={len(ext_a)}, b={len(ext_b)})")
    else:
        fail(f"ExternalId too short (a={len(ext_a)}, b={len(ext_b)})")

    # ── trust documents ─────────────────────────────────────────────────────
    def assert_member_trust(role, expected_ext_id, label):
        doc = role["AssumeRolePolicyDocument"]
        if isinstance(doc, str):
            doc = json.loads(doc)
        statements = as_list(doc["Statement"])
        if len(statements) != 1:
            fail(f"{label}: expected exactly 1 trust statement, got {len(statements)}")
            return
        stmt = statements[0]

        # principal = orchestrator role arn (not wildcard)
        principal = stmt.get("Principal")
        if principal == "*" or principal == {"AWS": "*"}:
            fail(f"{label}: trust principal is wildcard")
        elif isinstance(principal, dict) and principal.get("AWS") == orch_arn:
            ok(f"{label}: principal is OrchestratorRole arn (not *)")
        else:
            fail(f"{label}: principal is {principal!r}, expected {{'AWS': '{orch_arn}'}}")

        # action is exactly sts:AssumeRole, no wildcards
        action = stmt.get("Action")
        if action == "sts:AssumeRole" or action == ["sts:AssumeRole"]:
            ok(f"{label}: action is exactly sts:AssumeRole")
        else:
            fail(f"{label}: action is {action!r}, expected exactly sts:AssumeRole")

        cond = stmt.get("Condition") or {}

        # ExternalId guard against third-party impersonation
        ext = (cond.get("StringEquals") or {}).get("sts:ExternalId")
        if ext == expected_ext_id:
            ok(f"{label}: StringEquals[sts:ExternalId] matches secret value")
        else:
            fail(f"{label}: sts:ExternalId condition missing or wrong (got {ext!r})")

        # SourceArn guard against confused-deputy
        src = None
        for op in ("ArnEquals", "ArnLike"):
            v = (cond.get(op) or {}).get("aws:SourceArn")
            if v is not None:
                src = (op, v)
                break
        if src and src[1] == orch_arn:
            ok(f"{label}: {src[0]}[aws:SourceArn] equals OrchestratorRole arn")
        else:
            fail(f"{label}: aws:SourceArn condition missing or not equal to orchestrator arn (got {src!r})")

    assert_member_trust(mem_a, ext_a, "MemberRole-Tenant-A trust")
    assert_member_trust(mem_b, ext_b, "MemberRole-Tenant-B trust")

    # ── orchestrator identity policy: explicit member arns, no wildcards ───
    orch_id = json.loads(
        iam.get_role_policy(RoleName="OrchestratorRole", PolicyName="OrchestratorIdentityPolicy")[
            "PolicyDocument"
        ]
        if isinstance(
            iam.get_role_policy(RoleName="OrchestratorRole", PolicyName="OrchestratorIdentityPolicy")[
                "PolicyDocument"
            ],
            str,
        )
        else json.dumps(
            iam.get_role_policy(RoleName="OrchestratorRole", PolicyName="OrchestratorIdentityPolicy")[
                "PolicyDocument"
            ]
        )
    )
    assume_stmt = next(
        (s for s in as_list(orch_id["Statement"]) if "sts:AssumeRole" in as_list(s["Action"])),
        None,
    )
    if assume_stmt is None:
        fail("orchestrator identity: no sts:AssumeRole statement")
    else:
        res = as_list(assume_stmt["Resource"])
        if has_wildcard(res):
            fail(f"orchestrator identity: sts:AssumeRole has wildcard resource ({res!r})")
        elif set(res) == {mem_a_arn, mem_b_arn}:
            ok("orchestrator identity: sts:AssumeRole enumerates exactly the two member arns")
        else:
            fail(f"orchestrator identity: AssumeRole resources are {res!r}, expected {{{mem_a_arn}, {mem_b_arn}}}")

    secret_stmt = next(
        (
            s
            for s in as_list(orch_id["Statement"])
            if "secretsmanager:GetSecretValue" in as_list(s["Action"])
        ),
        None,
    )
    if secret_stmt is None:
        fail("orchestrator identity: no secretsmanager:GetSecretValue statement")
    else:
        res = as_list(secret_stmt["Resource"])
        if has_wildcard(res):
            fail("orchestrator identity: GetSecretValue has wildcard resource")
        elif set(res) == {sec_a_arn, sec_b_arn}:
            ok("orchestrator identity: GetSecretValue scoped to the two known secret arns")
        else:
            fail(f"orchestrator identity: GetSecretValue resources are {res!r}")

    # ── member identity policies: scoped to own bucket only ────────────────
    def assert_member_identity(role_name, policy_name, own_bucket, other_bucket):
        doc = iam.get_role_policy(RoleName=role_name, PolicyName=policy_name)["PolicyDocument"]
        if isinstance(doc, str):
            doc = json.loads(doc)
        own_arns = {f"arn:aws:s3:::{own_bucket}", f"arn:aws:s3:::{own_bucket}/*"}
        seen = set()
        for stmt in as_list(doc["Statement"]):
            actions = as_list(stmt["Action"])
            resources = as_list(stmt["Resource"])
            if has_wildcard(actions):
                fail(f"{role_name}: Action wildcard present")
            if has_wildcard(resources):
                fail(f"{role_name}: Resource wildcard present")
            for r in resources:
                if other_bucket in r:
                    fail(f"{role_name}: references other tenant's bucket: {r}")
                seen.add(r)
        if seen <= own_arns:
            ok(f"{role_name}: identity policy references only {own_bucket} arns")
        else:
            fail(f"{role_name}: unexpected resources {seen - own_arns}")

    assert_member_identity("MemberRole-Tenant-A", "TenantABucketPolicy", "tenant-a-bucket", "tenant-b-bucket")
    assert_member_identity("MemberRole-Tenant-B", "TenantBBucketPolicy", "tenant-b-bucket", "tenant-a-bucket")

    # ── secret resource policies admit the orchestrator ────────────────────
    def assert_secret_resource_policy(secret_arn, label):
        rp = sm.get_resource_policy(SecretId=secret_arn).get("ResourcePolicy")
        if not rp:
            fail(f"{label}: no resource policy attached")
            return
        doc = json.loads(rp) if isinstance(rp, str) else rp
        for stmt in as_list(doc["Statement"]):
            principal = stmt.get("Principal", {})
            principals = as_list(principal.get("AWS", [])) if isinstance(principal, dict) else []
            if (
                stmt.get("Effect") == "Allow"
                and "secretsmanager:GetSecretValue" in as_list(stmt["Action"])
                and orch_arn in principals
            ):
                ok(f"{label}: resource policy admits OrchestratorRole for GetSecretValue")
                return
        fail(f"{label}: resource policy does not name OrchestratorRole as principal for GetSecretValue")

    assert_secret_resource_policy(sec_a_arn, "tenant-A secret")
    assert_secret_resource_policy(sec_b_arn, "tenant-B secret")

    # ── KMS scoping in orchestrator policy ─────────────────────────────────
    kms_stmt = next(
        (s for s in as_list(orch_id["Statement"]) if any(a.startswith("kms:") for a in as_list(s["Action"]))),
        None,
    )
    if kms_stmt is None:
        fail("orchestrator identity: no kms statement (needed to decrypt secrets)")
    else:
        if has_wildcard(as_list(kms_stmt["Resource"])):
            fail("orchestrator identity: kms statement has wildcard resource")
        elif has_wildcard(as_list(kms_stmt["Action"])):
            fail("orchestrator identity: kms statement has wildcard action")
        else:
            ok("orchestrator identity: kms statement is scoped (no * action, no * resource)")

    print()


def main():
    ssm = base_client("ssm")
    iam = base_client("iam")
    sm = base_client("secretsmanager")

    happy_path(ssm, iam)
    shape_audit(iam, sm)

    if fail.count:
        print(f"VERIFIER: {fail.count} failure(s)")
        sys.exit(1)
    print("VERIFIER: all checks passed")


if __name__ == "__main__":
    main()