開発と本番が別アカウントのEC2（Ubuntu） に Patch Manager で順番にパッチを当ててみた

import datetime
import logging
import os

import boto3

logger = logging.getLogger()
logger.setLevel(logging.INFO)

BUCKET = os.environ["BUCKET"]
DEV_TAG_VALUE = os.environ.get("DEV_TAG_VALUE", "dev")
JST = datetime.timezone(datetime.timedelta(hours=9))

ec2 = boto3.client("ec2")
ssm = boto3.client("ssm")
s3 = boto3.client("s3")

def dev_instance_ids():
    """タグ Environment=dev の起動中・停止中の EC2 を返す"""
    pages = ec2.get_paginator("describe_instances").paginate(
        Filters=[
            {"Name": "tag:Environment", "Values": [DEV_TAG_VALUE]},
            {"Name": "instance-state-name", "Values": ["running", "stopped"]},
        ]
    )
    return [i["InstanceId"] for p in pages for r in p["Reservations"] for i in r["Instances"]]

def installed_patches(instance_id):
    """ベースラインの対象で、入っているパッチを {KBId: Title} で返す"""
    patches = {}
    pages = ssm.get_paginator("describe_instance_patches").paginate(
        InstanceId=instance_id,
        Filters=[{"Key": "State", "Values": ["Installed", "InstalledPendingReboot"]}],
    )
    for page in pages:
        for p in page["Patches"]:
            patches[p["KBId"]] = p["Title"]
    return patches

def to_yaml(patches):
    """InstallOverrideList の形式。id はパッケージ名.アーキテクチャ、title はバージョン付き"""
    lines = ["patches:"]
    for kb_id, title in sorted(patches.items()):
        lines.append("  -")
        lines.append("    id: '{}'".format(kb_id.replace("'", "''")))
        lines.append("    title: '{}'".format(title.replace("'", "''")))
    return "\n".join(lines) + "\n"

def lambda_handler(event, context):
    date = event.get("date") or datetime.datetime.now(JST).strftime("%Y-%m-%d")

    instance_ids = dev_instance_ids()
    if not instance_ids:
        raise RuntimeError("開発の EC2 が見つかりません")

    # 開発の全台に同じバージョンで入っているものだけを残す
    common = None
    for instance_id in instance_ids:
        patches = installed_patches(instance_id)
        if common is None:
            common = patches
        else:
            common = {k: v for k, v in common.items() if patches.get(k) == v}

    key = f"candidates/{date}.yaml"
    s3.put_object(Bucket=BUCKET, Key=key, Body=to_yaml(common).encode(), ContentType="application/x-yaml")

    logger.info("instances=%s patches=%d key=%s", instance_ids, len(common), key)
    return {"key": key, "patches": len(common)}

import datetime
import logging
import os

import boto3

logger = logging.getLogger()
logger.setLevel(logging.INFO)

BUCKET = os.environ["BUCKET"]
DEV_TAG_VALUE = os.environ.get("DEV_TAG_VALUE", "dev")
MAX_AGE_HOURS = int(os.environ.get("MAX_AGE_HOURS", "24"))

ec2 = boto3.client("ec2")
ssm = boto3.client("ssm")
s3 = boto3.client("s3")

def dev_instances():
    """タグ Environment=dev の EC2 を、終了したもの以外すべて {ID: 状態} で返す"""
    pages = ec2.get_paginator("describe_instances").paginate(
        Filters=[
            {"Name": "tag:Environment", "Values": [DEV_TAG_VALUE]},
            {"Name": "instance-state-name", "Values": ["pending", "running", "stopping", "stopped"]},
        ]
    )
    return {
        i["InstanceId"]: i["State"]["Name"]
        for p in pages for r in p["Reservations"] for i in r["Instances"]
    }

def latest_candidate_date():
    """candidates/ の中で、日付がいちばん新しい一覧の日付を返す。なければ None"""
    keys = []
    pages = s3.get_paginator("list_objects_v2").paginate(Bucket=BUCKET, Prefix="candidates/")
    for page in pages:
        keys += [o["Key"] for o in page.get("Contents", []) if o["Key"].endswith(".yaml")]
    return max(keys)[len("candidates/"):-len(".yaml")] if keys else None

