scripts/aws_killswitch.py

"""Targeted AWS kill-switch.

Deletes every billable resource that this project provisioned in the AWS
account. Scope is intentionally NARROW:
  * Anything tagged Project=incident-commander
  * Plus the explicit `ic-*` / `incident-commander-*` resource names from
    scripts/aws_inventory.py
  * Plus the EKS cluster `incident-commander` and its node groups

Will NOT touch anything else in the account. Always run --dry-run first,
review the planned actions, then re-run with --apply.

Usage:
    python scripts/aws_killswitch.py                 # dry-run
    python scripts/aws_killswitch.py --apply         # actually delete

Run from incident-commander/ project root (so .env.aws.local is found).
"""

from __future__ import annotations

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

# ---------------------------------------------------------------------------
# Bootstrap: load .env.aws.local for resource pointers + region.
# ---------------------------------------------------------------------------

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("'"))

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

REGION = os.environ.get("AWS_REGION", "us-east-1")
CFG    = Config(connect_timeout=5, read_timeout=20, retries={"max_attempts": 2})
ACCOUNT = boto3.client("sts", region_name=REGION, config=CFG)\
    .get_caller_identity()["Account"]

# Names we own. Anything not in this list (or matching the patterns below)
# is left strictly alone.
OWNED_LAMBDAS = [
    "ic-cold-start-target",
    "ic-runbook-restart-pods",
    "incident-commander-alarm-hook",
]
OWNED_SQS_QUEUES = ["ic-events", "ic-orders", "ic-orders-dlq"]
OWNED_DDB_TABLES = ["ic-inventory", "incident-commander-curriculum"]
OWNED_SECRETS    = ["incident-commander/openai-api-key", "ic-stripe-api-key"]
OWNED_SSM_PARAMS = ["/ic/payments/db-pool-size", "/ic/feature-flags/enable_chaos"]
OWNED_SNS_TOPICS = [f"arn:aws:sns:{REGION}:{ACCOUNT}:incident-commander-alerts"]
OWNED_EB_BUS     = "ic-bus"
OWNED_EB_RULE    = "ic-orders-rule"
OWNED_KMS_ALIAS  = "alias/ic-data-key"
OWNED_SFN_NAME   = "ic-order-saga"
OWNED_ALARMS     = ["ic-cw-alarm-storm"]
OWNED_LOG_GROUP_PREFIXES = [
    "/aws/eks/incident-commander/",
    "/aws/lambda/ic-",
    "/aws/lambda/incident-commander-",
]
OWNED_BUCKET_PREFIXES = ["ic-checkpoints-", "ic-task15-drift-"]
OWNED_EKS_CLUSTER = "incident-commander"
OWNED_ECR_REPO    = "incident-commander"
OWNED_IAM_ROLE_PREFIXES = ["ic-", "incident-commander-"]


# ---------------------------------------------------------------------------
# Pretty printing
# ---------------------------------------------------------------------------

class _Plan:
    def __init__(self, dry: bool) -> None:
        self.dry = dry
        self.deleted: list[str] = []
        self.skipped: list[str] = []
        self.errored: list[tuple[str, str]] = []

    def will(self, what: str, do: Callable[[], Any]) -> None:
        prefix = "[DRY]" if self.dry else "[DEL]"
        print(f"  {prefix} {what}")
        if self.dry:
            self.deleted.append(what)
            return
        try:
            do()
            self.deleted.append(what)
        except ClientError as e:
            code = e.response.get("Error", {}).get("Code", "?")
            if code in ("ResourceNotFoundException", "NoSuchEntity",
                        "NoSuchBucket", "NotFoundException",
                        "AWS.SimpleQueueService.NonExistentQueue",
                        "AlarmNotFound", "ResourceNotFound"):
                print(f"     (already gone: {code})")
                self.skipped.append(what)
            else:
                print(f"     [error] {code}: {e}")
                self.errored.append((what, f"{code}: {e}"))
        except Exception as e:
            print(f"     [error] {e}")
            self.errored.append((what, str(e)))

    def skip(self, what: str, why: str = "not found") -> None:
        print(f"  [skip] {what} ({why})")
        self.skipped.append(what)


def hdr(s: str) -> None:
    print(f"\n=== {s} ===")


# ---------------------------------------------------------------------------
# Per-service teardown
# ---------------------------------------------------------------------------

