iam-cross-account-externalid-sourcearn

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

# Build the cross-account trust topology end-to-end.
#
# Layout:
#   user agent-runner    -> assumes -> OrchestratorRole
#   OrchestratorRole     -> assumes -> MemberRole-Tenant-A (gated by ExternalId-A + aws:SourceArn)
#   OrchestratorRole     -> assumes -> MemberRole-Tenant-B (gated by ExternalId-B + aws:SourceArn)
#   MemberRole-Tenant-X  -> writes  -> tenant-X-bucket only

ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
REGION=us-east-1
PARTITION=aws

ORCH_ARN="arn:${PARTITION}:iam::${ACCOUNT_ID}:role/orchestrator/OrchestratorRole"
MEMBER_A_ARN="arn:${PARTITION}:iam::${ACCOUNT_ID}:role/member/MemberRole-Tenant-A"
MEMBER_B_ARN="arn:${PARTITION}:iam::${ACCOUNT_ID}:role/member/MemberRole-Tenant-B"
AGENT_USER_ARN="arn:${PARTITION}:iam::${ACCOUNT_ID}:user/agent-runner"

echo "==> Account: ${ACCOUNT_ID}"
echo "==> Orchestrator ARN (planned): ${ORCH_ARN}"

# -----------------------------------------------------------------------------
# 1) IAM users
# -----------------------------------------------------------------------------
aws iam create-user --user-name agent-runner >/dev/null
aws iam create-user --user-name attacker-runner >/dev/null
echo "==> Created users: agent-runner, attacker-runner"

# -----------------------------------------------------------------------------
# 2) Distinct ExternalIds (>=32 chars, random, per-tenant)
# -----------------------------------------------------------------------------
EXTID_A=$(python3 -c "import secrets; print(secrets.token_urlsafe(36))")
EXTID_B=$(python3 -c "import secrets; print(secrets.token_urlsafe(36))")
if [[ "$EXTID_A" == "$EXTID_B" ]]; then
  echo "ExternalIds collided (impossible but checking)" >&2
  exit 1
fi
echo "==> Generated distinct ExternalIds (A=${#EXTID_A} chars, B=${#EXTID_B} chars)"

# -----------------------------------------------------------------------------
# 3) Customer-managed KMS key + alias
# -----------------------------------------------------------------------------
KEY_ID=$(aws kms create-key --description "Cross-account ExternalId CMK" --query KeyMetadata.KeyId --output text)
KEY_ARN=$(aws kms describe-key --key-id "$KEY_ID" --query KeyMetadata.Arn --output text)
aws kms create-alias --alias-name alias/cross-account-cmk --target-key-id "$KEY_ID"
echo "==> Created CMK ${KEY_ARN} with alias alias/cross-account-cmk"

# Tighten the key policy: explicit principals, explicit resource ARN, no wildcards.
KEY_POLICY=$(cat <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "RootKeyAdmin",
      "Effect": "Allow",
      "Principal": {"AWS": "arn:${PARTITION}:iam::${ACCOUNT_ID}:root"},
      "Action": [
        "kms:CreateAlias",
        "kms:DeleteAlias",
        "kms:UpdateAlias",
        "kms:DescribeKey",
        "kms:EnableKey",
        "kms:DisableKey",
        "kms:GetKeyPolicy",
        "kms:PutKeyPolicy",
        "kms:ListAliases",
        "kms:ListKeys",
        "kms:TagResource",
        "kms:UntagResource",
        "kms:ScheduleKeyDeletion",
        "kms:CancelKeyDeletion"
      ],
      "Resource": "${KEY_ARN}"
    },
    {
      "Sid": "OrchestratorDecryptOnly",
      "Effect": "Allow",
      "Principal": {"AWS": "${ORCH_ARN}"},
      "Action": "kms:Decrypt",
      "Resource": "${KEY_ARN}"
    },
    {
      "Sid": "SecretsManagerUseKey",
      "Effect": "Allow",
      "Principal": {"Service": "secretsmanager.amazonaws.com"},
      "Action": [
        "kms:Encrypt",
        "kms:Decrypt",
        "kms:GenerateDataKey",
        "kms:DescribeKey"
      ],
      "Resource": "${KEY_ARN}"
    }
  ]
}
EOF
)
aws kms put-key-policy --key-id "$KEY_ID" --policy-name default --policy "$KEY_POLICY"
echo "==> CMK key policy tightened (no wildcards on actions or resources)"

