Chaos Kitty(ミニ！)を 作って、カオスエンジニアリングを体験してみた

IAMポリシー　(クリックでコードを表示)

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "EC2AndSGControl",
            "Effect": "Allow",
            "Action": [
                "ec2:AuthorizeSecurityGroupIngress",
                "ec2:RevokeSecurityGroupIngress",
                "ec2:DescribeSecurityGroups",
                "ec2:StopInstances",
                "ec2:StartInstances",
                "ec2:DescribeInstances"
            ],
            "Resource": "*"
        },
        {
            "Sid": "S3PublicAccessControl",
            "Effect": "Allow",
            "Action": [
                "s3:PutBucketPublicAccessBlock",
                "s3:GetBucketPublicAccessBlock"
            ],
            "Resource": "arn:aws:s3:::*"
        },
        {
            "Sid": "CloudTrailControl",
            "Effect": "Allow",
            "Action": [
                "cloudtrail:StopLogging",
                "cloudtrail:StartLogging",
                "cloudtrail:GetTrailStatus"
            ],
            "Resource": "*"
        }
    ]
}

障害を発生させるLambda関数　Lambdaコード　(クリックでコードを表示)

import boto3
import random
import os

ec2 = boto3.client('ec2')
s3 = boto3.client('s3')
cloudtrail = boto3.client('cloudtrail')

def open_sg():
    sg_id = os.environ['SG_ID']

    ec2.authorize_security_group_ingress(
        GroupId=sg_id,
        IpPermissions=[{
            'IpProtocol': 'tcp',
            'FromPort': 22,
            'ToPort': 22,
            'IpRanges': [{'CidrIp': '0.0.0.0/0'}]
        }]
    )
    return "SG opened"

def make_s3_public():
    bucket = os.environ['BUCKET_NAME']

    s3.put_public_access_block(
        Bucket=bucket,
        PublicAccessBlockConfiguration={
            'BlockPublicAcls': False,
            'IgnorePublicAcls': False,
            'BlockPublicPolicy': False,
            'RestrictPublicBuckets': False
        }
    )
    return "S3 public"

def stop_ec2():
    instance_id = os.environ['INSTANCE_ID']

    ec2.stop_instances(InstanceIds=[instance_id])
    return "EC2 stopped"

def stop_cloudtrail():
    trail = os.environ['TRAIL_NAME']

    cloudtrail.stop_logging(Name=trail)
    return "CloudTrail stopped"

def lambda_handler(event, context):
    action = random.choice(["sg", "s3", "ec2", "ct"])

    if action == "sg":
        result = open_sg()
    elif action == "s3":
        result = make_s3_public()
    elif action == "ec2":
        result = stop_ec2()
    else:
        result = stop_cloudtrail()

    return {
        "action": action,
        "result": result
    }

障害を復旧させるLambda関数　Lambdaコード　(クリックでコードを表示)

import boto3
import os
from botocore.exceptions import ClientError

ec2 = boto3.client('ec2')
s3 = boto3.client('s3')
cloudtrail = boto3.client('cloudtrail')

def close_sg():
    try:
        ec2.revoke_security_group_ingress(
            GroupId=os.environ['SG_ID'],
            IpPermissions=[{
                'IpProtocol': 'tcp',
                'FromPort': 22,
                'ToPort': 22,
                'IpRanges': [{'CidrIp': '0.0.0.0/0'}]
            }]
        )
    except ClientError:
        pass

def make_s3_private():
    s3.put_public_access_block(
        Bucket=os.environ['BUCKET_NAME'],
        PublicAccessBlockConfiguration={
            'BlockPublicAcls': True,
            'IgnorePublicAcls': True,
            'BlockPublicPolicy': True,
            'RestrictPublicBuckets': True
        }
    )

def start_ec2():
    ec2.start_instances(InstanceIds=[os.environ['INSTANCE_ID']])

def start_cloudtrail():
    cloudtrail.start_logging(Name=os.environ['TRAIL_NAME'])

def lambda_handler(event, context):
    close_sg()
    make_s3_private()
    start_ec2()
    start_cloudtrail()

    return {"status": "recovered"}

Raspberry Piのコード　(クリックでコードを表示)

import RPi.GPIO as GPIO
import time
import subprocess
import json

# ===== 設定 =====
GREEN_LED = 27
RED_LED = 17
BUTTON = 18

CHAOS_LAMBDA = "作成した障害挿入用のLambda関数名"
RECOVER_LAMBDA = "作成した復旧用のLambda関数名"

