acdl/scripts/rotate_spike_key.sh

#!/usr/bin/env bash
# scripts/rotate_spike_key.sh - rotate the nova-spike-runner IAM access key.
#
# Uses the bootstrap root key (NOVA_BOOTSTRAP_AWS_*
# fallback) from the env to:
#   1. List nova-spike-runner's access keys.
#   2. Create a new key.
#   3. Write the new key to gitignored .env.secrets (chmod 600).
#   4. Upload the new key to the consumer's Actions secret store + verify
#      (GET) that it propagated (SPEC §5.9 idempotency).
#   5. Deactivate + delete the old key(s) ONLY after the upload is verified.
#      If the upload/verify fails, the old key stays Active + the run exits
#      non-zero (the consumer's deploy keeps a working credential).
#
# Env vars (forge coords): NOVA_FORGE_TOKEN / NOVA_FORGE_BASE_URL /
# NOVA_FORGE_OWNER / NOVA_CONSUMER_REPO (the scheduled workflow passes these
# forge-agnostic names, REQ-230). NOVA_GITEA_* are a backward-compat
# fallback for ad-hoc local runs.
#
# Idempotent: re-running always ends with exactly 1 active key for the user
# (once the new key has propagated to the secret store).
# Does NOT rotate the bootstrap root key (D-034 closure = manual user step).
#
# Spike scope (D-039): the spike user key is per-run-rotated; real OIDC is
# v1.2 (blocked on go-gitea/gitea#36988).
# Nova rebrand (P4, REQ-163): IAM user renamed acdl-spike-runner →
# nova-spike-runner.
set -euo pipefail
ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$ROOT"
ENV_FILE="$ROOT/.env.secrets"

fail() { echo "FAIL: $*" >&2; exit 1; }

# Dual-read bootstrap creds: NOVA_* preferred, ACDL_* fallback (removed in P5).
: "${NOVA_BOOTSTRAP_AWS_ACCESS_KEY_ID:?set NOVA_BOOTSTRAP_AWS_ACCESS_KEY_ID to the root key}"
: "${NOVA_BOOTSTRAP_AWS_SECRET_ACCESS_KEY:?set NOVA_BOOTSTRAP_AWS_SECRET_ACCESS_KEY to the root key}"
REGION="${AWS_DEFAULT_REGION:-us-east-1}"
USER_NAME="nova-spike-runner"

# Confirm .env.secrets is gitignored before writing to it.
git check-ignore -q "$ENV_FILE" || fail "$ENV_FILE is not gitignored — refusing to write the key"

python3 - <<'PY'
import os
import sys
import json
import boto3

region = os.environ.get("AWS_DEFAULT_REGION", "us-east-1")
user = "nova-spike-runner"
env_file = os.path.join(os.getcwd(), ".env.secrets")

# Dual-read bootstrap creds: NOVA_* preferred, ACDL_* fallback (G-106, removed in P5).
bootstrap_key = os.environ["NOVA_BOOTSTRAP_AWS_ACCESS_KEY_ID"]
bootstrap_secret = os.environ["NOVA_BOOTSTRAP_AWS_SECRET_ACCESS_KEY"]

session = boto3.Session(
    aws_access_key_id=bootstrap_key,
    aws_secret_access_key=bootstrap_secret,
    region_name=region,
)
iam = session.client("iam")

# List current keys.
keys = iam.list_access_keys(UserName=user).get("AccessKeyMetadata", [])
active = [k for k in keys if k["Status"] == "Active"]

# Create a new key first (so the user always has a working key during rotation).
new = iam.create_access_key(UserName=user)["AccessKey"]
new_id = new["AccessKeyId"]
new_secret = new["SecretAccessKey"]
print(f"iam: created new key {new_id} for {user}", file=sys.stderr)

# Deactivation of the old keys is deferred to AFTER the new key propagates
# to the Gitea Actions secret store (SPEC §5.9 idempotency — see below).
# Writing .env.secrets first keeps the local operator's working key current.

# Write the new key to gitignored .env.secrets (chmod 600).
# Nova rebrand (P2): keys are NOVA_*; the ACDL_* legacy keys are the
# dual-read fallback source until P5 (kept as comments in .env.secrets).
with open(env_file, "w") as fh:
    fh.write(f"NOVA_AWS_ACCESS_KEY_ID={new_id}\n")
    fh.write(f"NOVA_AWS_SECRET_ACCESS_KEY={new_secret}\n")
    fh.write(f"AWS_DEFAULT_REGION={region}\n")
os.chmod(env_file, 0o600)
print(f"rotated key written to {env_file} (chmod 600)", file=sys.stderr)

