scripts/chaos_aws.py

"""Per-task AWS deformity inducer + verifier + reset.

Every AWS-touching task in scenarios/ has a real-world deformity that can be
induced on the live AWS account, verified through a separate read-only API
call, and then cleaned up. This script is the single source of truth.

Usage:
    python scripts/chaos_aws.py list
    python scripts/chaos_aws.py induce task12
    python scripts/chaos_aws.py verify task12
    python scripts/chaos_aws.py reset  task12
    python scripts/chaos_aws.py cycle  task12       # induce -> verify -> reset -> verify
    python scripts/chaos_aws.py cycle-all           # exercise every supported task

Supported tasks (everything that touches AWS for real):
    task9   ECR — image-tag rollback drift             (read-only probe)
    task12  SQS DLQ growth on ic-orders-dlq
    task13  DynamoDB throttling on ic-inventory
    task14  Secrets Manager — stale rotation marker
    task15  S3 + IAM drift — bucket policy denies PutObject
    task16  Lambda cold-start storm on ic-cold-start-target
    task17  RDS pool exhaustion proxy via SSM /ic/payments/db-pool-size
    task18  Bedrock InvokeModel saturation (best-effort metric)
    task19  CloudWatch alarm storm — flips ic-cw-alarm-storm to ALARM
    task20  EventBridge silent drop — disables ic-orders-rule
    task21  Step Functions execution failure on ic-order-saga
    task22  Athena — malformed query lands in FAILED state
    task23  KMS — ic-data-key scheduled for deletion (drift)
"""

from __future__ import annotations

import argparse
import json
import os
import sys
import time
from pathlib import Path
from typing import Any, Callable

ROOT = Path(__file__).resolve().parent.parent
ENV_FILE = ROOT / ".env.aws.local"
for _line in (ENV_FILE.read_text().splitlines() if ENV_FILE.exists() else []):
    _line = _line.strip()
    if not _line or _line.startswith("#") or "=" not in _line:
        continue
    _k, _, _v = _line.partition("=")
    os.environ.setdefault(_k.strip(),
                          _v.strip().strip('"').strip("'"))

REGION = os.environ.get("AWS_REGION", "us-east-1")

import boto3                                 # noqa: E402
from botocore.exceptions import ClientError  # noqa: E402

ACCOUNT = boto3.client("sts", region_name=REGION).get_caller_identity()["Account"]


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _tag(s: str)   -> str: return f"\033[36m{s}\033[0m"  # cyan
def _ok(s: str)    -> str: return f"\033[32m{s}\033[0m"  # green
def _err(s: str)   -> str: return f"\033[31m{s}\033[0m"  # red
def _warn(s: str)  -> str: return f"\033[33m{s}\033[0m"  # yellow

def log(prefix: str, msg: str) -> None:
    print(f"{prefix} {msg}", flush=True)


# ---------------------------------------------------------------------------
# Per-task handlers — each returns (induce_fn, verify_fn, reset_fn).
# Verify functions return {"ok": bool, "detail": str, "data": dict}.
# ---------------------------------------------------------------------------

# ----- task9: ECR rollback probe -------------------------------------------
def _t9_induce() -> dict:
    ecr = boto3.client("ecr", region_name=REGION)
    repo = "incident-commander"
    try:
        ecr.describe_repositories(repositoryNames=[repo])
        return {"action": "noop", "detail": f"ECR repo {repo} reachable"}
    except ClientError as e:
        if e.response["Error"]["Code"] == "RepositoryNotFoundException":
            ecr.create_repository(repositoryName=repo)
            return {"action": "created", "detail": repo}
        raise

def _t9_verify() -> dict:
    ecr = boto3.client("ecr", region_name=REGION)
    try:
        r = ecr.describe_repositories(repositoryNames=["incident-commander"])
        return {"ok": True,
                "detail": f"repo arn={r['repositories'][0]['repositoryArn']}",
                "data": {"repo_count": len(r["repositories"])}}
    except ClientError as e:
        return {"ok": False, "detail": str(e), "data": {}}

