iam-cross-account-externalid-sourcearn

#!/usr/bin/env python3
"""
Cross-account trust setup for SOC2: orchestrator -> per-tenant member roles.
Defends each member-role trust against:
  (1) third-party impersonation -> sts:ExternalId per-tenant secret
  (2) confused-deputy            -> aws:SourceArn = orchestrator role arn
"""
import json
import secrets
import string
import sys

import boto3

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

ORCH_ROLE = "OrchestratorRole"
MEMBER_A = "MemberRole-Tenant-A"
MEMBER_B = "MemberRole-Tenant-B"
AGENT_USER = "agent-runner"
ATTACKER_USER = "attacker-runner"
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"

ORCH_PATH = "/orchestrator/"
MEMBER_PATH = "/member/"


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


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


def ensure_user(iam, name):
    try:
        iam.create_user(UserName=name)
    except iam.exceptions.EntityAlreadyExistsException:
        pass
    try:
        keys = iam.list_access_keys(UserName=name)["AccessKeyMetadata"]
        if keys:
            return iam.list_access_keys(UserName=name)
    except Exception:
        pass
    return iam.create_access_key(UserName=name)


def delete_role_if_exists(iam, name):
    try:
        for p in iam.list_role_policies(RoleName=name)["PolicyNames"]:
            iam.delete_role_policy(RoleName=name, PolicyName=p)
        for ap in iam.list_attached_role_policies(RoleName=name)["AttachedPolicies"]:
            iam.detach_role_policy(RoleName=name, PolicyArn=ap["PolicyArn"])
        iam.delete_role(RoleName=name)
    except iam.exceptions.NoSuchEntityException:
        pass


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

    account_id = sts.get_caller_identity()["Account"]
    print(f"[+] account: {account_id}")

    # --- IAM users (orchestrator runtime identity + impersonator stand-in)
    print("[+] creating iam users")
    ensure_user(iam, AGENT_USER)
    ensure_user(iam, ATTACKER_USER)
    agent_arn = f"arn:aws:iam::{account_id}:user/{AGENT_USER}"

    # --- Orchestrator role: trusted by the agent-runner user
    print("[+] creating orchestrator role")
    delete_role_if_exists(iam, ORCH_ROLE)
    orch_trust = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Effect": "Allow",
                "Principal": {"AWS": agent_arn},
                "Action": "sts:AssumeRole",
            }
        ],
    }
    orch = iam.create_role(
        RoleName=ORCH_ROLE,
        Path=ORCH_PATH,
        AssumeRolePolicyDocument=json.dumps(orch_trust),
    )["Role"]
    orch_arn = orch["Arn"]
    print(f"    orch arn: {orch_arn}")

    # --- KMS customer-managed key + alias
    print("[+] creating kms cmk + alias")
    kms_key_policy = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "RootAccountAdmin",
                "Effect": "Allow",
                "Principal": {"AWS": f"arn:aws:iam::{account_id}:root"},
                "Action": "kms:*",
                "Resource": f"arn:aws:kms:{REGION}:{account_id}:key/*",
            },
            {
                "Sid": "AllowOrchestratorDecryptForSecrets",
                "Effect": "Allow",
                "Principal": {"AWS": orch_arn},
                "Action": ["kms:Decrypt", "kms:DescribeKey"],
                "Resource": f"arn:aws:kms:{REGION}:{account_id}:key/*",
                "Condition": {
                    "StringEquals": {
                        "kms:ViaService": f"secretsmanager.{REGION}.amazonaws.com"
                    }
                },
            },
        ],
    }
    cmk = kms.create_key(
        Description="harbor cross-account external-id encryption key",
        KeyUsage="ENCRYPT_DECRYPT",
        Policy=json.dumps(kms_key_policy),
    )["KeyMetadata"]
    cmk_id = cmk["KeyId"]
    cmk_arn = cmk["Arn"]
    try:
        kms.delete_alias(AliasName=KMS_ALIAS)
    except Exception:
        pass
    kms.create_alias(AliasName=KMS_ALIAS, TargetKeyId=cmk_id)
    print(f"    kms key arn: {cmk_arn}")

    # --- Per-tenant ExternalIds (random, distinct, >=32 chars)
    ext_a = gen_external_id()
    ext_b = gen_external_id()
    assert ext_a != ext_b
    assert len(ext_a) >= 32 and len(ext_b) >= 32
    print(f"[+] generated distinct external-ids ({len(ext_a)} chars each)")

    # --- Secrets Manager secrets (encrypted with the CMK)
    def put_secret(name, value):
        try:
            sm.delete_secret(SecretId=name, ForceDeleteWithoutRecovery=True)
        except sm.exceptions.ResourceNotFoundException:
            pass
        return sm.create_secret(
            Name=name,
            KmsKeyId=KMS_ALIAS,
            SecretString=value,
        )

    print("[+] writing external-id secrets (kms encrypted)")
    sec_a = put_secret(SECRET_A_NAME, ext_a)
    sec_b = put_secret(SECRET_B_NAME, ext_b)
    sec_a_arn = sec_a["ARN"]
    sec_b_arn = sec_b["ARN"]
    print(f"    tenant-a secret arn: {sec_a_arn}")
    print(f"    tenant-b secret arn: {sec_b_arn}")

    # --- Resource policy on each secret: admit orchestrator role
    def secret_resource_policy(secret_arn):
        return {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Sid": "AllowOrchestratorReadOnly",
                    "Effect": "Allow",
                    "Principal": {"AWS": orch_arn},
                    "Action": "secretsmanager:GetSecretValue",
                    "Resource": secret_arn,
                }
            ],
        }

    sm.put_resource_policy(
        SecretId=SECRET_A_NAME,
        ResourcePolicy=json.dumps(secret_resource_policy(sec_a_arn)),
    )
    sm.put_resource_policy(
        SecretId=SECRET_B_NAME,
        ResourcePolicy=json.dumps(secret_resource_policy(sec_b_arn)),
    )

    # --- Member roles: trust = orchestrator + ExternalId + SourceArn
    def member_trust(external_id_value):
        return {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Principal": {"AWS": orch_arn},
                    "Action": "sts:AssumeRole",
                    "Condition": {
                        "StringEquals": {"sts:ExternalId": external_id_value},
                        "ArnEquals": {"aws:SourceArn": orch_arn},
                    },
                }
            ],
        }

    print("[+] creating tenant member roles with hardened trust")
    delete_role_if_exists(iam, MEMBER_A)
    delete_role_if_exists(iam, MEMBER_B)
    mem_a = iam.create_role(
        RoleName=MEMBER_A,
        Path=MEMBER_PATH,
        AssumeRolePolicyDocument=json.dumps(member_trust(ext_a)),
    )["Role"]
    mem_b = iam.create_role(
        RoleName=MEMBER_B,
        Path=MEMBER_PATH,
        AssumeRolePolicyDocument=json.dumps(member_trust(ext_b)),
    )["Role"]
    mem_a_arn = mem_a["Arn"]
    mem_b_arn = mem_b["Arn"]
    print(f"    {MEMBER_A}: {mem_a_arn}")
    print(f"    {MEMBER_B}: {mem_b_arn}")

    # --- S3 buckets
    print("[+] creating tenant buckets")
    for b in (BUCKET_A, BUCKET_B):
        try:
            s3.create_bucket(Bucket=b)
        except s3.exceptions.BucketAlreadyOwnedByYou:
            pass

    # --- Orchestrator identity policy: explicit assume + scoped secret read
    bucket_a_arn = f"arn:aws:s3:::{BUCKET_A}"
    bucket_b_arn = f"arn:aws:s3:::{BUCKET_B}"
    orch_identity = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "AssumeMemberRolesOnly",
                "Effect": "Allow",
                "Action": "sts:AssumeRole",
                "Resource": [mem_a_arn, mem_b_arn],
            },
            {
                "Sid": "ReadExternalIdSecretsOnly",
                "Effect": "Allow",
                "Action": "secretsmanager:GetSecretValue",
                "Resource": [sec_a_arn, sec_b_arn],
            },
            {
                "Sid": "DecryptExternalIdSecretsViaSecretsManager",
                "Effect": "Allow",
                "Action": ["kms:Decrypt", "kms:DescribeKey"],
                "Resource": cmk_arn,
                "Condition": {
                    "StringEquals": {
                        "kms:ViaService": f"secretsmanager.{REGION}.amazonaws.com"
                    }
                },
            },
        ],
    }
    iam.put_role_policy(
        RoleName=ORCH_ROLE,
        PolicyName="orchestrator-identity",
        PolicyDocument=json.dumps(orch_identity),
    )

    # Also let the agent-runner user assume the orchestrator role
    iam.put_user_policy(
        UserName=AGENT_USER,
        PolicyName="assume-orchestrator-only",
        PolicyDocument=json.dumps(
            {
                "Version": "2012-10-17",
                "Statement": [
                    {
                        "Effect": "Allow",
                        "Action": "sts:AssumeRole",
                        "Resource": orch_arn,
                    }
                ],
            }
        ),
    )

    # --- Member identity policies: scoped to their own bucket only
    def member_identity(bucket_arn):
        return {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Sid": "TenantBucketObjectAccess",
                    "Effect": "Allow",
                    "Action": ["s3:PutObject", "s3:GetObject", "s3:DeleteObject"],
                    "Resource": f"{bucket_arn}/*",
                },
                {
                    "Sid": "TenantBucketList",
                    "Effect": "Allow",
                    "Action": ["s3:ListBucket"],
                    "Resource": bucket_arn,
                },
            ],
        }

    iam.put_role_policy(
        RoleName=MEMBER_A,
        PolicyName="tenant-a-bucket-only",
        PolicyDocument=json.dumps(member_identity(bucket_a_arn)),
    )
    iam.put_role_policy(
        RoleName=MEMBER_B,
        PolicyName="tenant-b-bucket-only",
        PolicyDocument=json.dumps(member_identity(bucket_b_arn)),
    )

    # --- SSM pointers for the verifier
    print("[+] writing ssm pointers")
    pointers = {
        "/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,
    }
    for k, v in pointers.items():
        ssm.put_parameter(Name=k, Value=v, Type="String", Overwrite=True)
        print(f"    {k} = {v}")

    print("[+] setup complete")


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