# -----------------------------------------------------------------------------
# 4) Orchestrator role: trusted by the agent-runner IAM user
# -----------------------------------------------------------------------------
ORCH_TRUST=$(cat <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AgentRunnerAssumesOrchestrator",
      "Effect": "Allow",
      "Principal": {"AWS": "${AGENT_USER_ARN}"},
      "Action": "sts:AssumeRole"
    }
  ]
}
EOF
)
aws iam create-role \
  --role-name OrchestratorRole \
  --path /orchestrator/ \
  --assume-role-policy-document "$ORCH_TRUST" >/dev/null
echo "==> Created OrchestratorRole (path /orchestrator/)"

# -----------------------------------------------------------------------------
# 5) Member role trust policies: ExternalId + aws:SourceArn together
# -----------------------------------------------------------------------------
build_member_trust() {
  local extid="$1"
  cat <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "OrchestratorAssumesMember",
      "Effect": "Allow",
      "Principal": {"AWS": "${ORCH_ARN}"},
      "Action": "sts:AssumeRole",
      "Condition": {
        "StringEquals": {
          "sts:ExternalId": "${extid}"
        },
        "ArnEquals": {
          "aws:SourceArn": "${ORCH_ARN}"
        }
      }
    }
  ]
}
EOF
}

aws iam create-role \
  --role-name MemberRole-Tenant-A \
  --path /member/ \
  --assume-role-policy-document "$(build_member_trust "$EXTID_A")" >/dev/null
echo "==> Created MemberRole-Tenant-A (gated by ExternalId-A + aws:SourceArn)"

aws iam create-role \
  --role-name MemberRole-Tenant-B \
  --path /member/ \
  --assume-role-policy-document "$(build_member_trust "$EXTID_B")" >/dev/null
echo "==> Created MemberRole-Tenant-B (gated by ExternalId-B + aws:SourceArn)"

# -----------------------------------------------------------------------------
# 6) Tenant buckets
# -----------------------------------------------------------------------------
aws s3api create-bucket --bucket tenant-a-bucket >/dev/null
aws s3api create-bucket --bucket tenant-b-bucket >/dev/null
echo "==> Created buckets: tenant-a-bucket, tenant-b-bucket"

# -----------------------------------------------------------------------------
# 7) Member-role inline policies: each scoped to its OWN bucket only
# -----------------------------------------------------------------------------
build_member_inline() {
  local bucket="$1"
  cat <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "TenantBucketReadWrite",
      "Effect": "Allow",
      "Action": [
        "s3:PutObject",
        "s3:GetObject",
        "s3:DeleteObject",
        "s3:ListBucket"
      ],
      "Resource": [
        "arn:${PARTITION}:s3:::${bucket}",
        "arn:${PARTITION}:s3:::${bucket}/*"
      ]
    }
  ]
}
EOF
}

aws iam put-role-policy \
  --role-name MemberRole-Tenant-A \
  --policy-name TenantABucketAccess \
  --policy-document "$(build_member_inline tenant-a-bucket)"

aws iam put-role-policy \
  --role-name MemberRole-Tenant-B \
  --policy-name TenantBBucketAccess \
  --policy-document "$(build_member_inline tenant-b-bucket)"
echo "==> Member role inline policies pinned to their own bucket only"

# -----------------------------------------------------------------------------
# 8) Secrets Manager: ExternalId values, encrypted with the CMK
# -----------------------------------------------------------------------------
SECRET_A_ARN=$(aws secretsmanager create-secret \
  --name harbor/cross-account/external-id-tenant-a \
  --description "ExternalId shared secret for MemberRole-Tenant-A trust" \
  --kms-key-id "$KEY_ARN" \
  --secret-string "$EXTID_A" \
  --query ARN --output text)

