iam-cross-account-externalid-sourcearn

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

EP="--endpoint-url=http://localstack:4566"
ACCOUNT_ID=000000000000

ORCH_USER=agent-runner
ATK_USER=attacker-runner
ORCH_ROLE=OrchestratorRole
MEMBER_A=MemberRole-Tenant-A
MEMBER_B=MemberRole-Tenant-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

EXT_A=$(python3 -c "import secrets;print(secrets.token_urlsafe(48))")
EXT_B=$(python3 -c "import secrets;print(secrets.token_urlsafe(48))")
if [ "${#EXT_A}" -lt 32 ] || [ "${#EXT_B}" -lt 32 ] || [ "$EXT_A" = "$EXT_B" ]; then
  echo "external-id generation failed (length/uniqueness)" >&2
  exit 1
fi

echo "==> KMS CMK"
KEY_ID=$(aws $EP kms create-key --description "cross-account cmk for external-id secrets" \
  --query 'KeyMetadata.KeyId' --output text)
aws $EP kms create-alias --alias-name "$KMS_ALIAS" --target-key-id "$KEY_ID" >/dev/null
KEY_ARN=$(aws $EP kms describe-key --key-id "$KEY_ID" --query 'KeyMetadata.Arn' --output text)

echo "==> IAM users"
aws $EP iam create-user --user-name "$ORCH_USER" >/dev/null
aws $EP iam create-user --user-name "$ATK_USER"  >/dev/null
ORCH_USER_ARN=$(aws $EP iam get-user --user-name "$ORCH_USER" --query 'User.Arn' --output text)

echo "==> Orchestrator role (trusted only by $ORCH_USER)"
ORCH_TRUST=$(cat <<JSON
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AgentRunnerOnly",
      "Effect": "Allow",
      "Principal": {"AWS": "$ORCH_USER_ARN"},
      "Action": "sts:AssumeRole"
    }
  ]
}
JSON
)
aws $EP iam create-role \
  --role-name "$ORCH_ROLE" \
  --path /orchestrator/ \
  --assume-role-policy-document "$ORCH_TRUST" >/dev/null
ORCH_ROLE_ARN=$(aws $EP iam get-role --role-name "$ORCH_ROLE" --query 'Role.Arn' --output text)

echo "==> Tenant buckets"
aws $EP s3api create-bucket --bucket "$BUCKET_A" >/dev/null
aws $EP s3api create-bucket --bucket "$BUCKET_B" >/dev/null

create_member_role () {
  local NAME="$1" EXT="$2" BUCKET="$3"
  local TRUST POL
  TRUST=$(cat <<JSON
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "OrchestratorOnlyWithExternalIdAndSourceArn",
      "Effect": "Allow",
      "Principal": {"AWS": "$ORCH_ROLE_ARN"},
      "Action": "sts:AssumeRole",
      "Condition": {
        "StringEquals": {"sts:ExternalId": "$EXT"},
        "ArnEquals":    {"aws:SourceArn": "$ORCH_ROLE_ARN"}
      }
    }
  ]
}
JSON
)
  aws $EP iam create-role \
    --role-name "$NAME" \
    --path /member/ \
    --assume-role-policy-document "$TRUST" >/dev/null

  POL=$(cat <<JSON
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "TenantBucketObjects",
      "Effect": "Allow",
      "Action": ["s3:PutObject","s3:GetObject","s3:DeleteObject"],
      "Resource": "arn:aws:s3:::${BUCKET}/*"
    },
    {
      "Sid": "TenantBucketList",
      "Effect": "Allow",
      "Action": ["s3:ListBucket"],
      "Resource": "arn:aws:s3:::${BUCKET}"
    }
  ]
}
JSON
)
  aws $EP iam put-role-policy \
    --role-name "$NAME" \
    --policy-name "TenantBucketAccess" \
    --policy-document "$POL"
}

echo "==> Member roles"
create_member_role "$MEMBER_A" "$EXT_A" "$BUCKET_A"
create_member_role "$MEMBER_B" "$EXT_B" "$BUCKET_B"
MEMBER_A_ARN=$(aws $EP iam get-role --role-name "$MEMBER_A" --query 'Role.Arn' --output text)
MEMBER_B_ARN=$(aws $EP iam get-role --role-name "$MEMBER_B" --query 'Role.Arn' --output text)

echo "==> Secrets (KMS-encrypted, value = ExternalId)"
SECRET_A_ARN=$(aws $EP secretsmanager create-secret \
  --name "$SECRET_A_NAME" \
  --kms-key-id "$KEY_ARN" \
  --secret-string "$EXT_A" \
  --query 'ARN' --output text)
SECRET_B_ARN=$(aws $EP secretsmanager create-secret \
  --name "$SECRET_B_NAME" \
  --kms-key-id "$KEY_ARN" \
  --secret-string "$EXT_B" \
  --query 'ARN' --output text)