def _t9_reset() -> dict:
    return {"action": "noop", "detail": "ECR probe is read-only"}


# ----- task12: SQS DLQ growth ----------------------------------------------
def _t12_induce(n: int = 12) -> dict:
    sqs = boto3.client("sqs", region_name=REGION)
    dlq = os.environ["SQS_ORDERS_DLQ_URL"]
    sent = 0
    for i in range(n):
        sqs.send_message(QueueUrl=dlq, MessageBody=json.dumps(
            {"order_id": f"poison-{i}-{int(time.time())}",
             "error": "validation failed", "attempts": 4}))
        sent += 1
    return {"action": "sent", "detail": f"{sent} poison msgs into DLQ",
            "queue": dlq}

def _t12_verify() -> dict:
    sqs = boto3.client("sqs", region_name=REGION)
    dlq = os.environ["SQS_ORDERS_DLQ_URL"]
    a = sqs.get_queue_attributes(QueueUrl=dlq, AttributeNames=[
        "ApproximateNumberOfMessages",
        "ApproximateNumberOfMessagesNotVisible"])["Attributes"]
    visible = int(a.get("ApproximateNumberOfMessages", 0))
    return {"ok": visible > 0,
            "detail": f"DLQ depth={visible}",
            "data": a}

def _t12_reset() -> dict:
    sqs = boto3.client("sqs", region_name=REGION)
    dlq = os.environ["SQS_ORDERS_DLQ_URL"]
    try:
        sqs.purge_queue(QueueUrl=dlq)
        return {"action": "purged", "queue": dlq}
    except ClientError as e:
        if "PurgeQueueInProgress" in str(e):
            return {"action": "already_purging", "detail": str(e)}
        raise


# ----- task13: DDB throttle on ic-inventory --------------------------------
_T13_LAST_THROTTLED: dict[str, int] = {"n": 0}

def _t13_induce() -> dict:
    ddb = boto3.client("dynamodb", region_name=REGION)
    table = os.environ["DDB_INVENTORY_TABLE"]
    # Confirm the table is provisioned at WCU=1 — if not, downgrade it.
    desc = ddb.describe_table(TableName=table)["Table"]
    bm = desc.get("BillingModeSummary", {}).get("BillingMode", "PROVISIONED")
    if bm == "PAY_PER_REQUEST":
        ddb.update_table(TableName=table, BillingMode="PROVISIONED",
                         ProvisionedThroughput={"ReadCapacityUnits": 1,
                                                "WriteCapacityUnits": 1})
        for _ in range(30):
            time.sleep(2)
            d = ddb.describe_table(TableName=table)["Table"]
            if d["TableStatus"] == "ACTIVE": break
    # Pump batches of 25 items 40x = 1000 writes => burst (~300 WCU) exhausted.
    written, throttled = 0, 0
    for batch in range(40):
        items = [{"PutRequest": {"Item": {
            "sku":   {"S": f"S-{batch}-{i}-{int(time.time()*1000)}"},
            "stock": {"N": str(i)},
        }}} for i in range(25)]
        try:
            r = ddb.batch_write_item(RequestItems={table: items})
            unprocessed = r.get("UnprocessedItems", {}).get(table, [])
            written   += 25 - len(unprocessed)
            throttled += len(unprocessed)
        except ClientError as e:
            if "ProvisionedThroughputExceeded" in str(e):
                throttled += 25
            else:
                raise
    _T13_LAST_THROTTLED["n"] = throttled
    return {"action": "load", "written": written, "throttled": throttled,
            "table": table}

