Amazon EventBridgeでAWS CodePipelineの手動承認を自動化する方法

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "codepipeline:GetPipelineState"
      ],
      "Resource": "arn:aws:codepipeline:REGION:ACCOUNT_ID:PIPELINE_NAME"
    },
    {
      "Effect": "Allow",
      "Action": [
        "codepipeline:PutApprovalResult"
      ],
      "Resource": "arn:aws:codepipeline:REGION:ACCOUNT_ID:PIPELINE_NAME/STAGE_NAME/ACTION_NAME"
    },
    {
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "*"
    }
  ]
}

# ロールの作成
aws iam create-role --role-name LambdaCodePipelineApprovalRole \
  --assume-role-policy-document '{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"Service":"lambda.amazonaws.com"},"Action":"sts:AssumeRole"}]}'

# ポリシーのアタッチ
aws iam put-role-policy --role-name LambdaCodePipelineApprovalRole \
  --policy-name CodePipelineApprovalPolicy \
  --policy-document file://lambda-codepipeline-policy.json

import json
import os

import boto3


def lambda_handler(event, context):
    # パラメータの取得
    pipeline_name = os.environ.get("PIPELINE_NAME")
    stage_name = os.environ.get("STAGE_NAME")
    action_name = os.environ.get("ACTION_NAME")

    client = boto3.client("codepipeline")

    # パイプラインの状態を取得
    response = client.get_pipeline_state(name=pipeline_name)

    # 指定したステージとアクションの承認トークンを取得
    token = get_approval_token(response, stage_name, action_name)

    if not token:
        return {
            "statusCode": 404,
            "body": json.dumps(
                f"No pending approval found for {pipeline_name}/{stage_name}/{action_name}"
            ),
        }

    # 承認を実行
    response = client.put_approval_result(
        pipelineName=pipeline_name,
        stageName=stage_name,
        actionName=action_name,
        result={
            "summary": "Automatically approved by EventBridge",
            "status": "Approved",
        },
        token=token,
    )

    return {
        "statusCode": 200,
        "body": json.dumps(
            f"Successfully approved {pipeline_name}/{stage_name}/{action_name}"
        ),
    }


def get_approval_token(pipeline_state, stage_name, action_name):
    """承認トークンを見つけて返す関数"""
    for stage in pipeline_state.get("stageStates", []):
        if stage["stageName"] != stage_name:
            continue

        for action in stage.get("actionStates", []):
            if action["actionName"] != action_name:
                continue

            execution = action.get("latestExecution", {})
            return execution.get("token")

    return None

zip lambda_function.zip lambda_function.py
aws lambda create-function \
  --function-name CodePipelineAutoApproval \
  --runtime python3.13 \
  --handler lambda_function.lambda_handler \
  --role arn:aws:iam::<ACCOUNT_ID>:role/LambdaCodePipelineApprovalRole \
  --zip-file fileb://lambda_function.zip \
  --timeout 30 \
  --environment Variables="{PIPELINE_NAME=PIPELINE_NAME,STAGE_NAME=STAGE_NAME,ACTION_NAME=ACTION_NAME}"

aws events put-rule \
  --name EventBridgeCodePipelineApproval \
  --schedule-expression "cron(0 3 1 4 ? 2025)" \
  --state ENABLED

aws events put-targets \
  --rule EventBridgeCodePipelineApproval \
  --targets "Id"="1","Arn"="arn:aws:lambda:<REGION>:<ACCOUNT_ID>:function:CodePipelineAutoApproval"

aws lambda add-permission \
  --function-name CodePipelineAutoApproval \
  --statement-id EventBridgeInvoke \
  --action lambda:InvokeFunction \
  --principal events.amazonaws.com \
  --source-arn arn:aws:events:<REGION>:<ACCOUNT_ID>:rule/EventBridgeCodePipelineApproval

aws lambda invoke \
  --function-name CodePipelineAutoApproval \
  --payload '{}' \
  response.json