SECRET_B_ARN=$(aws secretsmanager create-secret \
  --name harbor/cross-account/external-id-tenant-b \
  --description "ExternalId shared secret for MemberRole-Tenant-B trust" \
  --kms-key-id "$KEY_ARN" \
  --secret-string "$EXTID_B" \
  --query ARN --output text)
echo "==> Stored ExternalIds in Secrets Manager (CMK-encrypted)"

# -----------------------------------------------------------------------------
# 9) Resource policy on each secret: orchestrator role allowed by name
#    The auditor checks both sides , identity-side scope alone isn't enough.
# -----------------------------------------------------------------------------
SECRET_RESOURCE_POLICY=$(cat <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "OrchestratorReadsExternalId",
      "Effect": "Allow",
      "Principal": {"AWS": "${ORCH_ARN}"},
      "Action": "secretsmanager:GetSecretValue",
      "Resource": "*"
    }
  ]
}
EOF
)
aws secretsmanager put-resource-policy \
  --secret-id "$SECRET_A_ARN" \
  --resource-policy "$SECRET_RESOURCE_POLICY" >/dev/null
aws secretsmanager put-resource-policy \
  --secret-id "$SECRET_B_ARN" \
  --resource-policy "$SECRET_RESOURCE_POLICY" >/dev/null
echo "==> Secret resource policies admit OrchestratorRole only"

# -----------------------------------------------------------------------------
# 10) Orchestrator identity policy: explicit member ARNs, explicit secret ARNs,
#     explicit CMK ARN. No wildcards.
# -----------------------------------------------------------------------------
ORCH_INLINE=$(cat <<EOF
{
  "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": "DecryptExternalIdCMK",
      "Effect": "Allow",
      "Action": "kms:Decrypt",
      "Resource": "${KEY_ARN}"
    }
  ]
}
EOF
)
aws iam put-role-policy \
  --role-name OrchestratorRole \
  --policy-name OrchestratorLeastPrivilege \
  --policy-document "$ORCH_INLINE"
echo "==> OrchestratorRole inline policy attached (explicit ARNs only)"

# -----------------------------------------------------------------------------
# 11) Agent-runner user: only allowed to assume OrchestratorRole
# -----------------------------------------------------------------------------
AGENT_INLINE=$(cat <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AssumeOrchestratorOnly",
      "Effect": "Allow",
      "Action": "sts:AssumeRole",
      "Resource": "${ORCH_ARN}"
    }
  ]
}
EOF
)
aws iam put-user-policy \
  --user-name agent-runner \
  --policy-name AgentRunnerAssumeOrchestrator \
  --policy-document "$AGENT_INLINE"
echo "==> agent-runner user can assume OrchestratorRole and nothing else"

# -----------------------------------------------------------------------------
# 12) SSM pointers under /harbor/ so the verifier can find things by name
# -----------------------------------------------------------------------------
aws ssm put-parameter --name /harbor/orchestrator/role-arn       --type String --overwrite --value "$ORCH_ARN"     >/dev/null
aws ssm put-parameter --name /harbor/external-id-secret-arn-tenant-a --type String --overwrite --value "$SECRET_A_ARN" >/dev/null
aws ssm put-parameter --name /harbor/external-id-secret-arn-tenant-b --type String --overwrite --value "$SECRET_B_ARN" >/dev/null
aws ssm put-parameter --name /harbor/member-role-arn-tenant-a    --type String --overwrite --value "$MEMBER_A_ARN" >/dev/null
aws ssm put-parameter --name /harbor/member-role-arn-tenant-b    --type String --overwrite --value "$MEMBER_B_ARN" >/dev/null
echo "==> SSM pointers under /harbor/ written"

echo
echo "Setup complete."
echo "  OrchestratorRole : ${ORCH_ARN}"
echo "  MemberRole-A     : ${MEMBER_A_ARN}"
echo "  MemberRole-B     : ${MEMBER_B_ARN}"
echo "  Secret-A ARN     : ${SECRET_A_ARN}"
echo "  Secret-B ARN     : ${SECRET_B_ARN}"
echo "  CMK ARN          : ${KEY_ARN}"