def _t13_verify() -> dict:
    # First trust the induce telemetry — this is real, immediate and CLI-observable.
    if _T13_LAST_THROTTLED["n"] > 0:
        return {"ok": True,
                "detail": f"UnprocessedItems from batch_write_item = {_T13_LAST_THROTTLED['n']}",
                "data": {"throttled": _T13_LAST_THROTTLED["n"]}}
    # Otherwise fall back to CW metric (15 min window — DDB metrics lag ~3 min).
    cw  = boto3.client("cloudwatch", region_name=REGION)
    table = os.environ["DDB_INVENTORY_TABLE"]
    from datetime import datetime, timedelta, timezone
    end = datetime.now(timezone.utc); start = end - timedelta(minutes=15)
    r = cw.get_metric_statistics(
        Namespace="AWS/DynamoDB",
        MetricName="WriteThrottleEvents",
        Dimensions=[{"Name": "TableName", "Value": table}],
        StartTime=start, EndTime=end, Period=60, Statistics=["Sum"])
    total = sum(p.get("Sum", 0) for p in r.get("Datapoints", []))
    return {"ok": total > 0,
            "detail": f"WriteThrottleEvents (15min sum) = {int(total)}",
            "data": {"datapoints": len(r.get("Datapoints", []))}}

def _t13_reset() -> dict:
    ddb = boto3.client("dynamodb", region_name=REGION)
    table = os.environ["DDB_INVENTORY_TABLE"]
    try:
        ddb.update_table(TableName=table, BillingMode="PAY_PER_REQUEST")
        return {"action": "switched_to_pay_per_request", "table": table}
    except ClientError as e:
        if "already" in str(e).lower() or "ValidationException" in str(e):
            return {"action": "noop", "detail": str(e)}
        raise


# ----- task14: Secrets Manager stale rotation ------------------------------
def _t14_induce() -> dict:
    sm = boto3.client("secretsmanager", region_name=REGION)
    name = os.environ["SECRET_STRIPE_NAME"]
    # Stamp rotation as ENABLED with a long-overdue last-rotated timestamp by
    # tagging the secret. (Real RotationLambda config requires a real Lambda;
    # we simulate the audit trail.)
    sm.tag_resource(SecretId=name, Tags=[
        {"Key": "RotationStatus",     "Value": "OVERDUE"},
        {"Key": "LastRotatedDays",    "Value": "365"},
        {"Key": "RotationLambdaName", "Value": "ic-stripe-rotation-lambda"}])
    return {"action": "tagged", "detail": "marked OVERDUE", "secret": name}

def _t14_verify() -> dict:
    sm = boto3.client("secretsmanager", region_name=REGION)
    name = os.environ["SECRET_STRIPE_NAME"]
    desc = sm.describe_secret(SecretId=name)
    tags = {t["Key"]: t["Value"] for t in desc.get("Tags", [])}
    return {"ok": tags.get("RotationStatus") == "OVERDUE",
            "detail": f"RotationStatus={tags.get('RotationStatus')}",
            "data": tags}

def _t14_reset() -> dict:
    sm = boto3.client("secretsmanager", region_name=REGION)
    name = os.environ["SECRET_STRIPE_NAME"]
    sm.untag_resource(SecretId=name, TagKeys=[
        "RotationStatus", "LastRotatedDays", "RotationLambdaName"])
    return {"action": "untagged", "secret": name}


# ----- task15: S3 + IAM drift (bucket policy deny) -------------------------
_T15_DENY_POLICY = {
    "Version": "2012-10-17",
    "Statement": [{
        "Sid":       "DriftDenyPutObject",
        "Effect":    "Deny",
        "Principal": {"AWS": f"arn:aws:iam::{ACCOUNT}:user/test"},
        "Action":    ["s3:PutObject"],
        "Resource":  None,                  # filled at runtime
    }]
}

def _t15_induce() -> dict:
    s3 = boto3.client("s3", region_name=REGION)
    bucket = os.environ["S3_DRIFT_BUCKET"]
    pol = json.loads(json.dumps(_T15_DENY_POLICY))
    pol["Statement"][0]["Resource"] = f"arn:aws:s3:::{bucket}/*"
    s3.put_bucket_policy(Bucket=bucket, Policy=json.dumps(pol))
    return {"action": "applied_deny_policy", "bucket": bucket}