def _kill_eks(plan: _Plan) -> None:
    hdr("EKS")
    eks = boto3.client("eks", region_name=REGION, config=CFG)
    try:
        eks.describe_cluster(name=OWNED_EKS_CLUSTER)
    except ClientError:
        plan.skip(f"eks/{OWNED_EKS_CLUSTER}")
        return
    # Node groups
    ngs = eks.list_nodegroups(clusterName=OWNED_EKS_CLUSTER).get("nodegroups", [])
    for ng in ngs:
        plan.will(f"eks nodegroup {OWNED_EKS_CLUSTER}/{ng}",
                  lambda ng=ng: eks.delete_nodegroup(
                      clusterName=OWNED_EKS_CLUSTER, nodegroupName=ng))
    if ngs and not plan.dry:
        # Wait for nodegroups to actually go (cluster delete fails otherwise)
        print("     waiting for nodegroups to drain...")
        for ng in ngs:
            for _ in range(60):     # up to 30 min
                try:
                    eks.describe_nodegroup(
                        clusterName=OWNED_EKS_CLUSTER, nodegroupName=ng)
                    time.sleep(30)
                except ClientError:
                    break
    # Fargate profiles (idempotent if none)
    try:
        fps = eks.list_fargate_profiles(clusterName=OWNED_EKS_CLUSTER)\
            .get("fargateProfileNames", [])
    except ClientError:
        fps = []
    for fp in fps:
        plan.will(f"eks fargate profile {OWNED_EKS_CLUSTER}/{fp}",
                  lambda fp=fp: eks.delete_fargate_profile(
                      clusterName=OWNED_EKS_CLUSTER, fargateProfileName=fp))
    # Cluster
    plan.will(f"eks cluster {OWNED_EKS_CLUSTER}",
              lambda: eks.delete_cluster(name=OWNED_EKS_CLUSTER))


def _kill_ec2(plan: _Plan) -> None:
    """Kill EC2 only if tagged Project=incident-commander.
    Won't touch anything else."""
    hdr("EC2 (project-tagged only)")
    ec2 = boto3.client("ec2", region_name=REGION, config=CFG)
    flt = [{"Name": "tag:Project", "Values": ["incident-commander"]}]
    # Instances
    instances = []
    for r in ec2.describe_instances(Filters=flt).get("Reservations", []):
        for i in r.get("Instances", []):
            if i.get("State", {}).get("Name") not in ("terminated", "shutting-down"):
                instances.append(i["InstanceId"])
    if instances:
        plan.will(f"ec2 terminate instances {instances}",
                  lambda: ec2.terminate_instances(InstanceIds=instances))
    else:
        plan.skip("ec2 instances (project tag)")
    # NAT gateways (project tagged)
    nats = ec2.describe_nat_gateways(Filter=flt).get("NatGateways", [])
    for n in nats:
        if n.get("State") not in ("deleted", "deleting"):
            plan.will(f"ec2 nat-gateway {n['NatGatewayId']}",
                      lambda nid=n["NatGatewayId"]:
                          ec2.delete_nat_gateway(NatGatewayId=nid))
    # Elastic IPs (project tagged)
    eips = ec2.describe_addresses(Filters=flt).get("Addresses", [])
    for a in eips:
        if "AllocationId" in a:
            plan.will(f"ec2 release-eip {a['AllocationId']}",
                      lambda aid=a["AllocationId"]:
                          ec2.release_address(AllocationId=aid))


def _kill_lambda(plan: _Plan) -> None:
    hdr("Lambda")
    lam = boto3.client("lambda", region_name=REGION, config=CFG)
    for fn in OWNED_LAMBDAS:
        plan.will(f"lambda function {fn}",
                  lambda fn=fn: lam.delete_function(FunctionName=fn))


def _kill_sqs(plan: _Plan) -> None:
    hdr("SQS")
    sqs = boto3.client("sqs", region_name=REGION, config=CFG)
    for q in OWNED_SQS_QUEUES:
        try:
            url = sqs.get_queue_url(QueueName=q)["QueueUrl"]
        except ClientError:
            plan.skip(f"sqs/{q}")
            continue
        plan.will(f"sqs queue {q}",
                  lambda url=url: sqs.delete_queue(QueueUrl=url))


def _kill_ddb(plan: _Plan) -> None:
    hdr("DynamoDB")
    ddb = boto3.client("dynamodb", region_name=REGION, config=CFG)
    for t in OWNED_DDB_TABLES:
        plan.will(f"ddb table {t}",
                  lambda t=t: ddb.delete_table(TableName=t))