BUCKET_NAME = "作成したS3バケット名"
SG_ID = "作成したSecurityGroupのID"
INSTANCE_ID = "作成したEC2のインスタンスID"
TRAIL_NAME = "作成したCloudTrailの証跡名"

LONG_PRESS_SEC = 2

GPIO.setmode(GPIO.BCM)
GPIO.setup(GREEN_LED, GPIO.OUT)
GPIO.setup(RED_LED, GPIO.OUT)
GPIO.setup(BUTTON, GPIO.IN, pull_up_down=GPIO.PUD_UP)

state = "normal"

# ===== LED制御 =====
def set_normal():
    GPIO.output(GREEN_LED, GPIO.HIGH)
    GPIO.output(RED_LED, GPIO.LOW)

def set_error():
    GPIO.output(GREEN_LED, GPIO.LOW)
    GPIO.output(RED_LED, GPIO.HIGH)

# ===== Lambda実行 =====
def invoke_lambda(name):
    try:
        print(f"{name} 実行")
        subprocess.run([
            "aws", "lambda", "invoke",
            "--function-name", name,
            "out.json"
        ], timeout=5)
    except subprocess.TimeoutExpired:
        print("エラー：Lambda実行タイムアウト")

# ===== AWS共通 =====
def run_aws_command(cmd):
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=5)

        if result.returncode != 0:
            print("AWSエラー:", result.stderr)
            return None

        if result.stdout.strip() == "":
            return None

        return json.loads(result.stdout)

    except:
        return None

# ===== 各状態チェック =====
def is_s3_private():
    data = run_aws_command([
        "aws", "s3api", "get-public-access-block",
        "--bucket", BUCKET_NAME
    ])
    if not data:
        return False
    c = data["PublicAccessBlockConfiguration"]
    return c["BlockPublicAcls"] and c["IgnorePublicAcls"] and c["BlockPublicPolicy"] and c["RestrictPublicBuckets"]

def is_sg_closed():
    data = run_aws_command([
        "aws", "ec2", "describe-security-groups",
        "--group-ids", SG_ID
    ])
    if not data:
        return False

    for perm in data["SecurityGroups"][0]["IpPermissions"]:
        for ip in perm.get("IpRanges", []):
            if ip.get("CidrIp") == "0.0.0.0/0":
                return False
    return True

def is_ec2_running():
    data = run_aws_command([
        "aws", "ec2", "describe-instances",
        "--instance-ids", INSTANCE_ID
    ])
    if not data:
        return False

    state = data["Reservations"][0]["Instances"][0]["State"]["Name"]
    return state == "running"

def is_cloudtrail_enabled():
    data = run_aws_command([
        "aws", "cloudtrail", "get-trail-status",
        "--name", TRAIL_NAME
    ])
    if not data:
        return False
    return data["IsLogging"]

# ===== メイン =====
try:
    print("Chaos Kitty起動")
    set_normal()

    last_state = GPIO.input(BUTTON)
    press_start = None

    while True:
        current_state = GPIO.input(BUTTON)

        # 押した瞬間
        if last_state == GPIO.HIGH and current_state == GPIO.LOW:
            press_start = time.time()

        # 離した瞬間
        if last_state == GPIO.LOW and current_state == GPIO.HIGH and press_start:
            duration = time.time() - press_start
            print(f"押下時間: {duration:.2f}秒")

            # ===== 長押し（Recover）=====
            if duration >= LONG_PRESS_SEC:
                print("Recover実行")
                invoke_lambda(RECOVER_LAMBDA)
                state = "error"

            # ===== 短押し（Chaos）=====
            else:
                if state == "normal":
                    print("ChaosKittyスタート！🐱")
                    set_error()
                    invoke_lambda(CHAOS_LAMBDA)
                    state = "error"

            press_start = None

        last_state = current_state

        # ===== 復旧検知 =====
        if state == "error":
            set_error()

            s3_ok = is_s3_private()
            sg_ok = is_sg_closed()
            ec2_ok = is_ec2_running()
            ct_ok = is_cloudtrail_enabled()

            if not s3_ok:
                print("S3 NG")
            if not sg_ok:
                print("SG NG")
            if not ec2_ok:
                print("EC2 NG")
            if not ct_ok:
                print("CloudTrail NG")

            print(f"[状態] S3={s3_ok} SG={sg_ok} EC2={ec2_ok} CT={ct_ok}")

            if s3_ok and sg_ok and ec2_ok and ct_ok:
                print("復旧完了！🐱")
                set_normal()
                state = "normal"

        time.sleep(0.1)

except KeyboardInterrupt:
    GPIO.cleanup()