echo "==> Resource policy on each secret (admit orchestrator role explicitly)"
for ARN in "$SECRET_A_ARN" "$SECRET_B_ARN"; do
  RP=$(cat <<JSON
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "OrchestratorRoleMayRead",
      "Effect": "Allow",
      "Principal": {"AWS": "$ORCH_ROLE_ARN"},
      "Action": "secretsmanager:GetSecretValue",
      "Resource": "*"
    }
  ]
}
JSON
)
  aws $EP secretsmanager put-resource-policy \
    --secret-id "$ARN" \
    --resource-policy "$RP" >/dev/null
done

echo "==> Orchestrator identity policy (no wildcards for assume / secret / kms)"
ORCH_POL=$(cat <<JSON
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AssumeMemberRolesExplicit",
      "Effect": "Allow",
      "Action": "sts:AssumeRole",
      "Resource": ["$MEMBER_A_ARN","$MEMBER_B_ARN"]
    },
    {
      "Sid": "ReadExternalIdSecretsExplicit",
      "Effect": "Allow",
      "Action": "secretsmanager:GetSecretValue",
      "Resource": ["$SECRET_A_ARN","$SECRET_B_ARN"]
    },
    {
      "Sid": "DecryptExternalIdsWithCmk",
      "Effect": "Allow",
      "Action": "kms:Decrypt",
      "Resource": "$KEY_ARN"
    }
  ]
}
JSON
)
aws $EP iam put-role-policy \
  --role-name "$ORCH_ROLE" \
  --policy-name "OrchestratorPermissions" \
  --policy-document "$ORCH_POL"

echo "==> SSM pointers"
aws $EP ssm put-parameter --overwrite --type String \
  --name /harbor/orchestrator/role-arn --value "$ORCH_ROLE_ARN" >/dev/null
aws $EP ssm put-parameter --overwrite --type String \
  --name /harbor/external-id-secret-arn-tenant-a --value "$SECRET_A_ARN" >/dev/null
aws $EP ssm put-parameter --overwrite --type String \
  --name /harbor/external-id-secret-arn-tenant-b --value "$SECRET_B_ARN" >/dev/null
aws $EP ssm put-parameter --overwrite --type String \
  --name /harbor/member-role-arn-tenant-a --value "$MEMBER_A_ARN" >/dev/null
aws $EP ssm put-parameter --overwrite --type String \
  --name /harbor/member-role-arn-tenant-b --value "$MEMBER_B_ARN" >/dev/null

echo
echo "ORCH_ROLE_ARN=$ORCH_ROLE_ARN"
echo "MEMBER_A_ARN=$MEMBER_A_ARN"
echo "MEMBER_B_ARN=$MEMBER_B_ARN"
echo "SECRET_A_ARN=$SECRET_A_ARN"
echo "SECRET_B_ARN=$SECRET_B_ARN"
echo "KMS_KEY_ARN=$KEY_ARN"
echo "Done."

#!/usr/bin/env python3
"""Auditor-style shape checks against the cross-account trust setup.

Reads policy documents directly and asserts the structural properties
the SOC2 reviewer cares about: ExternalId + SourceArn on every member
role, principal pinned to the orchestrator role ARN, no wildcards on
the high-blast-radius statements, distinct ExternalIds per tenant,
secret resource policy admits the orchestrator role.
"""
from __future__ import annotations

import json
import sys
import boto3

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

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

failures: list[str] = []
passes:   list[str] = []

def check(label: str, ok: bool, detail: str = "") -> None:
    (passes if ok else failures).append(f"{label}{(' , ' + detail) if detail else ''}")

def ssm_get(name: str) -> str:
    return ssm.get_parameter(Name=name)["Parameter"]["Value"]

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

# --- pointers ---
ORCH_ROLE_ARN  = ssm_get("/harbor/orchestrator/role-arn")
MEMBER_A_ARN   = ssm_get("/harbor/member-role-arn-tenant-a")
MEMBER_B_ARN   = ssm_get("/harbor/member-role-arn-tenant-b")
SECRET_A_ARN   = ssm_get("/harbor/external-id-secret-arn-tenant-a")
SECRET_B_ARN   = ssm_get("/harbor/external-id-secret-arn-tenant-b")

# --- pull live secret values; trust docs reference these literally ---
EXT_A = sm.get_secret_value(SecretId=SECRET_A_ARN)["SecretString"]
EXT_B = sm.get_secret_value(SecretId=SECRET_B_ARN)["SecretString"]

check("ExternalId-A length >= 32", len(EXT_A) >= 32, f"len={len(EXT_A)}")
check("ExternalId-B length >= 32", len(EXT_B) >= 32, f"len={len(EXT_B)}")
check("ExternalIds are distinct per tenant", EXT_A != EXT_B)