def _empty_and_delete_bucket(s3, bucket: str) -> None:
    # Delete every object version + delete-marker, then the bucket.
    paginator = s3.get_paginator("list_object_versions")
    for page in paginator.paginate(Bucket=bucket):
        delete_objs = []
        for v in page.get("Versions", []) or []:
            delete_objs.append({"Key": v["Key"], "VersionId": v["VersionId"]})
        for m in page.get("DeleteMarkers", []) or []:
            delete_objs.append({"Key": m["Key"], "VersionId": m["VersionId"]})
        if delete_objs:
            for i in range(0, len(delete_objs), 1000):
                s3.delete_objects(Bucket=bucket,
                                  Delete={"Objects": delete_objs[i:i+1000],
                                          "Quiet": True})
    s3.delete_bucket(Bucket=bucket)


def _kill_s3(plan: _Plan) -> None:
    hdr("S3 buckets")
    s3 = boto3.client("s3", region_name=REGION, config=CFG)
    buckets = [b["Name"] for b in s3.list_buckets().get("Buckets", [])]
    matches = [b for b in buckets
               if any(b.startswith(p) for p in OWNED_BUCKET_PREFIXES)]
    if not matches:
        plan.skip("s3 (no ic-* buckets)")
        return
    for b in matches:
        plan.will(f"s3 bucket {b} (empty + delete)",
                  lambda b=b: _empty_and_delete_bucket(s3, b))


def _kill_secrets(plan: _Plan) -> None:
    hdr("Secrets Manager")
    sm = boto3.client("secretsmanager", region_name=REGION, config=CFG)
    for s in OWNED_SECRETS:
        plan.will(f"secret {s}",
                  lambda s=s: sm.delete_secret(SecretId=s,
                                               ForceDeleteWithoutRecovery=True))


def _kill_kms(plan: _Plan) -> None:
    hdr("KMS")
    kms = boto3.client("kms", region_name=REGION, config=CFG)
    try:
        meta = kms.describe_key(KeyId=OWNED_KMS_ALIAS)["KeyMetadata"]
        key_id = meta["KeyId"]
    except ClientError:
        plan.skip(f"kms {OWNED_KMS_ALIAS}")
        return
    plan.will(f"kms alias {OWNED_KMS_ALIAS}",
              lambda: kms.delete_alias(AliasName=OWNED_KMS_ALIAS))
    if meta.get("KeyState") != "PendingDeletion":
        plan.will(f"kms schedule_key_deletion {key_id} (7d window)",
                  lambda: kms.schedule_key_deletion(
                      KeyId=key_id, PendingWindowInDays=7))


def _kill_eventbridge(plan: _Plan) -> None:
    hdr("EventBridge")
    eb = boto3.client("events", region_name=REGION, config=CFG)
    # Targets first
    try:
        ts = eb.list_targets_by_rule(Rule=OWNED_EB_RULE,
                                     EventBusName=OWNED_EB_BUS)\
            .get("Targets", [])
        if ts:
            plan.will(f"eb rule targets {OWNED_EB_RULE}",
                      lambda: eb.remove_targets(
                          Rule=OWNED_EB_RULE, EventBusName=OWNED_EB_BUS,
                          Ids=[t["Id"] for t in ts], Force=True))
    except ClientError:
        pass
    plan.will(f"eb rule {OWNED_EB_RULE}",
              lambda: eb.delete_rule(Name=OWNED_EB_RULE,
                                     EventBusName=OWNED_EB_BUS, Force=True))
    plan.will(f"eb bus {OWNED_EB_BUS}",
              lambda: eb.delete_event_bus(Name=OWNED_EB_BUS))


def _kill_cloudwatch(plan: _Plan) -> None:
    hdr("CloudWatch")
    cw  = boto3.client("cloudwatch", region_name=REGION, config=CFG)
    cwl = boto3.client("logs",       region_name=REGION, config=CFG)
    for a in OWNED_ALARMS:
        plan.will(f"cw alarm {a}",
                  lambda a=a: cw.delete_alarms(AlarmNames=[a]))
    # Log groups by prefix
    for prefix in OWNED_LOG_GROUP_PREFIXES:
        try:
            paginator = cwl.get_paginator("describe_log_groups")
            for page in paginator.paginate(logGroupNamePrefix=prefix):
                for lg in page.get("logGroups", []):
                    name = lg["logGroupName"]
                    plan.will(f"cw log group {name}",
                              lambda n=name: cwl.delete_log_group(logGroupName=n))
        except ClientError:
            pass


def _kill_sns(plan: _Plan) -> None:
    hdr("SNS")
    sns = boto3.client("sns", region_name=REGION, config=CFG)
    for arn in OWNED_SNS_TOPICS:
        plan.will(f"sns topic {arn}",
                  lambda arn=arn: sns.delete_topic(TopicArn=arn))


def _kill_ssm(plan: _Plan) -> None:
    hdr("SSM Parameter Store")
    ssm = boto3.client("ssm", region_name=REGION, config=CFG)
    for p in OWNED_SSM_PARAMS:
        plan.will(f"ssm param {p}",
                  lambda p=p: ssm.delete_parameter(Name=p))