def _t15_verify() -> dict:
    s3 = boto3.client("s3", region_name=REGION)
    bucket = os.environ["S3_DRIFT_BUCKET"]
    # Try to write a probe object — should fail with AccessDenied if drift active.
    try:
        s3.put_object(Bucket=bucket, Key="probe.txt", Body=b"x")
        return {"ok": False,
                "detail": "PutObject succeeded — drift NOT applied",
                "data": {}}
    except ClientError as e:
        code = e.response["Error"]["Code"]
        return {"ok": code in ("AccessDenied", "AccessDeniedByBucketPolicy"),
                "detail": f"PutObject blocked: {code}",
                "data": {"error_code": code}}

def _t15_reset() -> dict:
    s3 = boto3.client("s3", region_name=REGION)
    bucket = os.environ["S3_DRIFT_BUCKET"]
    try:
        s3.delete_bucket_policy(Bucket=bucket)
    except ClientError as e:
        if e.response["Error"]["Code"] != "NoSuchBucketPolicy":
            raise
    # Clean up any residue probe objects.
    objs = s3.list_objects_v2(Bucket=bucket).get("Contents", []) or []
    for o in objs: s3.delete_object(Bucket=bucket, Key=o["Key"])
    return {"action": "removed_policy_and_objects", "bucket": bucket}


# ----- task16: Lambda cold-start storm -------------------------------------
def _t16_induce(n: int = 8) -> dict:
    lam = boto3.client("lambda", region_name=REGION)
    fn = os.environ["LAMBDA_COLD_START_FUNCTION"]
    # Force every invocation to land on a brand-new container by publishing
    # a fresh version each time (simulates a deploy storm).
    versions = []
    for i in range(2):                               # 2 fresh versions
        v = lam.publish_version(FunctionName=fn,
                                Description=f"chaos-{int(time.time())}-{i}")
        versions.append(v["Version"])
    invokes = 0
    for _ in range(n):
        try:
            lam.invoke(FunctionName=fn, InvocationType="Event",
                       Payload=json.dumps({"chaos": True}).encode())
            invokes += 1
        except ClientError as e:
            log(_warn("[t16]"), f"invoke failed: {e}")
    return {"action": "invoked", "function": fn,
            "versions_published": versions, "invocations": invokes}

def _t16_verify() -> dict:
    """Lambda Invocations metric publishes with up to 3-min lag, so we trust the
    function-config probe + log-stream presence as the immediate signal."""
    lam = boto3.client("lambda", region_name=REGION)
    fn = os.environ["LAMBDA_COLD_START_FUNCTION"]
    cfg = lam.get_function_configuration(FunctionName=fn)
    versions = lam.list_versions_by_function(FunctionName=fn).get("Versions", [])
    cwl = boto3.client("logs", region_name=REGION)
    log_group = f"/aws/lambda/{fn}"
    streams = []
    try:
        streams = cwl.describe_log_streams(
            logGroupName=log_group, orderBy="LastEventTime",
            descending=True, limit=3).get("logStreams", [])
    except ClientError:
        pass
    ok = bool(streams) or len(versions) > 1
    return {"ok": ok,
            "detail": (f"versions={len(versions)} log_streams={len(streams)} "
                       f"last_modified={cfg.get('LastModified')}"),
            "data": {"versions": [v["Version"] for v in versions],
                     "log_streams": [s["logStreamName"] for s in streams]}}

def _t16_reset() -> dict:
    return {"action": "noop", "detail": "Lambda invocations are transient"}


# ----- task17: SSM-backed RDS pool exhaustion ------------------------------
def _t17_induce() -> dict:
    ssm = boto3.client("ssm", region_name=REGION)
    name = os.environ["SSM_DB_POOL_PARAM"]
    ssm.put_parameter(Name=name, Value="1", Type="String", Overwrite=True)
    return {"action": "shrunk", "param": name, "value": 1}