#!/usr/bin/env python3
"""
Happy path: orchestrator -> tenant-A.
1. Read tenant-A ExternalId from secrets manager.
2. AssumeRole into MemberRole-Tenant-A with --external-id <secret>.
3. PutObject into tenant-a-bucket.
"""
import boto3

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


def c(svc, **kw):
    return boto3.client(svc, endpoint_url=ENDPOINT, region_name=REGION, **kw)


def main():
    ssm = c("ssm")

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

    orch_arn = p("/harbor/orchestrator/role-arn")
    sec_a_arn = p("/harbor/external-id-secret-arn-tenant-a")
    mem_a_arn = p("/harbor/member-role-arn-tenant-a")

    # Step 1: act as the orchestrator (assume the orchestrator role from agent-runner).
    # In LocalStack the default test creds already act as account root, so we directly
    # assume the orchestrator role to get an orchestrator session.
    sts = c("sts")
    orch_session = sts.assume_role(
        RoleArn=orch_arn, RoleSessionName="orchestrator-session"
    )["Credentials"]
    print(f"[+] orchestrator session: {orch_session['AccessKeyId'][:6]}…")

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

    # Step 2: orchestrator reads tenant-A ExternalId from Secrets Manager.
    sm = c("secretsmanager", **orch_creds)
    ext_a = sm.get_secret_value(SecretId=sec_a_arn)["SecretString"]
    print(f"[+] tenant-a external-id retrieved (len={len(ext_a)})")

    # Step 3: orchestrator assumes tenant-A member role, presenting the ExternalId.
    sts_orch = c("sts", **orch_creds)
    mem_a_session = sts_orch.assume_role(
        RoleArn=mem_a_arn,
        RoleSessionName="orch-to-tenant-a",
        ExternalId=ext_a,
    )["Credentials"]
    print(f"[+] tenant-a member session: {mem_a_session['AccessKeyId'][:6]}…")

    mem_a_creds = dict(
        aws_access_key_id=mem_a_session["AccessKeyId"],
        aws_secret_access_key=mem_a_session["SecretAccessKey"],
        aws_session_token=mem_a_session["SessionToken"],
    )

    # Step 4: write to tenant-a-bucket as the tenant-A role.
    s3 = c("s3", **mem_a_creds)
    s3.put_object(
        Bucket="tenant-a-bucket",
        Key="hello.txt",
        Body=b"cross-account write succeeded\n",
    )
    body = s3.get_object(Bucket="tenant-a-bucket", Key="hello.txt")["Body"].read()
    print(f"[+] put+get round-trip: {body!r}")

    print("[+] happy path OK")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Auditor-style verifier. Reads policy DOCUMENTS and asserts shape.
