acdl/terraform/bootstrap/apply_iam_baseline.py

"""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=<root key id>
    export ACDL_BOOTSTRAP_AWS_SECRET_ACCESS_KEY=<root key secret>
    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 = "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 Exception as e:
                print(f"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()