"""Apply the ACDL spike-runner managed policy + OIDC act_runner role. Phase 56 (REQ-116, v1.11). Idempotent: re-running creates the managed policy if absent (or creates a new version if the policy document differs), attaches it to the spike-runner user, deletes any leftover inline policy, and re-creates the OIDC act_runner role if absent. Requires the bootstrap root key (ACDL_BOOTSTRAP_AWS_* or ACDL_AWS_* when the provided key is a root principal). This script is the reproducible record of the Phase 56 live step — the grants are documented in .ciagent/IAM_POLICY.md and regression-tested by tests/test_iam_policy_baseline.py. Usage: export ACDL_BOOTSTRAP_AWS_ACCESS_KEY_ID= export ACDL_BOOTSTRAP_AWS_SECRET_ACCESS_KEY= export AWS_DEFAULT_REGION=us-east-1 python3 terraform/bootstrap/apply_iam_baseline.py """ from __future__ import annotations import json import os import sys import time from pathlib import Path import boto3 ROOT = Path(__file__).resolve().parent.parent.parent POLICY_PATH = ROOT / "terraform" / "bootstrap" / "spike_runner_policy.json" ACCOUNT = os.environ.get("ACDL_AWS_ACCOUNT_ID", "581513795199") USER = "acdl-spike-runner" POLICY_NAME = "acdl-spike-runner-policy" POLICY_ARN = f"arn:aws:iam::{ACCOUNT}:policy/{POLICY_NAME}" ROLE_NAME = "acdl-act-runner-role" def _session(): key_id = os.environ.get("ACDL_BOOTSTRAP_AWS_ACCESS_KEY_ID") or os.environ.get("ACDL_AWS_ACCESS_KEY_ID") secret = os.environ.get("ACDL_BOOTSTRAP_AWS_SECRET_ACCESS_KEY") or os.environ.get("ACDL_AWS_SECRET_ACCESS_KEY") if not key_id or not secret: sys.exit("FAIL: set ACDL_BOOTSTRAP_AWS_ACCESS_KEY_ID + ACDL_BOOTSTRAP_AWS_SECRET_ACCESS_KEY (root key)") region = os.environ.get("AWS_DEFAULT_REGION", "us-east-1") return boto3.Session(aws_access_key_id=key_id, aws_secret_access_key=secret, region_name=region) def _trust_policy_for_runner(): return { "Version": "2012-10-17", "Statement": [ { "Sid": "AllowRootAssumeUntilOIDCLands", "Effect": "Allow", "Principal": {"AWS": f"arn:aws:iam::{ACCOUNT}:root"}, "Action": "sts:AssumeRole", } ], } def apply_managed_policy(iam, policy_doc: str) -> str: try: existing = iam.get_policy(PolicyArn=POLICY_ARN) print(f"managed policy exists: {POLICY_ARN} (default version {existing['Policy']['DefaultVersionId']})") new_version = iam.create_policy_version( PolicyArn=POLICY_ARN, PolicyDocument=policy_doc, SetAsDefault=True, ) print(f"created new version {new_version['PolicyVersion']['VersionId']} (set as default)") default = existing["Policy"]["DefaultVersionId"] if default != new_version["PolicyVersion"]["VersionId"]: try: iam.delete_policy_version(PolicyArn=POLICY_ARN, VersionId=default) print(f"deleted old default version {default}") except iam.exceptions.NoSuchEntityException: pass # already deleted except Exception as e: print(f"WARNING: could not delete old version {default}: {e}") return POLICY_ARN except iam.exceptions.NoSuchEntityException: print(f"creating managed policy {POLICY_NAME}...") created = iam.create_policy( PolicyName=POLICY_NAME, Path="/", PolicyDocument=policy_doc, Description="ACDL spike-runner baseline (v1.11 REQ-116). Extended from inline user policy to managed policy to fit the 6144-byte limit.", ) print(f"created: {created['Policy']['Arn']}") return created["Policy"]["Arn"] def attach_and_cleanup_inline(iam): attached = iam.list_attached_user_policies(UserName=USER).get("AttachedPolicies", []) if any(p["PolicyArn"] == POLICY_ARN for p in attached): print(f"{POLICY_NAME} already attached to {USER}") else: print(f"attaching {POLICY_ARN} to {USER}...") iam.attach_user_policy(UserName=USER, PolicyArn=POLICY_ARN) print("attached") inline = iam.list_user_policies(UserName=USER).get("PolicyNames", []) if POLICY_NAME in inline: print(f"deleting leftover inline policy {POLICY_NAME} from {USER}...") iam.delete_user_policy(UserName=USER, PolicyName=POLICY_NAME) print("inline policy deleted") def ensure_runner_role(iam): try: iam.get_role(RoleName=ROLE_NAME) print(f"role {ROLE_NAME} already exists — updating trust policy + ensuring policy attached") iam.update_assume_role_policy(RoleName=ROLE_NAME, PolicyDocument=json.dumps(_trust_policy_for_runner())) except iam.exceptions.NoSuchEntityException: print(f"creating role {ROLE_NAME}...") iam.create_role( RoleName=ROLE_NAME, AssumeRolePolicyDocument=json.dumps(_trust_policy_for_runner()), Description="ACDL act_runner OIDC role (CAP-022, v1.11 Phase 56 re-creation). Trust policy permits root assume until go-gitea/gitea#36988 merges real OIDC federation.", MaxSessionDuration=3600, Tags=[ {"Key": "Project", "Value": "acdl"}, {"Key": "Capability", "Value": "CAP-022"}, {"Key": "Milestone", "Value": "v1.11"}, {"Key": "ManagedBy", "Value": "ciagent"}, ], ) time.sleep(2) attached = iam.list_attached_role_policies(RoleName=ROLE_NAME).get("AttachedPolicies", []) if not any(p["PolicyArn"] == POLICY_ARN for p in attached): print(f"attaching {POLICY_ARN} to {ROLE_NAME}...") iam.attach_role_policy(RoleName=ROLE_NAME, PolicyArn=POLICY_ARN) print("attached") def main(): policy_doc = POLICY_PATH.read_text() sess = _session() sts = sess.client("sts") ident = sts.get_caller_identity() print(f"caller: {ident['Arn']}") if ":root" not in ident["Arn"] and "assumed-role" not in ident["Arn"]: sys.exit(f"FAIL: caller {ident['Arn']} is not a root or admin principal — cannot put IAM policy") iam = sess.client("iam") apply_managed_policy(iam, policy_doc) attach_and_cleanup_inline(iam) ensure_runner_role(iam) print("\n=== verification ===") attached = iam.list_attached_user_policies(UserName=USER).get("AttachedPolicies", []) print(f"spike-runner attached managed policies: {[p['PolicyName'] for p in attached]}") inline = iam.list_user_policies(UserName=USER).get("PolicyNames", []) print(f"spike-runner inline policies: {inline}") role = iam.get_role(RoleName=ROLE_NAME)["Role"] print(f"act_runner role: {role['Arn']}") role_attached = iam.list_attached_role_policies(RoleName=ROLE_NAME).get("AttachedPolicies", []) print(f"act_runner attached policies: {[p['PolicyName'] for p in role_attached]}") print("\nOK: IAM baseline applied") if __name__ == "__main__": main()