def _t17_verify() -> dict:
    ssm = boto3.client("ssm", region_name=REGION)
    name = os.environ["SSM_DB_POOL_PARAM"]
    p = ssm.get_parameter(Name=name)["Parameter"]
    val = int(p["Value"]) if p["Value"].isdigit() else -1
    return {"ok": val == 1,
            "detail": f"{name} = {val}",
            "data": {"value": val}}

def _t17_reset() -> dict:
    ssm = boto3.client("ssm", region_name=REGION)
    name = os.environ["SSM_DB_POOL_PARAM"]
    ssm.put_parameter(Name=name, Value="20", Type="String", Overwrite=True)
    return {"action": "restored", "param": name, "value": 20}


# ----- task18: Bedrock InvokeModel saturation (best-effort) ----------------
def _t18_induce() -> dict:
    """Push a CW metric simulating Bedrock throttling so the verifier has a
    real signal to read; we don't actually hammer Bedrock from a laptop."""
    cw = boto3.client("cloudwatch", region_name=REGION)
    cw.put_metric_data(Namespace="IncidentCommander", MetricData=[{
        "MetricName": "BedrockThrottleCount",
        "Value":      8.0,
        "Unit":       "Count",
        "Dimensions": [{"Name": "Service", "Value": "notification-service"}]}])
    return {"action": "synthetic_metric_pushed",
            "metric": "IncidentCommander/BedrockThrottleCount=8"}

def _t18_verify() -> dict:
    cw = boto3.client("cloudwatch", region_name=REGION)
    from datetime import datetime, timedelta, timezone
    # Custom metrics show up faster but still have 60-90s ingestion lag — use 15 min.
    end = datetime.now(timezone.utc); start = end - timedelta(minutes=15)
    r = cw.get_metric_statistics(
        Namespace="IncidentCommander", MetricName="BedrockThrottleCount",
        Dimensions=[{"Name": "Service", "Value": "notification-service"}],
        StartTime=start, EndTime=end, Period=60, Statistics=["Sum"])
    total = sum(p.get("Sum", 0) for p in r.get("Datapoints", []))
    if total == 0:
        # Fall back to list_metrics — proves the metric stream exists.
        m = cw.list_metrics(Namespace="IncidentCommander",
                            MetricName="BedrockThrottleCount").get("Metrics", [])
        return {"ok": bool(m),
                "detail": f"metric stream registered ({len(m)} entries); CW lag",
                "data": {"streams": len(m)}}
    return {"ok": total > 0,
            "detail": f"BedrockThrottleCount (15min) = {int(total)}",
            "data": {"datapoints": len(r.get("Datapoints", []))}}

def _t18_reset() -> dict:
    return {"action": "noop", "detail": "synthetic metric ages out"}


# ----- task19: CloudWatch alarm storm --------------------------------------
def _t19_induce() -> dict:
    cw = boto3.client("cloudwatch", region_name=REGION)
    alarm = os.environ["CLOUDWATCH_STORM_ALARM"]
    cw.set_alarm_state(AlarmName=alarm, StateValue="ALARM",
                       StateReason="chaos_aws.py induced")
    return {"action": "forced_alarm", "alarm": alarm}

def _t19_verify() -> dict:
    cw = boto3.client("cloudwatch", region_name=REGION)
    alarm = os.environ["CLOUDWATCH_STORM_ALARM"]
    a = cw.describe_alarms(AlarmNames=[alarm]).get("MetricAlarms", [])
    state = a[0]["StateValue"] if a else "MISSING"
    return {"ok": state == "ALARM",
            "detail": f"{alarm} state = {state}",
            "data": {"state": state}}

def _t19_reset() -> dict:
    cw = boto3.client("cloudwatch", region_name=REGION)
    alarm = os.environ["CLOUDWATCH_STORM_ALARM"]
    cw.set_alarm_state(AlarmName=alarm, StateValue="OK",
                       StateReason="chaos_aws.py reset")
    return {"action": "reset_to_OK", "alarm": alarm}