def _kill_sfn(plan: _Plan) -> None:
    hdr("Step Functions")
    sfn = boto3.client("stepfunctions", region_name=REGION, config=CFG)
    arn = f"arn:aws:states:{REGION}:{ACCOUNT}:stateMachine:{OWNED_SFN_NAME}"
    plan.will(f"sfn state machine {OWNED_SFN_NAME}",
              lambda: sfn.delete_state_machine(stateMachineArn=arn))


def _kill_iam(plan: _Plan) -> None:
    hdr("IAM (project-prefixed roles only)")
    iam = boto3.client("iam", region_name=REGION, config=CFG)
    paginator = iam.get_paginator("list_roles")
    targets: list[str] = []
    for page in paginator.paginate():
        for r in page.get("Roles", []):
            n = r["RoleName"]
            if any(n.startswith(p) for p in OWNED_IAM_ROLE_PREFIXES):
                targets.append(n)
    for name in targets:
        # Detach managed policies
        try:
            for pol in iam.list_attached_role_policies(RoleName=name)\
                    .get("AttachedPolicies", []):
                plan.will(
                    f"iam detach {pol['PolicyArn']} from {name}",
                    lambda n=name, arn=pol["PolicyArn"]:
                        iam.detach_role_policy(RoleName=n, PolicyArn=arn))
        except ClientError:
            pass
        # Delete inline policies
        try:
            for pn in iam.list_role_policies(RoleName=name)\
                    .get("PolicyNames", []):
                plan.will(
                    f"iam inline-policy {name}/{pn}",
                    lambda n=name, p=pn:
                        iam.delete_role_policy(RoleName=n, PolicyName=p))
        except ClientError:
            pass
        # Remove from instance profiles
        try:
            for ipf in iam.list_instance_profiles_for_role(RoleName=name)\
                    .get("InstanceProfiles", []):
                plan.will(
                    f"iam remove role {name} from instance-profile {ipf['InstanceProfileName']}",
                    lambda n=name, ip=ipf["InstanceProfileName"]:
                        iam.remove_role_from_instance_profile(
                            InstanceProfileName=ip, RoleName=n))
        except ClientError:
            pass
        plan.will(f"iam role {name}",
                  lambda n=name: iam.delete_role(RoleName=n))


def _kill_ecr(plan: _Plan) -> None:
    hdr("ECR")
    ecr = boto3.client("ecr", region_name=REGION, config=CFG)
    plan.will(f"ecr repo {OWNED_ECR_REPO}",
              lambda: ecr.delete_repository(repositoryName=OWNED_ECR_REPO,
                                            force=True))


# ---------------------------------------------------------------------------
# Entrypoint
# ---------------------------------------------------------------------------

PHASES: list[tuple[str, Callable[[_Plan], None]]] = [
    # Order matters: dependents first.
    ("eks",          _kill_eks),
    ("lambda",       _kill_lambda),
    ("eventbridge",  _kill_eventbridge),
    ("sfn",          _kill_sfn),
    ("ec2",          _kill_ec2),
    ("sqs",          _kill_sqs),
    ("ddb",          _kill_ddb),
    ("s3",           _kill_s3),
    ("secrets",      _kill_secrets),
    ("kms",          _kill_kms),
    ("cloudwatch",   _kill_cloudwatch),
    ("sns",          _kill_sns),
    ("ssm",          _kill_ssm),
    ("ecr",          _kill_ecr),
    ("iam",          _kill_iam),
]


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--apply", action="store_true",
                    help="Actually delete (default is dry-run)")
    ap.add_argument("--only", action="append", default=[],
                    help="Only run these phases (repeatable)")
    args = ap.parse_args()
    plan = _Plan(dry=not args.apply)
    print(f"\nAccount: {ACCOUNT}   Region: {REGION}   Mode: "
          f"{'APPLY (will delete)' if args.apply else 'DRY-RUN'}\n")
    for name, fn in PHASES:
        if args.only and name not in args.only:
            continue
        try:
            fn(plan)
        except Exception as e:
            print(f"  [phase-error] {name}: {e}")
            plan.errored.append((name, str(e)))
    print(f"\n=== summary ===")
    print(f"  planned/deleted: {len(plan.deleted)}")
    print(f"  skipped (gone):  {len(plan.skipped)}")
    print(f"  errored:         {len(plan.errored)}")
    for what, why in plan.errored:
        print(f"   - {what}  =>  {why}")
    return 0 if not plan.errored else 1


if __name__ == "__main__":
    sys.exit(main())