iam-cross-account-externalid-sourcearn

#!/usr/bin/env python3
"""
Build the cross-account-style trust scaffold in LocalStack:
- KMS CMK + alias
- Two ExternalId secrets (KMS-encrypted), with resource policies
- Orchestrator role (trusted by agent-runner IAM user)
- Two member roles, one per tenant, with hardened trust (ExternalId + SourceArn)
- Per-tenant S3 buckets
- Tightly scoped identity policies
- SSM pointers for the verifier
"""
import json
import os
import secrets
import string
import sys

import boto3
from botocore.config import Config
from botocore.exceptions import ClientError

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

NAMES = {
    "user_orchestrator": "agent-runner",
    "user_attacker": "attacker-runner",
    "role_orchestrator": "OrchestratorRole",
    "role_member_a": "MemberRole-Tenant-A",
    "role_member_b": "MemberRole-Tenant-B",
    "kms_alias": "alias/cross-account-cmk",
    "secret_a": "harbor/cross-account/external-id-tenant-a",
    "secret_b": "harbor/cross-account/external-id-tenant-b",
    "bucket_a": "tenant-a-bucket",
    "bucket_b": "tenant-b-bucket",
    "ssm_orch_arn": "/harbor/orchestrator/role-arn",
    "ssm_secret_a": "/harbor/external-id-secret-arn-tenant-a",
    "ssm_secret_b": "/harbor/external-id-secret-arn-tenant-b",
    "ssm_member_a": "/harbor/member-role-arn-tenant-a",
    "ssm_member_b": "/harbor/member-role-arn-tenant-b",
}

PATH_ORCH = "/orchestrator/"
PATH_MEMBER = "/member/"

ORCHESTRATOR_ROLE_ARN = (
    f"arn:aws:iam::{ACCOUNT_ID}:role{PATH_ORCH}{NAMES['role_orchestrator']}"
)
MEMBER_A_ARN = f"arn:aws:iam::{ACCOUNT_ID}:role{PATH_MEMBER}{NAMES['role_member_a']}"
MEMBER_B_ARN = f"arn:aws:iam::{ACCOUNT_ID}:role{PATH_MEMBER}{NAMES['role_member_b']}"
USER_ORCH_ARN = f"arn:aws:iam::{ACCOUNT_ID}:user/{NAMES['user_orchestrator']}"
USER_ATTACKER_ARN = f"arn:aws:iam::{ACCOUNT_ID}:user/{NAMES['user_attacker']}"
BUCKET_A_ARN = f"arn:aws:s3:::{NAMES['bucket_a']}"
BUCKET_B_ARN = f"arn:aws:s3:::{NAMES['bucket_b']}"


def client(service):
    return boto3.client(
        service,
        endpoint_url=ENDPOINT,
        region_name=REGION,
        config=Config(retries={"max_attempts": 3}),
    )


def gen_external_id():
    """At least 32 chars, URL-safe alphabet."""
    alphabet = string.ascii_letters + string.digits
    return "".join(secrets.choice(alphabet) for _ in range(48))


def safe_create_user(iam, name):
    try:
        iam.create_user(UserName=name)
    except ClientError as e:
        if e.response["Error"]["Code"] != "EntityAlreadyExists":
            raise
    return iam.get_user(UserName=name)["User"]["Arn"]


def safe_create_access_key(iam, name):
    keys = iam.list_access_keys(UserName=name)["AccessKeyMetadata"]
    for k in keys:
        iam.delete_access_key(UserName=name, AccessKeyId=k["AccessKeyId"])
    resp = iam.create_access_key(UserName=name)["AccessKey"]
    return resp["AccessKeyId"], resp["SecretAccessKey"]


def safe_create_role(iam, name, path, trust_doc):
    try:
        iam.create_role(
            RoleName=name,
            Path=path,
            AssumeRolePolicyDocument=json.dumps(trust_doc),
        )
    except ClientError as e:
        if e.response["Error"]["Code"] != "EntityAlreadyExists":
            raise
        iam.update_assume_role_policy(
            RoleName=name, PolicyDocument=json.dumps(trust_doc)
        )
    return iam.get_role(RoleName=name)["Role"]["Arn"]