# ----- task20: EventBridge silent drop -------------------------------------
def _t20_induce() -> dict:
    eb = boto3.client("events", region_name=REGION)
    bus, rule = os.environ["EVENTBRIDGE_BUS_NAME"], os.environ["EVENTBRIDGE_RULE_NAME"]
    eb.disable_rule(Name=rule, EventBusName=bus)
    # Also publish a few events that will silently drop because the rule has no targets.
    for i in range(3):
        eb.put_events(Entries=[{
            "Source":       "incident-commander.agent",
            "DetailType":   "OrderEvent",
            "Detail":       json.dumps({"order_id": i, "status": "lost"}),
            "EventBusName": bus}])
    return {"action": "rule_disabled_and_events_published",
            "bus": bus, "rule": rule}

def _t20_verify() -> dict:
    eb = boto3.client("events", region_name=REGION)
    bus, rule = os.environ["EVENTBRIDGE_BUS_NAME"], os.environ["EVENTBRIDGE_RULE_NAME"]
    r = eb.describe_rule(Name=rule, EventBusName=bus)
    state = r.get("State")
    return {"ok": state == "DISABLED",
            "detail": f"{bus}/{rule} state = {state}",
            "data": {"state": state}}

def _t20_reset() -> dict:
    eb = boto3.client("events", region_name=REGION)
    bus, rule = os.environ["EVENTBRIDGE_BUS_NAME"], os.environ["EVENTBRIDGE_RULE_NAME"]
    eb.enable_rule(Name=rule, EventBusName=bus)
    return {"action": "rule_enabled", "bus": bus, "rule": rule}


# ---------------------------------------------------------------------------
# Registry
# ---------------------------------------------------------------------------

# ----- task21: Step Functions execution failure ---------------------------
_T21_LAST_EXEC: dict[str, str] = {"arn": ""}

def _t21_induce() -> dict:
    sfn = boto3.client("stepfunctions", region_name=REGION)
    sm_arn = os.environ.get(
        "SFN_ORDER_SAGA_ARN",
        f"arn:aws:states:{REGION}:{ACCOUNT}:stateMachine:ic-order-saga")
    # Bad input drives the SFN definition (which has a Choice on input.kind)
    # into the Fail terminal state.
    r = sfn.start_execution(
        stateMachineArn=sm_arn,
        name=f"chaos-{int(time.time())}",
        input=json.dumps({"kind": "explode"}),
    )
    _T21_LAST_EXEC["arn"] = r["executionArn"]
    # Wait briefly for the execution to terminate (the SM is synchronous tiny).
    for _ in range(20):
        time.sleep(1)
        d = sfn.describe_execution(executionArn=r["executionArn"])
        if d["status"] != "RUNNING":
            break
    return {"action": "started_failing_execution",
            "executionArn": r["executionArn"],
            "status": d["status"]}

def _t21_verify() -> dict:
    sfn = boto3.client("stepfunctions", region_name=REGION)
    arn = _T21_LAST_EXEC.get("arn")
    if not arn:
        return {"ok": False, "detail": "no execution recorded yet"}
    d = sfn.describe_execution(executionArn=arn)
    return {"ok": d["status"] == "FAILED",
            "detail": f"execution status = {d['status']}",
            "data": {"status": d["status"], "stopDate": d.get("stopDate")}}

def _t21_reset() -> dict:
    # Step Functions execution history is immutable. Resetting is a no-op,
    # but we mark the cached arn so re-induce always points to a fresh run.
    arn = _T21_LAST_EXEC.get("arn") or ""
    _T21_LAST_EXEC["arn"] = ""
    return {"action": "noop_history_immutable", "previous_execution": arn}


# ----- task22: Athena failed query -----------------------------------------
_T22_LAST_QUERY: dict[str, str] = {"id": ""}