# Upload the new key to the consumer's Actions secret store BEFORE
# deactivating the old key (SPEC §5.9 — idempotency: the old key is
# deactivated only after the new one propagates). If the upload or the
# post-upload verification fails, the old key is left Active so the
# consumer's deploy still has a working credential; the run exits non-zero
# so the scheduled workflow surfaces the failure (rather than silently
# stranding the consumer with a key that never reached the secret store).
#
# Forge + consumer coords come from env vars. The scheduled workflow passes
# forge-agnostic NOVA_FORGE_* names (REQ-230 — no forge hostnames in the
# synced workflow file); NOVA_GITEA_* are accepted as a backward-compat
# fallback for ad-hoc local runs. Defaults keep the legacy platform-repo
# target when nothing is set.
# Dual-read token: NOVA_FORGE_TOKEN preferred, NOVA_GITEA_TOKEN fallback (G-106).
gitea_token = os.environ.get("NOVA_FORGE_TOKEN") or os.environ.get("NOVA_GITEA_TOKEN")
gitea_base = (
    os.environ.get("NOVA_FORGE_BASE_URL")
    or os.environ.get("NOVA_GITEA_BASE_URL")
    or "https://git.cloudinit.dev"
).rstrip("/")
gitea_owner = (
    os.environ.get("NOVA_FORGE_OWNER")
    or os.environ.get("NOVA_GITEA_OWNER")
    or "continuous-intelligence"
)
gitea_repo = (
    os.environ.get("NOVA_CONSUMER_REPO")
    or os.environ.get("NOVA_GITEA_REPO")
    or "acdl"
)
secrets_api = f"{gitea_base}/api/v1/repos/{gitea_owner}/{gitea_repo}/actions/secrets"

if gitea_token:
    import urllib.request
    import urllib.error
    import time

    def _put_secret(name, value):
        req = urllib.request.Request(
            f"{secrets_api}/{name}",
            data=json.dumps({"value": value}).encode(),
            method="PUT",
            headers={"Authorization": f"token {gitea_token}",
                     "Content-Type": "application/json"},
        )
        urllib.request.urlopen(req).read()
        print(f"gitea: secret {name} uploaded to {gitea_owner}/{gitea_repo}", file=sys.stderr)

    def _verify_secret(name):
        # Gitea does not return secret *values*; a 200 confirms the secret
        # exists with the expected name. Retry briefly so eventual
        # consistency on the secrets API settles (observed sub-second lag).
        for attempt in range(5):
            req = urllib.request.Request(
                f"{secrets_api}/{name}",
                method="GET",
                headers={"Authorization": f"token {gitea_token}"},
            )
            try:
                with urllib.request.urlopen(req) as resp:
                    if resp.status == 200:
                        print(f"gitea: secret {name} verified present", file=sys.stderr)
                        return True
            except urllib.error.HTTPError as e:
                if e.code == 404:
                    time.sleep(0.5)
                    continue
                raise
        return False

    try:
        _put_secret("NOVA_AWS_ACCESS_KEY_ID", new_id)
        _put_secret("NOVA_AWS_SECRET_ACCESS_KEY", new_secret)
        ok = _verify_secret("NOVA_AWS_ACCESS_KEY_ID") and \
            _verify_secret("NOVA_AWS_SECRET_ACCESS_KEY")
        if not ok:
            raise RuntimeError("gitea secret verification failed (404 after PUT)")
    except Exception as e:
        # Upload/verify failed: leave the old key Active so the consumer's
        # deploy still works. Surface non-zero so the schedule is noisy.
        print(f"gitea: secret upload/verify FAILED ({e}); old key left Active", file=sys.stderr)
        sys.exit(2)
else:
    print("gitea: NOVA_FORGE_TOKEN/NOVA_GITEA_TOKEN not set; secret upload skipped (v1.2 hardening)", file=sys.stderr)
    # No forge target → the new key is already in .env.secrets, so the
    # operator's local env works. The old key is deactivated below so the
    # user ends with exactly 1 active key (D-039 local-rotation contract).

# Deactivate + delete the old keys. When a forge token was set, this runs
# ONLY after the new key propagated to the consumer's secret store (the
# sys.exit(2) above prevents reaching here on upload/verify failure). When
# no token was set, the new key is already in .env.secrets so deactivating
# is safe (D-039 local-rotation contract).
for k in active:
    old_id = k["AccessKeyId"]
    if old_id == new_id:
        continue
    iam.update_access_key(UserName=user, AccessKeyId=old_id, Status="Inactive")
    iam.delete_access_key(UserName=user, AccessKeyId=old_id)
    print(f"iam: deactivated+deleted old key {old_id} (after propagation)", file=sys.stderr)

print(f"OK: {user} now has exactly 1 active key: {new_id}")
PY