def put_inline(iam, role, policy_name, doc):
    iam.put_role_policy(
        RoleName=role, PolicyName=policy_name, PolicyDocument=json.dumps(doc)
    )


def put_user_inline(iam, user, policy_name, doc):
    iam.put_user_policy(
        UserName=user, PolicyName=policy_name, PolicyDocument=json.dumps(doc)
    )


def kms_setup(kms):
    # Reuse existing alias if it exists.
    try:
        existing = kms.describe_key(KeyId=NAMES["kms_alias"])
        return existing["KeyMetadata"]["Arn"]
    except ClientError as e:
        if e.response["Error"]["Code"] not in ("NotFoundException", "NotFound"):
            raise

    key_policy = {
        "Version": "2012-10-17",
        "Id": "cross-account-cmk-policy",
        "Statement": [
            {
                "Sid": "RootAccountAdmin",
                "Effect": "Allow",
                "Principal": {"AWS": f"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": "OrchestratorDecryptOnly",
                "Effect": "Allow",
                "Principal": {"AWS": ORCHESTRATOR_ROLE_ARN},
                "Action": ["kms:Decrypt", "kms:DescribeKey"],
                "Resource": "*",
            },
            {
                "Sid": "SecretsManagerUseOfTheKey",
                "Effect": "Allow",
                "Principal": {"Service": "secretsmanager.amazonaws.com"},
                "Action": [
                    "kms:Encrypt",
                    "kms:Decrypt",
                    "kms:ReEncrypt*",
                    "kms:GenerateDataKey*",
                    "kms:DescribeKey",
                ],
                "Resource": "*",
                "Condition": {
                    "StringEquals": {"kms:ViaService": f"secretsmanager.{REGION}.amazonaws.com"}
                },
            },
        ],
    }

    key = kms.create_key(
        Description="Cross-account ExternalId secret encryption",
        Policy=json.dumps(key_policy),
        KeyUsage="ENCRYPT_DECRYPT",
    )["KeyMetadata"]
    kms.create_alias(AliasName=NAMES["kms_alias"], TargetKeyId=key["KeyId"])
    return key["Arn"]


def secret_setup(sm, name, kms_key_arn, value, allow_principal_arn):
    try:
        sm.delete_secret(SecretId=name, ForceDeleteWithoutRecovery=True)
    except ClientError as e:
        if e.response["Error"]["Code"] != "ResourceNotFoundException":
            raise

    created = sm.create_secret(
        Name=name,
        KmsKeyId=kms_key_arn,
        SecretString=value,
        Description=f"ExternalId for {name}",
    )
    secret_arn = created["ARN"]

    resource_policy = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "AllowOrchestratorRead",
                "Effect": "Allow",
                "Principal": {"AWS": allow_principal_arn},
                "Action": ["secretsmanager:GetSecretValue", "secretsmanager:DescribeSecret"],
                "Resource": secret_arn,
            }
        ],
    }
    sm.put_resource_policy(
        SecretId=name, ResourcePolicy=json.dumps(resource_policy)
    )
    return secret_arn


def s3_setup(s3, bucket):
    try:
        s3.create_bucket(Bucket=bucket)
    except ClientError as e:
        if e.response["Error"]["Code"] not in (
            "BucketAlreadyOwnedByYou",
            "BucketAlreadyExists",
        ):
            raise


def ssm_put(ssm, name, value):
    ssm.put_parameter(Name=name, Value=value, Type="String", Overwrite=True)