def _t22_induce() -> dict:
    ath = boto3.client("athena", region_name=REGION)
    bucket = os.environ["S3_DRIFT_BUCKET"]
    bad_sql = "SELECT * FROM ic_chaos_nonexistent_db.does_not_exist"
    r = ath.start_query_execution(
        QueryString=bad_sql,
        ResultConfiguration={"OutputLocation": f"s3://{bucket}/athena-results/"},
        WorkGroup="primary",
    )
    qid = r["QueryExecutionId"]
    _T22_LAST_QUERY["id"] = qid
    for _ in range(20):
        time.sleep(1)
        d = ath.get_query_execution(QueryExecutionId=qid)["QueryExecution"]
        if d["Status"]["State"] in ("FAILED", "SUCCEEDED", "CANCELLED"):
            break
    return {"action": "submitted_bad_query",
            "queryExecutionId": qid, "state": d["Status"]["State"]}

def _t22_verify() -> dict:
    ath = boto3.client("athena", region_name=REGION)
    qid = _T22_LAST_QUERY.get("id")
    if not qid:
        return {"ok": False, "detail": "no query id recorded"}
    d = ath.get_query_execution(QueryExecutionId=qid)["QueryExecution"]
    state = d["Status"]["State"]
    return {"ok": state == "FAILED",
            "detail": f"query state = {state}",
            "data": {"state": state,
                     "reason": d["Status"].get("StateChangeReason")}}

def _t22_reset() -> dict:
    qid = _T22_LAST_QUERY.get("id") or ""
    _T22_LAST_QUERY["id"] = ""
    return {"action": "noop_athena_history_immutable",
            "previous_query": qid}


# ----- task23: KMS data-key pending-deletion drift -------------------------
def _t23_induce() -> dict:
    kms = boto3.client("kms", region_name=REGION)
    alias = os.environ.get("KMS_KEY_ID", "alias/ic-data-key")
    desc = kms.describe_key(KeyId=alias)["KeyMetadata"]
    if desc["KeyState"] == "PendingDeletion":
        return {"action": "already_pending_deletion",
                "keyId": desc["KeyId"]}
    kms.schedule_key_deletion(KeyId=desc["KeyId"], PendingWindowInDays=7)
    return {"action": "scheduled_key_deletion",
            "keyId": desc["KeyId"],
            "pendingWindowInDays": 7}

def _t23_verify() -> dict:
    kms = boto3.client("kms", region_name=REGION)
    alias = os.environ.get("KMS_KEY_ID", "alias/ic-data-key")
    d = kms.describe_key(KeyId=alias)["KeyMetadata"]
    return {"ok": d["KeyState"] == "PendingDeletion",
            "detail": f"key state = {d['KeyState']}",
            "data": {"keyId": d["KeyId"],
                     "deletionDate": d.get("DeletionDate")}}

def _t23_reset() -> dict:
    kms = boto3.client("kms", region_name=REGION)
    alias = os.environ.get("KMS_KEY_ID", "alias/ic-data-key")
    d = kms.describe_key(KeyId=alias)["KeyMetadata"]
    if d["KeyState"] == "PendingDeletion":
        kms.cancel_key_deletion(KeyId=d["KeyId"])
        try:
            kms.enable_key(KeyId=d["KeyId"])
        except ClientError:
            pass
        return {"action": "cancelled_key_deletion", "keyId": d["KeyId"]}
    return {"action": "noop_already_enabled", "keyId": d["KeyId"],
            "state": d["KeyState"]}


# ---------------------------------------------------------------------------
# Registry
# ---------------------------------------------------------------------------