#!/usr/bin/env python3
"""Auditor-style policy-shape verifier.

Reads role and policy documents directly via IAM/SecretsManager/SSM and
asserts the cross-account trust pattern is shaped correctly. Mirrors what
an external auditor would check: nothing here exercises STS at runtime,
the verdict comes from the JSON of the policies themselves.
"""

import json
import sys

import boto3


PARTITION = "aws"
REGION = "us-east-1"

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

failures: list[str] = []
checks_passed: list[str] = []


def fail(msg: str) -> None:
    failures.append(msg)
    print(f"  FAIL: {msg}")


def ok(msg: str) -> None:
    checks_passed.append(msg)
    print(f"  ok:   {msg}")


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


def has_literal_star_action(stmt: dict) -> bool:
    return any(a == "*" for a in as_list(stmt.get("Action")))


def has_literal_star_resource(stmt: dict) -> bool:
    return any(r == "*" for r in as_list(stmt.get("Resource")))


def touches_service(stmt: dict, service: str) -> bool:
    actions = as_list(stmt.get("Action"))
    return any(a == "*" or a.startswith(f"{service}:") for a in actions)


# -----------------------------------------------------------------------------
# Resolve identifiers via SSM (pointers the operator placed under /harbor/)
# -----------------------------------------------------------------------------
def ssm_get(name: str) -> str:
    return ssm.get_parameter(Name=name)["Parameter"]["Value"]


print("=== resolving SSM pointers ===")
orch_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")

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}")


# -----------------------------------------------------------------------------
# Pull current ExternalId values out of Secrets Manager
# -----------------------------------------------------------------------------
extid_a = sm.get_secret_value(SecretId=secret_a_arn)["SecretString"]
extid_b = sm.get_secret_value(SecretId=secret_b_arn)["SecretString"]

print("\n=== ExternalId distinctness ===")
if extid_a == extid_b:
    fail("ExternalId values are identical between tenants , defeats the per-tenant secret")
else:
    ok("ExternalId-A and ExternalId-B differ")

if len(extid_a) >= 32 and len(extid_b) >= 32:
    ok(f"both ExternalIds are >= 32 chars (A={len(extid_a)}, B={len(extid_b)})")
else:
    fail(f"ExternalId length too short (A={len(extid_a)}, B={len(extid_b)})")


# -----------------------------------------------------------------------------
# Enumerate member roles under /member/ , by path, not by name
# -----------------------------------------------------------------------------
print("\n=== member role trust documents (under path /member/) ===")
member_roles = iam.list_roles(PathPrefix="/member/")["Roles"]
member_role_names = sorted(r["RoleName"] for r in member_roles)
expected_member_names = sorted(["MemberRole-Tenant-A", "MemberRole-Tenant-B"])
if member_role_names != expected_member_names:
    fail(f"unexpected member roles under /member/: {member_role_names}")
else:
    ok(f"member roles under /member/: {member_role_names}")

tenant_to_extid = {
    "MemberRole-Tenant-A": extid_a,
    "MemberRole-Tenant-B": extid_b,
}
tenant_to_bucket = {
    "MemberRole-Tenant-A": "tenant-a-bucket",
    "MemberRole-Tenant-B": "tenant-b-bucket",
    "other": {"MemberRole-Tenant-A": "tenant-b-bucket", "MemberRole-Tenant-B": "tenant-a-bucket"},
}