def main():
    iam = client("iam")
    kms = client("kms")
    sm = client("secretsmanager")
    s3 = client("s3")
    ssm = client("ssm")

    # 1. KMS first , secrets need it.
    kms_key_arn = kms_setup(kms)

    # 2. Users.
    safe_create_user(iam, NAMES["user_orchestrator"])
    safe_create_user(iam, NAMES["user_attacker"])
    orch_ak, orch_sk = safe_create_access_key(iam, NAMES["user_orchestrator"])
    atk_ak, atk_sk = safe_create_access_key(iam, NAMES["user_attacker"])

    # 3. Orchestrator role , trusted by the agent-runner user.
    orch_trust = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "TrustAgentRunnerUser",
                "Effect": "Allow",
                "Principal": {"AWS": USER_ORCH_ARN},
                "Action": "sts:AssumeRole",
            }
        ],
    }
    orch_arn = safe_create_role(
        iam, NAMES["role_orchestrator"], PATH_ORCH, orch_trust
    )
    assert orch_arn == ORCHESTRATOR_ROLE_ARN, (orch_arn, ORCHESTRATOR_ROLE_ARN)

    # 4. ExternalIds (distinct, >=32 chars).
    ext_a = gen_external_id()
    ext_b = gen_external_id()
    while ext_b == ext_a:
        ext_b = gen_external_id()
    assert len(ext_a) >= 32 and len(ext_b) >= 32

    # 5. Secrets , store ExternalId values, encrypted with the CMK.
    secret_a_arn = secret_setup(
        sm, NAMES["secret_a"], kms_key_arn, ext_a, ORCHESTRATOR_ROLE_ARN
    )
    secret_b_arn = secret_setup(
        sm, NAMES["secret_b"], kms_key_arn, ext_b, ORCHESTRATOR_ROLE_ARN
    )

    # 6. Member-role trust documents , the heart of the change.
    def member_trust(external_id_value):
        return {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Sid": "OrchestratorAssumeWithExternalIdAndSourceArn",
                    "Effect": "Allow",
                    "Principal": {"AWS": ORCHESTRATOR_ROLE_ARN},
                    "Action": "sts:AssumeRole",
                    "Condition": {
                        "StringEquals": {"sts:ExternalId": external_id_value},
                        "ArnLike": {"aws:SourceArn": ORCHESTRATOR_ROLE_ARN},
                    },
                }
            ],
        }

    member_a_arn = safe_create_role(
        iam, NAMES["role_member_a"], PATH_MEMBER, member_trust(ext_a)
    )
    member_b_arn = safe_create_role(
        iam, NAMES["role_member_b"], PATH_MEMBER, member_trust(ext_b)
    )
    assert member_a_arn == MEMBER_A_ARN and member_b_arn == MEMBER_B_ARN

    # 7. Member identity policies , scoped to their own bucket only.
    put_inline(
        iam,
        NAMES["role_member_a"],
        "TenantABucketAccess",
        {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Sid": "TenantABucketObjects",
                    "Effect": "Allow",
                    "Action": ["s3:PutObject", "s3:GetObject", "s3:DeleteObject"],
                    "Resource": f"{BUCKET_A_ARN}/*",
                },
                {
                    "Sid": "TenantABucketList",
                    "Effect": "Allow",
                    "Action": ["s3:ListBucket", "s3:GetBucketLocation"],
                    "Resource": BUCKET_A_ARN,
                },
            ],
        },
    )
    put_inline(
        iam,
        NAMES["role_member_b"],
        "TenantBBucketAccess",
        {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Sid": "TenantBBucketObjects",
                    "Effect": "Allow",
                    "Action": ["s3:PutObject", "s3:GetObject", "s3:DeleteObject"],
                    "Resource": f"{BUCKET_B_ARN}/*",
                },
                {
                    "Sid": "TenantBBucketList",
                    "Effect": "Allow",
                    "Action": ["s3:ListBucket", "s3:GetBucketLocation"],
                    "Resource": BUCKET_B_ARN,
                },
            ],
        },
    )

    # 8. Orchestrator identity policy , explicit member arns + scoped secret reads + scoped kms decrypt.
    put_inline(
        iam,
        NAMES["role_orchestrator"],
        "OrchestratorFanOut",
        {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Sid": "AssumeNamedMembersOnly",
                    "Effect": "Allow",
                    "Action": "sts:AssumeRole",
                    "Resource": [MEMBER_A_ARN, MEMBER_B_ARN],
                },
                {
                    "Sid": "ReadExternalIdSecretsOnly",
                    "Effect": "Allow",
                    "Action": "secretsmanager:GetSecretValue",
                    "Resource": [secret_a_arn, secret_b_arn],
                },
                {
                    "Sid": "DecryptExternalIdSecretsKMS",
                    "Effect": "Allow",
                    "Action": ["kms:Decrypt", "kms:DescribeKey"],
                    "Resource": kms_key_arn,
                    "Condition": {
                        "StringEquals": {
                            "kms:ViaService": f"secretsmanager.{REGION}.amazonaws.com"
                        }
                    },
                },
            ],
        },
    )

    # 9. agent-runner user policy , only sts:AssumeRole on the orchestrator role.
    put_user_inline(
        iam,
        NAMES["user_orchestrator"],
        "AssumeOrchestratorOnly",
        {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Sid": "AssumeOrchestratorRole",
                    "Effect": "Allow",
                    "Action": "sts:AssumeRole",
                    "Resource": ORCHESTRATOR_ROLE_ARN,
                }
            ],
        },
    )

    # 10. Buckets.
    s3_setup(s3, NAMES["bucket_a"])
    s3_setup(s3, NAMES["bucket_b"])

    # 11. SSM pointers.
    ssm_put(ssm, NAMES["ssm_orch_arn"], ORCHESTRATOR_ROLE_ARN)
    ssm_put(ssm, NAMES["ssm_secret_a"], secret_a_arn)
    ssm_put(ssm, NAMES["ssm_secret_b"], secret_b_arn)
    ssm_put(ssm, NAMES["ssm_member_a"], MEMBER_A_ARN)
    ssm_put(ssm, NAMES["ssm_member_b"], MEMBER_B_ARN)

    # Stash credentials so the happy-path script can use them. Local-only.
    creds_path = "/app/setup/creds.json"
    with open(creds_path, "w") as fh:
        json.dump(
            {
                "agent_runner": {"access_key": orch_ak, "secret_key": orch_sk},
                "attacker_runner": {"access_key": atk_ak, "secret_key": atk_sk},
            },
            fh,
            indent=2,
        )
    os.chmod(creds_path, 0o600)

    print("BUILD OK")
    print(f"  KMS key arn:        {kms_key_arn}")
    print(f"  Orchestrator role:  {ORCHESTRATOR_ROLE_ARN}")
    print(f"  Member A role:      {MEMBER_A_ARN}")
    print(f"  Member B role:      {MEMBER_B_ARN}")
    print(f"  Secret A arn:       {secret_a_arn}")
    print(f"  Secret B arn:       {secret_b_arn}")
    print(f"  Tenant A bucket:    {BUCKET_A_ARN}")
    print(f"  Tenant B bucket:    {BUCKET_B_ARN}")


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