def check_instance(instance_id, state_name, now):
    """開発の 1 台が合格かを調べ、不合格の理由のリストを返す（空なら合格）"""
    if state_name != "running":
        return [f"起動していない（{state_name}）"]
    reasons = []

    info = ssm.describe_instance_information(
        Filters=[{"Key": "InstanceIds", "Values": [instance_id]}]
    )["InstanceInformationList"]
    if not info or info[0]["PingStatus"] != "Online":
        reasons.append("Systems Manager に Online で登録されていない")

    states = ssm.describe_instance_patch_states(InstanceIds=[instance_id])["InstancePatchStates"]
    if not states:
        reasons.append("パッチの状態がまだ報告されていない")
        return reasons
    state = states[0]
    if state.get("FailedCount", 0) > 0:
        reasons.append(f"失敗したパッチがある（{state['FailedCount']} 件）")
    if state.get("InstalledPendingRebootCount", 0) > 0:
        reasons.append(f"再起動待ちのパッチがある（{state['InstalledPendingRebootCount']} 件）")
    age = now - state["OperationEndTime"]
    if age > datetime.timedelta(hours=MAX_AGE_HOURS):
        reasons.append(f"最後のパッチ操作が {MAX_AGE_HOURS} 時間より前（{state['OperationEndTime']}）")
    return reasons

def ng(approved, reasons):
    """不合格のときは、同じ日付の承認済みの一覧が前の実行で残っていても消す"""
    s3.delete_object(Bucket=BUCKET, Key=approved)
    logger.warning("NG: %s", reasons)
    return {"result": "NG", "reasons": reasons}

def lambda_handler(event, context):
    now = datetime.datetime.now(datetime.timezone.utc)
    # 日付を省略したときは、いちばん新しい候補の一覧（前夜に作ったもの）を確かめる
    date = event.get("date") or latest_candidate_date()
    if date is None:
        logger.warning("NG: 候補の一覧が 1 つもない")
        return {"result": "NG", "reasons": ["候補の一覧が 1 つもない"]}
    candidate = f"candidates/{date}.yaml"
    approved = f"approved/{date}.yaml"

    try:
        s3.head_object(Bucket=BUCKET, Key=candidate)
    except s3.exceptions.ClientError:
        return ng(approved, [f"{candidate} がない"])

    instances = dev_instances()
    if not instances:
        return ng(approved, ["開発の EC2 がない"])

    results = {i: check_instance(i, s, now) for i, s in instances.items()}
    failed = {i: r for i, r in results.items() if r}
    if failed:
        return ng(approved, failed)

    s3.copy_object(Bucket=BUCKET, Key=approved, CopySource={"Bucket": BUCKET, "Key": candidate})
    logger.info("OK: instances=%s approved=%s", list(instances), approved)
    return {"result": "OK", "approved": approved}

import logging
import os

import boto3

logger = logging.getLogger()
logger.setLevel(logging.INFO)

LIST_BUCKET = os.environ["LIST_BUCKET"]
PROD_TAG_VALUE = os.environ.get("PROD_TAG_VALUE", "prod")
REBOOT_OPTION = os.environ.get("REBOOT_OPTION", "NoReboot")

s3 = boto3.client("s3")
ssm = boto3.client("ssm")

def latest_approved_key():
    """approved/ の中で、日付がいちばん新しい一覧のキーを返す。なければ None"""
    keys = []
    pages = s3.get_paginator("list_objects_v2").paginate(Bucket=LIST_BUCKET, Prefix="approved/")
    for page in pages:
        keys += [o["Key"] for o in page.get("Contents", []) if o["Key"].endswith(".yaml")]
    return max(keys) if keys else None

def lambda_handler(event, context):
    if event.get("date"):
        key = f"approved/{event['date']}.yaml"
        try:
            s3.head_object(Bucket=LIST_BUCKET, Key=key)
        except s3.exceptions.ClientError:
            key = None
    else:
        key = latest_approved_key()

    if key is None:
        logger.warning("SKIP: 承認済みの一覧がないため、本番には当てません event=%s", event)
        return {"result": "SKIP"}

    list_url = f"s3://{LIST_BUCKET}/{key}"
    command = ssm.send_command(
        DocumentName="AWS-RunPatchBaseline",
        Targets=[{"Key": "tag:Environment", "Values": [PROD_TAG_VALUE]}],
        Parameters={
            "Operation": ["Install"],
            "RebootOption": [REBOOT_OPTION],
            "InstallOverrideList": [list_url],
        },
        TimeoutSeconds=600,
        MaxConcurrency="50",
        MaxErrors="0",
        Comment=f"apply {key}",
    )["Command"]

    logger.info("SENT: command_id=%s list=%s", command["CommandId"], list_url)
    return {"result": "SENT", "commandId": command["CommandId"], "list": list_url}