for role in member_roles:
    rn = role["RoleName"]
    print(f"\n-- {rn} trust --")
    trust = role["AssumeRolePolicyDocument"]
    statements = as_list(trust.get("Statement"))

    if len(statements) != 1:
        fail(f"{rn}: trust has {len(statements)} statements, expected exactly 1")
        continue
    s = statements[0]

    # Action must be exactly sts:AssumeRole.
    actions = as_list(s.get("Action"))
    if actions == ["sts:AssumeRole"]:
        ok(f"{rn}: trust action is exactly sts:AssumeRole")
    else:
        fail(f"{rn}: trust action is {actions}, expected ['sts:AssumeRole']")

    # Principal must be the orchestrator role ARN, NOT '*'.
    principal = s.get("Principal", {})
    if principal == "*" or principal.get("AWS") == "*":
        fail(f"{rn}: principal is wildcard")
    else:
        principal_aws = as_list(principal.get("AWS"))
        if principal_aws == [orch_arn]:
            ok(f"{rn}: principal is exactly OrchestratorRole ARN")
        else:
            fail(f"{rn}: principal.AWS={principal_aws}, expected [{orch_arn}]")

    # Effect must be Allow (sanity).
    if s.get("Effect") != "Allow":
        fail(f"{rn}: trust effect is {s.get('Effect')}, expected Allow")

    # Condition must include BOTH ExternalId AND aws:SourceArn.
    conditions = s.get("Condition", {})
    extid_block = conditions.get("StringEquals", {})
    extid_value = extid_block.get("sts:ExternalId")
    if extid_value is None:
        fail(f"{rn}: missing StringEquals['sts:ExternalId'] , third-party impersonation defense absent")
    elif extid_value != tenant_to_extid[rn]:
        fail(f"{rn}: ExternalId in trust does not match the secret value")
    else:
        ok(f"{rn}: trust requires ExternalId matching this tenant's secret")

    src_arn_value = None
    for op in ("ArnEquals", "ArnLike"):
        block = conditions.get(op, {})
        if "aws:SourceArn" in block:
            src_arn_value = block["aws:SourceArn"]
            src_arn_op = op
            break
    if src_arn_value is None:
        fail(f"{rn}: missing ArnEquals/ArnLike on aws:SourceArn , confused-deputy defense absent")
    elif src_arn_value != orch_arn:
        fail(f"{rn}: aws:SourceArn={src_arn_value}, expected {orch_arn}")
    else:
        ok(f"{rn}: trust requires aws:SourceArn={orch_arn} ({src_arn_op})")


# -----------------------------------------------------------------------------
# Orchestrator role identity policy: explicit ARN enumeration, no wildcards
# -----------------------------------------------------------------------------
print("\n=== orchestrator role inline policy ===")
orch_roles = iam.list_roles(PathPrefix="/orchestrator/")["Roles"]
orch_role_names = [r["RoleName"] for r in orch_roles]
if orch_role_names != ["OrchestratorRole"]:
    fail(f"unexpected roles under /orchestrator/: {orch_role_names}")
else:
    ok("orchestrator role found under /orchestrator/")

orch_inline_names = iam.list_role_policies(RoleName="OrchestratorRole")["PolicyNames"]
saw_assume = False
saw_secrets = False
for pn in orch_inline_names:
    doc = iam.get_role_policy(RoleName="OrchestratorRole", PolicyName=pn)["PolicyDocument"]
    for s in as_list(doc.get("Statement")):
        actions = as_list(s.get("Action"))
        resources = as_list(s.get("Resource"))

        # No literal-star anywhere on the orchestrator's own policy.
        if has_literal_star_action(s):
            fail(f"OrchestratorRole/{pn}: statement uses Action '*'")
        if has_literal_star_resource(s):
            fail(f"OrchestratorRole/{pn}: statement uses Resource '*'")

        if "sts:AssumeRole" in actions:
            saw_assume = True
            target_set = sorted(resources)
            expected = sorted([member_a_arn, member_b_arn])
            if target_set == expected:
                ok("orchestrator's sts:AssumeRole resources are exactly [MemberRole-Tenant-A, MemberRole-Tenant-B]")
            else:
                fail(f"orchestrator sts:AssumeRole resources={target_set}, expected {expected}")
            if "sts:*" in actions or any(a.endswith(":*") for a in actions):
                fail(f"OrchestratorRole/{pn}: AssumeRole statement also has wildcard sts action")

        if "secretsmanager:GetSecretValue" in actions:
            saw_secrets = True
            target_set = sorted(resources)
            expected = sorted([secret_a_arn, secret_b_arn])
            if target_set == expected:
                ok("orchestrator's secretsmanager:GetSecretValue resources are the two known secret ARNs")
            else:
                fail(f"orchestrator GetSecretValue resources={target_set}, expected {expected}")