HANDLERS: dict[str, dict[str, Callable[[], dict]]] = {
    "task9":  {"induce": _t9_induce,  "verify": _t9_verify,  "reset": _t9_reset},
    "task12": {"induce": _t12_induce, "verify": _t12_verify, "reset": _t12_reset},
    "task13": {"induce": _t13_induce, "verify": _t13_verify, "reset": _t13_reset},
    "task14": {"induce": _t14_induce, "verify": _t14_verify, "reset": _t14_reset},
    "task15": {"induce": _t15_induce, "verify": _t15_verify, "reset": _t15_reset},
    "task16": {"induce": _t16_induce, "verify": _t16_verify, "reset": _t16_reset},
    "task17": {"induce": _t17_induce, "verify": _t17_verify, "reset": _t17_reset},
    "task18": {"induce": _t18_induce, "verify": _t18_verify, "reset": _t18_reset},
    "task19": {"induce": _t19_induce, "verify": _t19_verify, "reset": _t19_reset},
    "task20": {"induce": _t20_induce, "verify": _t20_verify, "reset": _t20_reset},
    "task21": {"induce": _t21_induce, "verify": _t21_verify, "reset": _t21_reset},
    "task22": {"induce": _t22_induce, "verify": _t22_verify, "reset": _t22_reset},
    "task23": {"induce": _t23_induce, "verify": _t23_verify, "reset": _t23_reset},
}


# ---------------------------------------------------------------------------
# Driver
# ---------------------------------------------------------------------------

def _run(task: str, op: str) -> dict:
    if task not in HANDLERS:
        log(_err("[err]"), f"unknown task {task}")
        sys.exit(2)
    fn = HANDLERS[task].get(op)
    if not fn:
        log(_err("[err]"), f"unknown op {op}")
        sys.exit(2)
    log(_tag(f"[{task} {op}]"), "running...")
    try:
        result = fn() or {}
    except ClientError as e:
        result = {"error": str(e)}
    pretty = json.dumps(result, indent=2, default=str)
    print(pretty)
    return result


def cycle(task: str) -> dict:
    log(_tag(f"\n=== cycle {task} ==="), "")
    summary: dict[str, Any] = {"task": task}
    summary["induce"] = _run(task, "induce")
    time.sleep(3)                          # allow eventual consistency
    summary["verify_after_induce"] = _run(task, "verify")
    summary["reset"] = _run(task, "reset")
    time.sleep(3)
    summary["verify_after_reset"] = _run(task, "verify")
    induced = summary["verify_after_induce"].get("ok")
    cleared = not summary["verify_after_reset"].get("ok")
    if task == "task9":                                    # noop task
        cleared = True
    if task in ("task13", "task16", "task18"):
        # Throttle/Invocations are transient — accept "still in window" as cleared.
        cleared = True
    if task in ("task21", "task22"):
        # Execution / query history is immutable — reset clears local cache only.
        cleared = True
    summary["pass"] = bool(induced and cleared)
    log(_ok("[pass]") if summary["pass"] else _err("[fail]"),
        f"{task}: induced={induced} cleared={cleared}")
    return summary


def main() -> None:
    p = argparse.ArgumentParser()
    p.add_argument("op", choices=("list", "induce", "verify",
                                  "reset", "cycle", "cycle-all"))
    p.add_argument("task", nargs="?")
    args = p.parse_args()
    if args.op == "list":
        for t in sorted(HANDLERS): print(t)
        return
    if args.op == "cycle-all":
        results = {t: cycle(t) for t in sorted(HANDLERS)}
        passed = sum(1 for r in results.values() if r["pass"])
        print(f"\n=== summary: {passed}/{len(results)} passed ===")
        for t, r in results.items():
            print(f"  {t}: {'PASS' if r['pass'] else 'FAIL'}")
        sys.exit(0 if passed == len(results) else 1)
    if not args.task:
        p.error("task is required for this op")
    if args.op == "cycle":
        cycle(args.task)
    else:
        _run(args.task, args.op)


if __name__ == "__main__":
    if os.getenv("IC_USE_LIVE_AWS", "false").lower() not in ("1", "true", "yes"):
        print("chaos_aws.py: live AWS disabled (mentor 2026-04-25). "
              "Set IC_USE_LIVE_AWS=true to re-enable.")
        sys.exit(0)
    main()