Treats LocalStack as a JSON validator: runtime denial isn't checked,
the policy text itself must be correct.
"""
import json
import sys
import urllib.parse

import boto3

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

ORCH_ROLE = "OrchestratorRole"
MEMBER_A = "MemberRole-Tenant-A"
MEMBER_B = "MemberRole-Tenant-B"
SECRET_A = "harbor/cross-account/external-id-tenant-a"
SECRET_B = "harbor/cross-account/external-id-tenant-b"
BUCKET_A_ARN = "arn:aws:s3:::tenant-a-bucket"
BUCKET_B_ARN = "arn:aws:s3:::tenant-b-bucket"
ORCH_PATH = "/orchestrator/"
MEMBER_PATH = "/member/"


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


def asdoc(maybe_encoded):
    """IAM returns policy docs URL-encoded sometimes; normalize to dict."""
    if isinstance(maybe_encoded, dict):
        return maybe_encoded
    if isinstance(maybe_encoded, str):
        try:
            return json.loads(maybe_encoded)
        except json.JSONDecodeError:
            return json.loads(urllib.parse.unquote(maybe_encoded))
    raise TypeError(type(maybe_encoded))


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


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


class Checks:
    def __init__(self):
        self.passed = []
        self.failed = []

    def check(self, label, ok, detail=""):
        (self.passed if ok else self.failed).append((label, detail))
        marker = "PASS" if ok else "FAIL"
        line = f"[{marker}] {label}"
        if detail:
            line += f"  -- {detail}"
        print(line)


def find_role_by_path(iam, path, expected_name):
    out = iam.list_roles(PathPrefix=path)["Roles"]
    for r in out:
        if r["RoleName"] == expected_name:
            return r
    return None


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

    chk = Checks()

    # --- enumerate by path (the verifier uses these paths) ---
    orch_role = find_role_by_path(iam, ORCH_PATH, ORCH_ROLE)
    mem_a = find_role_by_path(iam, MEMBER_PATH, MEMBER_A)
    mem_b = find_role_by_path(iam, MEMBER_PATH, MEMBER_B)
    chk.check("orchestrator role exists under /orchestrator/", orch_role is not None)
    chk.check("tenant-A member role exists under /member/", mem_a is not None)
    chk.check("tenant-B member role exists under /member/", mem_b is not None)
    if not (orch_role and mem_a and mem_b):
        return 1
    orch_arn = orch_role["Arn"]

    # --- ssm pointers populated ---
    pointers = {
        "/harbor/orchestrator/role-arn": orch_arn,
        "/harbor/external-id-secret-arn-tenant-a": None,
        "/harbor/external-id-secret-arn-tenant-b": None,
        "/harbor/member-role-arn-tenant-a": mem_a["Arn"],
        "/harbor/member-role-arn-tenant-b": mem_b["Arn"],
    }
    resolved = {}
    for k, expected in pointers.items():
        try:
            v = ssm.get_parameter(Name=k)["Parameter"]["Value"]
            resolved[k] = v
            ok = expected is None or v == expected
            chk.check(f"ssm pointer {k} present and correct", ok, f"value={v}")
        except Exception as e:
            chk.check(f"ssm pointer {k} present", False, str(e))

    sec_a_arn = resolved.get("/harbor/external-id-secret-arn-tenant-a")
    sec_b_arn = resolved.get("/harbor/external-id-secret-arn-tenant-b")

    # --- ExternalId secret values: distinct, >=32 chars, kms-encrypted ---
    val_a = sm.get_secret_value(SecretId=SECRET_A)["SecretString"]
    val_b = sm.get_secret_value(SecretId=SECRET_B)["SecretString"]
    chk.check("tenant-A external-id is >=32 chars", len(val_a) >= 32, f"len={len(val_a)}")
    chk.check("tenant-B external-id is >=32 chars", len(val_b) >= 32, f"len={len(val_b)}")
    chk.check("tenant external-ids differ", val_a != val_b)

    desc_a = sm.describe_secret(SecretId=SECRET_A)
    desc_b = sm.describe_secret(SecretId=SECRET_B)
    chk.check(
        "tenant-A secret encrypted with customer-managed kms",
        bool(desc_a.get("KmsKeyId")) and desc_a["KmsKeyId"] != "",
        f"kms={desc_a.get('KmsKeyId')}",
    )
    chk.check(
        "tenant-B secret encrypted with customer-managed kms",
        bool(desc_b.get("KmsKeyId")) and desc_b["KmsKeyId"] != "",
        f"kms={desc_b.get('KmsKeyId')}",
    )

    # --- Member trust docs: principal, action, ExternalId, SourceArn ---
    def assert_member_trust(role, expected_external_id, label):
        trust = asdoc(role["AssumeRolePolicyDocument"])
        ss = stmts(trust)
        chk.check(f"{label}: trust has exactly one statement", len(ss) == 1)
        s = ss[0]
        # Effect
        chk.check(f"{label}: Effect=Allow", s.get("Effect") == "Allow")
        # Principal = orch role arn (not "*")
        principal = s.get("Principal", {})
        principal_aws = aslist(principal.get("AWS"))
        chk.check(
            f"{label}: Principal.AWS is the orchestrator role arn (not '*')",
            principal_aws == [orch_arn],
            f"got={principal_aws}",
        )
        chk.check(
            f"{label}: Principal does not contain '*'",
            "*" not in principal_aws and principal != "*" and principal.get("AWS") != "*",
        )
        # Action = sts:AssumeRole exactly, no wildcards
        actions = aslist(s.get("Action"))
        chk.check(
            f"{label}: Action is exactly ['sts:AssumeRole']",
            actions == ["sts:AssumeRole"],
            f"got={actions}",
        )
        # Conditions
        cond = s.get("Condition", {})
        # ExternalId
        ext_block = cond.get("StringEquals", {})
        ext_val = ext_block.get("sts:ExternalId")
        chk.check(
            f"{label}: Condition.StringEquals['sts:ExternalId'] present",
            ext_val is not None,
        )
        chk.check(
            f"{label}: ExternalId in trust matches the per-tenant secret value",
            ext_val == expected_external_id,
        )
        # SourceArn: ArnEquals or ArnLike
        src = None
        for op in ("ArnEquals", "ArnLike"):
            src = cond.get(op, {}).get("aws:SourceArn")
            if src is not None:
                break
        chk.check(
            f"{label}: Condition.{{ArnEquals|ArnLike}}['aws:SourceArn'] present",
            src is not None,
        )
        chk.check(
            f"{label}: aws:SourceArn equals the orchestrator role arn",
            src == orch_arn,
            f"got={src}",
        )

    assert_member_trust(mem_a, val_a, "tenant-A trust")
    assert_member_trust(mem_b, val_b, "tenant-B trust")

    # --- Orchestrator identity policy: explicit member arns + scoped secrets ---
    orch_pols = iam.list_role_policies(RoleName=ORCH_ROLE)["PolicyNames"]
    chk.check("orchestrator has at least one inline policy", len(orch_pols) >= 1)
    saw_assume = False
    saw_secret = False
    for pn in orch_pols:
        doc = asdoc(
            iam.get_role_policy(RoleName=ORCH_ROLE, PolicyName=pn)["PolicyDocument"]
        )
        for s in stmts(doc):
            actions = aslist(s.get("Action"))
            resources = aslist(s.get("Resource"))
            # Reject any blanket Action:"*" or Resource:"*"
            chk.check(
                f"orchestrator stmt in '{pn}': no Action:'*'",
                "*" not in actions,
                f"actions={actions}",
            )
            chk.check(
                f"orchestrator stmt in '{pn}': no Resource:'*'",
                "*" not in resources,
                f"resources={resources}",
            )
            if "sts:AssumeRole" in actions:
                saw_assume = True
                chk.check(
                    "orchestrator AssumeRole resource enumerates both member arns",
                    set(resources) == {mem_a["Arn"], mem_b["Arn"]},
                    f"got={resources}",
                )
            if "secretsmanager:GetSecretValue" in actions:
                saw_secret = True
                chk.check(
                    "orchestrator GetSecretValue resource enumerates both secret arns",
                    set(resources) == {sec_a_arn, sec_b_arn},
                    f"got={resources}",
                )
    chk.check("orchestrator has explicit AssumeRole statement", saw_assume)
    chk.check("orchestrator has explicit GetSecretValue statement", saw_secret)

    # --- Member identity policies: each scoped to its own bucket only ---
    def assert_member_identity(role_name, expected_bucket_arn, other_bucket_arn, label):
        pols = iam.list_role_policies(RoleName=role_name)["PolicyNames"]
        chk.check(f"{label}: has at least one inline policy", len(pols) >= 1)
        for pn in pols:
            doc = asdoc(
                iam.get_role_policy(RoleName=role_name, PolicyName=pn)[
                    "PolicyDocument"
                ]
            )
            for s in stmts(doc):
                actions = aslist(s.get("Action"))
                resources = aslist(s.get("Resource"))
                chk.check(
                    f"{label} stmt in '{pn}': no Action:'*'",
                    "*" not in actions,
                    f"actions={actions}",
                )
                chk.check(
                    f"{label} stmt in '{pn}': no Resource:'*'",
                    "*" not in resources,
                    f"resources={resources}",
                )
                # Resource must reference its own bucket arn (with optional /*),
                # never the other tenant's bucket.
                for r in resources:
                    chk.check(
                        f"{label} stmt in '{pn}': resource confined to own bucket",
                        r.startswith(expected_bucket_arn),
                        f"resource={r}",
                    )
                    chk.check(
                        f"{label} stmt in '{pn}': resource does NOT reference other tenant's bucket",
                        not r.startswith(other_bucket_arn),
                        f"resource={r}",
                    )

    assert_member_identity(MEMBER_A, BUCKET_A_ARN, BUCKET_B_ARN, "tenant-A identity")
    assert_member_identity(MEMBER_B, BUCKET_B_ARN, BUCKET_A_ARN, "tenant-B identity")

    # --- Secrets Manager resource policies admit the orchestrator ---
    def assert_secret_resource_policy(secret_name, label):
        try:
            rp_raw = sm.get_resource_policy(SecretId=secret_name).get("ResourcePolicy")
        except Exception as e:
            chk.check(f"{label}: resource policy fetch", False, str(e))
            return
        chk.check(f"{label}: secret has a resource policy", bool(rp_raw))
        if not rp_raw:
            return
        rp = asdoc(rp_raw)
        admit = False
        for s in stmts(rp):
            if s.get("Effect") != "Allow":
                continue
            principals = aslist(s.get("Principal", {}).get("AWS"))
            actions = aslist(s.get("Action"))
            if orch_arn in principals and "secretsmanager:GetSecretValue" in actions:
                admit = True
                chk.check(
                    f"{label}: resource policy doesn't use Principal:'*'",
                    "*" not in principals,
                )
                chk.check(
                    f"{label}: resource policy doesn't use Action:'*'",
                    "*" not in actions,
                )
        chk.check(
            f"{label}: orchestrator admitted by secret resource policy", admit
        )

    assert_secret_resource_policy(SECRET_A, "tenant-A secret")
    assert_secret_resource_policy(SECRET_B, "tenant-B secret")

    # --- KMS: no wildcard usage anywhere we control ---
    # Walk every policy doc we've written and ensure no kms statement uses Action:"*" or Resource:"*"
    def scan_for_kms_wildcards(doc, where):
        for s in stmts(doc):
            actions = aslist(s.get("Action"))
            resources = aslist(s.get("Resource"))
            kms_actions = [a for a in actions if a.startswith("kms:") or a == "*"]
            if kms_actions:
                chk.check(
                    f"{where}: kms statement has no Action:'*'",
                    "*" not in actions,
                    f"actions={actions}",
                )
                chk.check(
                    f"{where}: kms statement has no Resource:'*'",
                    "*" not in resources,
                    f"resources={resources}",
                )
            s3_actions = [a for a in actions if a.startswith("s3:") or a == "*"]
            if s3_actions:
                chk.check(
                    f"{where}: s3 statement has no Action:'*'",
                    "*" not in actions,
                )
                chk.check(
                    f"{where}: s3 statement has no Resource:'*'",
                    "*" not in resources,
                )

    for rn in (ORCH_ROLE, MEMBER_A, MEMBER_B):
        for pn in iam.list_role_policies(RoleName=rn)["PolicyNames"]:
            d = asdoc(iam.get_role_policy(RoleName=rn, PolicyName=pn)["PolicyDocument"])
            scan_for_kms_wildcards(d, f"{rn}/{pn}")

    print()
    print(f"PASSED: {len(chk.passed)}    FAILED: {len(chk.failed)}")
    if chk.failed:
        print("\nFailures:")
        for label, detail in chk.failed:
            print(f"  - {label}: {detail}")
        return 1
    return 0


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

---
name: SOC2 cross-account trust hardening
description: Active audit finding requiring per-tenant ExternalId + aws:SourceArn on every member-role trust before next sprint
type: project
---

External auditor flagged the cross-account assume-role chain (orchestrator -> per-tenant member roles) as relying only on Principal=orchestrator-arn, which fails to defend against (1) third-party impersonation if a member-role ARN leaks, and (2) AWS-service confused-deputy where another principal in the account is tricked into calling AssumeRole on the orchestrator's behalf. Fix landed in /app/setup.py: every member-role trust now combines `StringEquals sts:ExternalId=<per-tenant secret>` and `ArnEquals aws:SourceArn=<orchestrator role arn>`. ExternalIds are 48-char random, distinct per tenant, stored in Secrets Manager under `harbor/cross-account/external-id-tenant-{a,b}` encrypted by CMK `alias/cross-account-cmk`. Verifier in /app/verify.py asserts policy-document shape (LocalStack doesn't enforce IAM at runtime). Project namespace is "harbor" (visible in SSM path `/harbor/...`).

**Why:** Carryover finding; losing it blocks SOC2 carryover. Auditor will check the JSON shape of trust + identity policies on both sides (identity-side scope alone isn't enough , secret resource policy must also admit the orchestrator).

**How to apply:** Any new tenant onboarding or new cross-account role in this project must replicate the dual-condition trust pattern (ExternalId + SourceArn, both, never one). Don't accept `Principal:"*"`, wildcard `sts:AssumeRole` resources, or shared ExternalIds across tenants , the auditor explicitly checks for these. KMS and S3 statements must avoid `Action:"*"` and `Resource:"*"`. When adding a member role, also add: identity-policy entry on orchestrator enumerating the new ARN, secret resource policy admitting the orchestrator, and SSM pointers under `/harbor/...`.

# Memory index

- [SOC2 cross-account trust hardening](project_soc2_cross_account.md) , active audit finding; member-role trusts must combine sts:ExternalId + aws:SourceArn, ExternalIds per-tenant + KMS-encrypted