if not saw_assume:
    fail("orchestrator policy has no sts:AssumeRole statement")
if not saw_secrets:
    fail("orchestrator policy has no secretsmanager:GetSecretValue statement")


# -----------------------------------------------------------------------------
# Member-role inline policies: scoped to OWN bucket, no other tenant's bucket
# -----------------------------------------------------------------------------
print("\n=== member role inline policies (bucket isolation) ===")
for role_name, own_bucket, other_bucket in [
    ("MemberRole-Tenant-A", "tenant-a-bucket", "tenant-b-bucket"),
    ("MemberRole-Tenant-B", "tenant-b-bucket", "tenant-a-bucket"),
]:
    inline_names = iam.list_role_policies(RoleName=role_name)["PolicyNames"]
    if not inline_names:
        fail(f"{role_name}: no inline policies attached")
        continue
    for pn in inline_names:
        doc = iam.get_role_policy(RoleName=role_name, PolicyName=pn)["PolicyDocument"]
        for s in as_list(doc.get("Statement")):
            actions = as_list(s.get("Action"))
            resources = as_list(s.get("Resource"))

            if has_literal_star_action(s) and (touches_service(s, "s3") or touches_service(s, "kms")):
                fail(f"{role_name}/{pn}: Action '*' on s3/kms statement")
            if has_literal_star_resource(s) and (touches_service(s, "s3") or touches_service(s, "kms")):
                fail(f"{role_name}/{pn}: Resource '*' on s3/kms statement")

            if touches_service(s, "s3"):
                if "s3:*" in actions:
                    fail(f"{role_name}/{pn}: uses s3:* wildcard action")
                bad_resource = False
                for r in resources:
                    if other_bucket in r:
                        fail(f"{role_name}/{pn}: references the OTHER tenant's bucket: {r}")
                        bad_resource = True
                    elif own_bucket not in r:
                        fail(f"{role_name}/{pn}: unexpected resource: {r}")
                        bad_resource = True
                if not bad_resource and any(own_bucket in r for r in resources):
                    ok(f"{role_name}/{pn}: s3 resources reference only {own_bucket}")


# -----------------------------------------------------------------------------
# Secret resource policies: orchestrator role admitted by name
# -----------------------------------------------------------------------------
print("\n=== secret resource policies ===")
for label, secret_arn in [("tenant-a", secret_a_arn), ("tenant-b", secret_b_arn)]:
    rp_raw = sm.get_resource_policy(SecretId=secret_arn).get("ResourcePolicy")
    if not rp_raw:
        fail(f"{label} secret: no resource policy attached")
        continue
    rp = json.loads(rp_raw)
    found = False
    for s in as_list(rp.get("Statement")):
        principals = as_list(s.get("Principal", {}).get("AWS"))
        actions = as_list(s.get("Action"))
        if (
            s.get("Effect") == "Allow"
            and orch_arn in principals
            and "secretsmanager:GetSecretValue" in actions
        ):
            found = True
            if "*" in principals:
                fail(f"{label} secret resource policy has wildcard principal alongside orchestrator")
    if found:
        ok(f"{label} secret resource policy admits OrchestratorRole for GetSecretValue")
    else:
        fail(f"{label} secret resource policy does not admit OrchestratorRole")


# -----------------------------------------------------------------------------
# Global wildcard scan: any policy in scope with Action:'*' or Resource:'*' on s3/kms
# -----------------------------------------------------------------------------
print("\n=== global wildcard scan over in-scope policies ===")
all_role_pairs = []
for path in ("/orchestrator/", "/member/"):
    for role in iam.list_roles(PathPrefix=path)["Roles"]:
        for pn in iam.list_role_policies(RoleName=role["RoleName"])["PolicyNames"]:
            doc = iam.get_role_policy(RoleName=role["RoleName"], PolicyName=pn)["PolicyDocument"]
            all_role_pairs.append((f"{role['RoleName']}/{pn}", doc))