# --- member trust shape ---
def check_member_trust(role_name: str, expected_ext: str, tenant: str) -> None:
    role = iam.get_role(RoleName=role_name)["Role"]
    doc  = role["AssumeRolePolicyDocument"]
    stmts = as_list(doc["Statement"])
    check(f"[{tenant}] trust has exactly one statement", len(stmts) == 1)
    s = stmts[0]

    check(f"[{tenant}] effect=Allow", s.get("Effect") == "Allow")

    actions = as_list(s.get("Action", []))
    check(f"[{tenant}] action is exactly sts:AssumeRole",
          actions == ["sts:AssumeRole"], f"actions={actions}")

    principal = s.get("Principal", {})
    aws_p = as_list(principal.get("AWS", []))
    check(f"[{tenant}] principal is the orchestrator role ARN (not '*')",
          aws_p == [ORCH_ROLE_ARN], f"principal.AWS={aws_p}")
    check(f"[{tenant}] principal does not contain '*'",
          "*" not in aws_p and principal.get("AWS") != "*")

    cond = s.get("Condition", {})
    se   = cond.get("StringEquals", {})
    check(f"[{tenant}] StringEquals.sts:ExternalId matches secret value",
          se.get("sts:ExternalId") == expected_ext)

    arn_block = cond.get("ArnEquals") or cond.get("ArnLike") or {}
    src = arn_block.get("aws:SourceArn")
    check(f"[{tenant}] aws:SourceArn condition equals orchestrator role ARN",
          src == ORCH_ROLE_ARN, f"src={src}")

check_member_trust("MemberRole-Tenant-A", EXT_A, "tenant-A")
check_member_trust("MemberRole-Tenant-B", EXT_B, "tenant-B")

# --- member identity policy: bucket scoping, no wildcards ---
def check_member_identity(role_name: str, own_bucket: str, other_bucket: str, tenant: str) -> None:
    polnames = iam.list_role_policies(RoleName=role_name)["PolicyNames"]
    check(f"[{tenant}] member role has at least one inline policy", len(polnames) >= 1)
    seen_resources: list[str] = []
    for pn in polnames:
        doc = iam.get_role_policy(RoleName=role_name, PolicyName=pn)["PolicyDocument"]
        for s in as_list(doc["Statement"]):
            actions = as_list(s.get("Action", []))
            resources = as_list(s.get("Resource", []))
            seen_resources.extend(resources)
            check(f"[{tenant}] no Action:'*' on {pn}", "*" not in actions)
            for a in actions:
                if a.startswith("s3:"):
                    check(f"[{tenant}] s3 action {a} not wildcard suffix",
                          not a.endswith(":*"), f"action={a}")
            for r in resources:
                check(f"[{tenant}] no Resource:'*' on {pn}", r != "*")
                check(f"[{tenant}] resource does not reference other tenant bucket",
                      other_bucket not in r, f"resource={r}")
    check(f"[{tenant}] inline policy references own bucket arn",
          any(own_bucket in r for r in seen_resources))

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

# --- orchestrator identity policy: explicit assume + explicit secret-get + scoped kms ---
orch_pols = iam.list_role_policies(RoleName="OrchestratorRole")["PolicyNames"]
check("orchestrator has inline policy", len(orch_pols) >= 1)

found_assume = found_secret = found_kms = False
for pn in orch_pols:
    doc = iam.get_role_policy(RoleName="OrchestratorRole", PolicyName=pn)["PolicyDocument"]
    for s in as_list(doc["Statement"]):
        actions   = as_list(s.get("Action", []))
        resources = as_list(s.get("Resource", []))
        if "sts:AssumeRole" in actions:
            check("orchestrator assume-role resource is exact 2-arn list",
                  set(resources) == {MEMBER_A_ARN, MEMBER_B_ARN},
                  f"resources={resources}")
            check("orchestrator assume-role has no '*' resource", "*" not in resources)
            found_assume = True
        if "secretsmanager:GetSecretValue" in actions:
            check("orchestrator secret-get resource is exact 2-arn list",
                  set(resources) == {SECRET_A_ARN, SECRET_B_ARN},
                  f"resources={resources}")
            check("orchestrator secret-get has no '*' resource", "*" not in resources)
            found_secret = True
        if any(a.startswith("kms:") for a in actions):
            check("orchestrator kms action is not wildcard",
                  "*" not in actions and not any(a == "kms:*" for a in actions),
                  f"actions={actions}")
            check("orchestrator kms resource is not '*'",
                  "*" not in resources, f"resources={resources}")
            found_kms = True

check("orchestrator policy includes sts:AssumeRole statement", found_assume)
check("orchestrator policy includes secretsmanager:GetSecretValue", found_secret)
check("orchestrator policy includes scoped kms:Decrypt", found_kms)