#!/usr/bin/env python3
"""
Happy path:
  1. Authenticate as agent-runner IAM user.
  2. Assume OrchestratorRole.
  3. Read tenant-A ExternalId from Secrets Manager.
  4. Assume MemberRole-Tenant-A using that ExternalId.
  5. PutObject into tenant-a-bucket.

Demonstrates that the chain works end-to-end with real ExternalId presentation.
LocalStack does not enforce IAM, but the call sequence is the production-shaped
flow and it succeeds (the would-be denial branches are exercised in
verify.py via document-shape inspection).
"""
import json
import os

import boto3

ENDPOINT = os.environ["AWS_ENDPOINT_URL"]
REGION = os.environ["AWS_DEFAULT_REGION"]


def make_client(service, creds=None):
    kw = dict(endpoint_url=ENDPOINT, region_name=REGION)
    if creds:
        kw.update(
            aws_access_key_id=creds["AccessKeyId"],
            aws_secret_access_key=creds["SecretAccessKey"],
        )
        if "SessionToken" in creds:
            kw["aws_session_token"] = creds["SessionToken"]
    return boto3.client(service, **kw)


def main():
    with open("/app/setup/creds.json") as fh:
        local_creds = json.load(fh)
    agent = local_creds["agent_runner"]

    # Step 1+2: agent-runner user creds → AssumeRole(OrchestratorRole).
    ssm_root = boto3.client(
        "ssm",
        endpoint_url=ENDPOINT,
        region_name=REGION,
        aws_access_key_id=agent["access_key"],
        aws_secret_access_key=agent["secret_key"],
    )
    orch_arn = ssm_root.get_parameter(Name="/harbor/orchestrator/role-arn")[
        "Parameter"
    ]["Value"]
    secret_a_arn = ssm_root.get_parameter(
        Name="/harbor/external-id-secret-arn-tenant-a"
    )["Parameter"]["Value"]
    member_a_arn = ssm_root.get_parameter(Name="/harbor/member-role-arn-tenant-a")[
        "Parameter"
    ]["Value"]

    sts_user = boto3.client(
        "sts",
        endpoint_url=ENDPOINT,
        region_name=REGION,
        aws_access_key_id=agent["access_key"],
        aws_secret_access_key=agent["secret_key"],
    )
    orch_creds = sts_user.assume_role(
        RoleArn=orch_arn, RoleSessionName="agent-runner-to-orchestrator"
    )["Credentials"]
    print(f"[1] assumed orchestrator: {orch_arn}")

    # Step 3: orchestrator reads tenant-A ExternalId.
    sm = make_client("secretsmanager", orch_creds)
    secret_value = sm.get_secret_value(SecretId=secret_a_arn)["SecretString"]
    print(f"[2] retrieved external-id from {secret_a_arn} (len={len(secret_value)})")

    # Step 4: orchestrator assumes MemberRole-Tenant-A WITH ExternalId.
    sts_orch = make_client("sts", orch_creds)
    member_creds = sts_orch.assume_role(
        RoleArn=member_a_arn,
        RoleSessionName="orchestrator-to-tenant-a",
        ExternalId=secret_value,
    )["Credentials"]
    print(f"[3] assumed member role: {member_a_arn} with ExternalId")

    # Step 5: PutObject into tenant-a-bucket as the member role.
    s3 = make_client("s3", member_creds)
    payload = b'{"event": "happy-path-write", "tenant": "A"}'
    s3.put_object(
        Bucket="tenant-a-bucket",
        Key="audits/happy-path.json",
        Body=payload,
        ContentType="application/json",
    )
    head = s3.head_object(Bucket="tenant-a-bucket", Key="audits/happy-path.json")
    print(
        f"[4] wrote tenant-a-bucket/audits/happy-path.json "
        f"({head['ContentLength']} bytes)"
    )

    print("HAPPY PATH OK")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Auditor-style verifier. Reads policy DOCUMENTS directly (no runtime calls past
fetching the JSON) and asserts each invariant from the spec. Exits non-zero on
any failure with a precise message naming the violated rule.
"""
import json
import os
import sys

import boto3

ENDPOINT = os.environ["AWS_ENDPOINT_URL"]
REGION = os.environ["AWS_DEFAULT_REGION"]
ACCOUNT_ID = "000000000000"


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


# ----- helpers ---------------------------------------------------------------

FAILS = []
PASSES = []


def check(label, ok, detail=""):
    if ok:
        PASSES.append(label)
        print(f"  PASS  {label}")
    else:
        FAILS.append((label, detail))
        print(f"  FAIL  {label} :: {detail}")


def stmts(doc):
    s = doc.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 has_wildcard_s3_or_kms(doc):
    """True if any statement targets s3:/kms: with Action=* or Resource=*."""
    for st in stmts(doc):
        actions = as_list(st.get("Action"))
        resources = as_list(st.get("Resource"))
        action_wild = any(a == "*" or a == "s3:*" or a == "kms:*" for a in actions)
        action_touches = any(
            a == "*" or a.startswith("s3:") or a.startswith("kms:") for a in actions
        )
        if action_wild:
            return True
        if action_touches and any(r == "*" for r in resources):
            return True
    return False


# ----- pull every artifact ---------------------------------------------------

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

    # SSM pointers , the verifier finds everything via these.
    def ssm_val(name):
        return ssm.get_parameter(Name=name)["Parameter"]["Value"]

    orch_arn = ssm_val("/harbor/orchestrator/role-arn")
    member_a_arn = ssm_val("/harbor/member-role-arn-tenant-a")
    member_b_arn = ssm_val("/harbor/member-role-arn-tenant-b")
    secret_a_arn = ssm_val("/harbor/external-id-secret-arn-tenant-a")
    secret_b_arn = ssm_val("/harbor/external-id-secret-arn-tenant-b")

    print("== Discovered via SSM ==")
    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()

    # Roles + trust documents.
    orch_role = iam.get_role(RoleName="OrchestratorRole")["Role"]
    member_a = iam.get_role(RoleName="MemberRole-Tenant-A")["Role"]
    member_b = iam.get_role(RoleName="MemberRole-Tenant-B")["Role"]

    # boto returns trust docs already-decoded.
    def trust(role):
        td = role["AssumeRolePolicyDocument"]
        return td if isinstance(td, dict) else json.loads(td)

    trust_a = trust(member_a)
    trust_b = trust(member_b)

    # Path discovery: spec says verifier uses /orchestrator/ and /member/ paths.
    orch_by_path = iam.list_roles(PathPrefix="/orchestrator/")["Roles"]
    member_by_path = iam.list_roles(PathPrefix="/member/")["Roles"]

    print("== Path enumeration ==")
    check(
        "orchestrator role lives under /orchestrator/",
        any(r["RoleName"] == "OrchestratorRole" for r in orch_by_path),
    )
    check(
        "both member roles live under /member/",
        {r["RoleName"] for r in member_by_path}
        == {"MemberRole-Tenant-A", "MemberRole-Tenant-B"},
        detail=str([r["RoleName"] for r in member_by_path]),
    )

    # ----- per-tenant trust shape ------------------------------------------
    def check_member_trust(label, doc, expected_external_id, expected_orch_arn):
        sts_list = stmts(doc)
        check(f"{label}: trust has exactly one statement", len(sts_list) == 1,
              detail=f"got {len(sts_list)}")
        st = sts_list[0]

        # Action
        actions = as_list(st.get("Action"))
        check(
            f"{label}: action is exactly sts:AssumeRole",
            actions == ["sts:AssumeRole"],
            detail=f"got {actions}",
        )

        # Effect
        check(f"{label}: effect Allow", st.get("Effect") == "Allow")

        # Principal: must be the orchestrator role arn, NOT *
        principal = st.get("Principal", {})
        aws_principals = as_list(principal.get("AWS")) if isinstance(principal, dict) else []
        check(
            f"{label}: principal is the orchestrator role arn (not *)",
            aws_principals == [expected_orch_arn],
            detail=f"got principal={principal}",
        )
        check(
            f"{label}: principal is not wildcard",
            principal != "*"
            and (not isinstance(principal, dict) or principal.get("AWS") != "*"),
        )

        # Conditions
        cond = st.get("Condition", {})
        ext_eq = cond.get("StringEquals", {})
        check(
            f"{label}: Condition.StringEquals['sts:ExternalId'] matches secret value",
            ext_eq.get("sts:ExternalId") == expected_external_id,
            detail=f"got {ext_eq.get('sts:ExternalId')!r}",
        )
        # SourceArn , accept ArnEquals or ArnLike (spec allows either).
        src_arn_block = None
        for op in ("ArnEquals", "ArnLike"):
            if op in cond and "aws:SourceArn" in cond[op]:
                src_arn_block = (op, cond[op]["aws:SourceArn"])
                break
        check(
            f"{label}: Condition has aws:SourceArn under ArnEquals or ArnLike",
            src_arn_block is not None,
            detail=str(cond),
        )
        if src_arn_block is not None:
            check(
                f"{label}: aws:SourceArn equals orchestrator role arn",
                src_arn_block[1] == expected_orch_arn,
                detail=f"got {src_arn_block[1]!r}",
            )

    # Extract the secret VALUES , trust docs reference these.
    val_a = sm.get_secret_value(SecretId=secret_a_arn)["SecretString"]
    val_b = sm.get_secret_value(SecretId=secret_b_arn)["SecretString"]

    print()
    print("== Member trust documents ==")
    check_member_trust("MemberRole-Tenant-A", trust_a, val_a, orch_arn)
    check_member_trust("MemberRole-Tenant-B", trust_b, val_b, orch_arn)

    # ExternalIds: distinct, both >= 32.
    print()
    print("== ExternalId hygiene ==")
    check("tenant-A ExternalId >= 32 chars", len(val_a) >= 32, detail=f"len={len(val_a)}")
    check("tenant-B ExternalId >= 32 chars", len(val_b) >= 32, detail=f"len={len(val_b)}")
    check("ExternalIds differ between tenants", val_a != val_b)

    # ----- least-privilege checks on identity policies ---------------------
    print()
    print("== Orchestrator identity policy ==")
    orch_policy_names = iam.list_role_policies(RoleName="OrchestratorRole")["PolicyNames"]
    check("orchestrator has at least one inline policy", bool(orch_policy_names))
    orch_doc = None
    for n in orch_policy_names:
        d = iam.get_role_policy(RoleName="OrchestratorRole", PolicyName=n)["PolicyDocument"]
        orch_doc = d if isinstance(d, dict) else json.loads(d)
    assume_resources = []
    secret_resources = []
    for st in stmts(orch_doc):
        actions = as_list(st.get("Action"))
        resources = as_list(st.get("Resource"))
        if "sts:AssumeRole" in actions:
            assume_resources.extend(resources)
        if "secretsmanager:GetSecretValue" in actions:
            secret_resources.extend(resources)
    check(
        "orchestrator AssumeRole resources enumerate exactly the two member arns",
        sorted(assume_resources) == sorted([member_a_arn, member_b_arn]),
        detail=f"got {assume_resources}",
    )
    check(
        "orchestrator AssumeRole has no wildcard resource",
        all(r != "*" for r in assume_resources),
    )
    check(
        "orchestrator GetSecretValue scoped to the two known secret arns only",
        sorted(secret_resources) == sorted([secret_a_arn, secret_b_arn]),
        detail=f"got {secret_resources}",
    )
    check(
        "orchestrator policy has no Action:* or Resource:* on s3/kms",
        not has_wildcard_s3_or_kms(orch_doc),
    )

    # ----- member identity policies ---------------------------------------
    print()
    print("== Member identity policies ==")
    for tenant, role_name, own_bucket, other_bucket in [
        ("A", "MemberRole-Tenant-A", "tenant-a-bucket", "tenant-b-bucket"),
        ("B", "MemberRole-Tenant-B", "tenant-b-bucket", "tenant-a-bucket"),
    ]:
        names = iam.list_role_policies(RoleName=role_name)["PolicyNames"]
        check(f"member {tenant} has at least one inline policy", bool(names))
        merged = {"Statement": []}
        for n in names:
            d = iam.get_role_policy(RoleName=role_name, PolicyName=n)["PolicyDocument"]
            d = d if isinstance(d, dict) else json.loads(d)
            merged["Statement"].extend(stmts(d))
        all_resources = []
        for st in stmts(merged):
            all_resources.extend(as_list(st.get("Resource")))
        check(
            f"member {tenant}: every resource references only its own tenant bucket",
            all(
                r.startswith(f"arn:aws:s3:::{own_bucket}")
                or r == f"arn:aws:s3:::{own_bucket}"
                for r in all_resources
            ),
            detail=f"resources={all_resources}",
        )
        check(
            f"member {tenant}: no resource references the other tenant bucket",
            not any(other_bucket in r for r in all_resources),
        )
        check(
            f"member {tenant}: no Resource:* on s3/kms",
            not has_wildcard_s3_or_kms(merged),
        )

    # ----- secret resource policies ---------------------------------------
    print()
    print("== Secret resource policies ==")
    for tenant, secret_name, secret_arn in [
        ("A", "harbor/cross-account/external-id-tenant-a", secret_a_arn),
        ("B", "harbor/cross-account/external-id-tenant-b", secret_b_arn),
    ]:
        rp = sm.get_resource_policy(SecretId=secret_name).get("ResourcePolicy")
        check(f"secret {tenant} has a resource policy", bool(rp))
        if not rp:
            continue
        rp_doc = json.loads(rp)
        admits_orch = False
        for st in stmts(rp_doc):
            principals = st.get("Principal", {})
            aws_p = as_list(
                principals.get("AWS") if isinstance(principals, dict) else None
            )
            actions = as_list(st.get("Action"))
            if (
                st.get("Effect") == "Allow"
                and orch_arn in aws_p
                and "secretsmanager:GetSecretValue" in actions
            ):
                admits_orch = True
        check(
            f"secret {tenant}: resource policy names orchestrator role as Principal.AWS for GetSecretValue",
            admits_orch,
        )

    # ----- KMS use ---------------------------------------------------------
    print()
    print("== KMS scope ==")
    kms = cli("kms")
    key_meta = kms.describe_key(KeyId="alias/cross-account-cmk")["KeyMetadata"]
    # Make sure both secrets actually use this key.
    desc_a = sm.describe_secret(SecretId=secret_a_arn)
    desc_b = sm.describe_secret(SecretId=secret_b_arn)
    check(
        "secret A is encrypted with the customer-managed CMK",
        desc_a.get("KmsKeyId") in (key_meta["Arn"], "alias/cross-account-cmk")
        or key_meta["KeyId"] in (desc_a.get("KmsKeyId") or ""),
        detail=str(desc_a.get("KmsKeyId")),
    )
    check(
        "secret B is encrypted with the customer-managed CMK",
        desc_b.get("KmsKeyId") in (key_meta["Arn"], "alias/cross-account-cmk")
        or key_meta["KeyId"] in (desc_b.get("KmsKeyId") or ""),
        detail=str(desc_b.get("KmsKeyId")),
    )

    # Final tally.
    print()
    print(f"== Summary ==  passed={len(PASSES)}  failed={len(FAILS)}")
    if FAILS:
        for label, detail in FAILS:
            print(f"  FAIL :: {label} :: {detail}")
        sys.exit(1)
    print("VERIFY OK")


if __name__ == "__main__":
    main()

- [Project context: cross-account trust hardening](project_cross_account.md) , SOC2 carryover sprint; OrchestratorRole + per-tenant member roles with ExternalId+SourceArn
- [LocalStack IAM enforcement quirk](reference_localstack_iam.md) , IAM is mocked at runtime; verify document SHAPE, not denial behavior

---
name: Cross-account trust hardening (SOC2 carryover)
description: Why the OrchestratorRole/MemberRole structure exists and what conditions must be on every member trust
type: project
---

External auditor flagged the cross-account assume-role chain in 2026-04 as too weak ("trust the orchestrator role arn" and nothing else). SOC2 carryover at risk if not fixed by next sprint.

The fix: one OrchestratorRole fans out to per-tenant MemberRole-Tenant-X roles. Each member role's trust must defend BOTH attacks at once:

1. Third-party impersonation → `Condition.StringEquals['sts:ExternalId']` matching a per-tenant shared secret stored in Secrets Manager (KMS-encrypted, distinct value per tenant).
2. AWS-service confused-deputy → `Condition.ArnLike['aws:SourceArn']` equal to the orchestrator role arn.

Both conditions, on every member trust. Principal must be the orchestrator role arn (never `*`); action must be exactly `sts:AssumeRole`.

**Why:** auditor explicitly required both controls; reusing one ExternalId across tenants defeats the point.

**How to apply:** any future tenant onboarding must mint a fresh ExternalId secret, attach a resource policy on the secret naming the orchestrator role as Principal.AWS, and use the same hardened trust template. Identity policies stay scoped , no `Resource: "*"` on s3/kms, no wildcard `sts:AssumeRole`. Build/verify scripts live at `/app/setup/{build,happy_path,verify}.py`.

---
name: LocalStack IAM is not enforced at runtime
description: Treat LocalStack as a JSON validator with an S3 backend; verify policy DOCUMENT shape, not runtime denial
type: reference
---

LocalStack at `http://localstack:4566` (default in this project) does not enforce IAM at runtime , AssumeRole/PutObject calls succeed regardless of policy denials that would fire in real AWS. Auditor-style verification therefore reads the policy DOCUMENTS directly (trust docs, inline role policies, secret resource policies) and asserts shape: principal, action, condition keys, resource enumeration, no wildcards.

Use the runtime path only to confirm the happy-path call sequence resolves end-to-end (assume → get-secret → assume-with-external-id → s3 put). For "negative" assertions (impersonator denied, missing ExternalId denied), prove them by reading the trust doc, not by attempting the call and watching it fail.

Project creds default: `AWS_ACCESS_KEY_ID=test`, `AWS_SECRET_ACCESS_KEY=test`, region `us-east-1`, endpoint `http://localstack:4566`.