violations = 0
for label, doc in all_role_pairs:
    for s in as_list(doc.get("Statement")):
        if touches_service(s, "s3") or touches_service(s, "kms"):
            if has_literal_star_action(s):
                fail(f"{label}: Action '*' on s3/kms statement")
                violations += 1
            if has_literal_star_resource(s):
                fail(f"{label}: Resource '*' on s3/kms statement")
                violations += 1
if violations == 0:
    ok("no Action:'*' or Resource:'*' found in any in-scope s3/kms statement")


# -----------------------------------------------------------------------------
# Verdict
# -----------------------------------------------------------------------------
print(f"\n=== verdict ===")
print(f"checks passed : {len(checks_passed)}")
print(f"checks failed : {len(failures)}")
if failures:
    print("\nFAILURES:")
    for f in failures:
        print(f"  - {f}")
    sys.exit(1)
print("\nALL CHECKS PASSED , auditor is satisfied with the policy shape.")

#!/usr/bin/env python3
"""End-to-end happy path: orchestrator fans out into tenant A's bucket.

Flow:
  agent-runner user
    -> AssumeRole(OrchestratorRole)
    -> GetSecretValue(external-id-tenant-a)
    -> AssumeRole(MemberRole-Tenant-A, ExternalId=<secret value>)
    -> PutObject(tenant-a-bucket, hello.txt)
"""

import sys

import boto3


REGION = "us-east-1"

ssm = boto3.client("ssm", region_name=REGION)


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


orch_arn = ssm_get("/harbor/orchestrator/role-arn")
secret_a_arn = ssm_get("/harbor/external-id-secret-arn-tenant-a")
member_a_arn = ssm_get("/harbor/member-role-arn-tenant-a")

print(f"orchestrator    : {orch_arn}")
print(f"secret-A        : {secret_a_arn}")
print(f"member-A        : {member_a_arn}")

# Step 1: as agent-runner, assume OrchestratorRole.
agent_sts = boto3.client("sts", region_name=REGION)
orch_session = agent_sts.assume_role(
    RoleArn=orch_arn,
    RoleSessionName="agent-runner-session",
)["Credentials"]
print("\n[1] agent-runner -> OrchestratorRole: session acquired")

orch_kw = dict(
    aws_access_key_id=orch_session["AccessKeyId"],
    aws_secret_access_key=orch_session["SecretAccessKey"],
    aws_session_token=orch_session["SessionToken"],
    region_name=REGION,
)

# Step 2: as OrchestratorRole, fetch the tenant-A ExternalId secret.
orch_sm = boto3.client("secretsmanager", **orch_kw)
ext_id_a = orch_sm.get_secret_value(SecretId=secret_a_arn)["SecretString"]
print(f"[2] OrchestratorRole -> read tenant-A ExternalId ({len(ext_id_a)} chars)")

# Step 3: as OrchestratorRole, AssumeRole on MemberRole-Tenant-A with ExternalId.
orch_sts = boto3.client("sts", **orch_kw)
member_session = orch_sts.assume_role(
    RoleArn=member_a_arn,
    RoleSessionName="member-tenant-a-session",
    ExternalId=ext_id_a,
)["Credentials"]
print("[3] OrchestratorRole -> MemberRole-Tenant-A: session acquired (ExternalId presented)")

# Step 4: as MemberRole-Tenant-A, write an object into tenant-a-bucket.
member_s3 = boto3.client(
    "s3",
    aws_access_key_id=member_session["AccessKeyId"],
    aws_secret_access_key=member_session["SecretAccessKey"],
    aws_session_token=member_session["SessionToken"],
    region_name=REGION,
)
member_s3.put_object(
    Bucket="tenant-a-bucket",
    Key="hello.txt",
    Body=b"orchestrated write , tenant A only\n",
)
print("[4] MemberRole-Tenant-A -> PutObject(tenant-a-bucket/hello.txt): OK")

# Sanity: read it back through the same session.
body = member_s3.get_object(Bucket="tenant-a-bucket", Key="hello.txt")["Body"].read()
print(f"[5] readback: {body!r}")

print("\nhappy path complete.")