# --- secret resource policies admit the orchestrator role ---
def check_secret_resource_policy(secret_arn: str, tag: str) -> None:
    rp = sm.get_resource_policy(SecretId=secret_arn).get("ResourcePolicy")
    check(f"[{tag}] secret has a resource policy attached", bool(rp))
    if not rp:
        return
    doc = json.loads(rp)
    admits = False
    for s in as_list(doc["Statement"]):
        if s.get("Effect") != "Allow":
            continue
        principals = s.get("Principal", {}).get("AWS")
        principals = as_list(principals) if principals else []
        actions    = as_list(s.get("Action", []))
        if ORCH_ROLE_ARN in principals and "secretsmanager:GetSecretValue" in actions:
            admits = True
    check(f"[{tag}] resource policy admits orchestrator role for GetSecretValue", admits)

check_secret_resource_policy(SECRET_A_ARN, "tenant-A")
check_secret_resource_policy(SECRET_B_ARN, "tenant-B")

# --- path enumeration sanity ---
orch_listed = [r["RoleName"] for r in iam.list_roles(PathPrefix="/orchestrator/")["Roles"]]
member_listed = [r["RoleName"] for r in iam.list_roles(PathPrefix="/member/")["Roles"]]
check("/orchestrator/ path lists OrchestratorRole",
      orch_listed == ["OrchestratorRole"], f"got={orch_listed}")
check("/member/ path lists both member roles",
      sorted(member_listed) == ["MemberRole-Tenant-A", "MemberRole-Tenant-B"],
      f"got={member_listed}")

# --- report ---
print(f"\n{len(passes)} checks passed")
for p in passes:
    print(f"  OK  {p}")
if failures:
    print(f"\n{len(failures)} checks FAILED")
    for f in failures:
        print(f"  !!  {f}")
    sys.exit(1)
print("\nALL SHAPE CHECKS PASS")

#!/usr/bin/env python3
"""Happy path: agent-runner -> OrchestratorRole -> read external-id ->
AssumeRole MemberRole-Tenant-A with that external-id -> PutObject in
tenant-a-bucket. Each step uses the previous step's credentials.
"""
from __future__ import annotations

import sys
import boto3

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

ssm_root = boto3.client("ssm", endpoint_url=EP, region_name=REGION)
def ptr(name: str) -> str:
    return ssm_root.get_parameter(Name=name)["Parameter"]["Value"]

ORCH_ROLE_ARN = ptr("/harbor/orchestrator/role-arn")
MEMBER_A_ARN  = ptr("/harbor/member-role-arn-tenant-a")
SECRET_A_ARN  = ptr("/harbor/external-id-secret-arn-tenant-a")

print(f"orchestrator role  : {ORCH_ROLE_ARN}")
print(f"tenant-A member arn: {MEMBER_A_ARN}")
print(f"tenant-A secret arn: {SECRET_A_ARN}")

# --- Step 1: agent-runner identity -> OrchestratorRole ---
sts = boto3.client("sts", endpoint_url=EP, region_name=REGION)
orch = sts.assume_role(RoleArn=ORCH_ROLE_ARN, RoleSessionName="agent-runner-session")["Credentials"]
print("\nstep 1: assumed OrchestratorRole")

def session_from(creds):
    return boto3.Session(
        aws_access_key_id=creds["AccessKeyId"],
        aws_secret_access_key=creds["SecretAccessKey"],
        aws_session_token=creds["SessionToken"],
        region_name=REGION,
    )

orch_sess = session_from(orch)

# --- Step 2: orchestrator reads tenant-A external id from secrets manager ---
sm = orch_sess.client("secretsmanager", endpoint_url=EP)
ext_a = sm.get_secret_value(SecretId=SECRET_A_ARN)["SecretString"]
print(f"step 2: pulled tenant-A external-id (len={len(ext_a)})")

# --- Step 3: orchestrator assumes MemberRole-Tenant-A using the external id ---
orch_sts = orch_sess.client("sts", endpoint_url=EP)
member = orch_sts.assume_role(
    RoleArn=MEMBER_A_ARN,
    RoleSessionName="tenant-a-work",
    ExternalId=ext_a,
)["Credentials"]
print("step 3: assumed MemberRole-Tenant-A with ExternalId")

# --- Step 4: member writes a proof object into tenant-a-bucket ---
member_sess = session_from(member)
s3 = member_sess.client("s3", endpoint_url=EP)
s3.put_object(Bucket="tenant-a-bucket", Key="proof.txt",
              Body=b"hello from MemberRole-Tenant-A via OrchestratorRole\n")
print("step 4: wrote s3://tenant-a-bucket/proof.txt")

obj = s3.get_object(Bucket="tenant-a-bucket", Key="proof.txt")["Body"].read()
print(f"        readback: {obj!r}")

print("\nHAPPY PATH OK")