本記事は AWSアワード受賞者祭り 17日目の記事です。
✨🏆 16日目 ▶▶ 本記事 ▶▶ 18日目 🏆✨

はじめに
アーカイブルールとは
- アーカイブルール概要
- アーカイブルールを作成しても既存の結果には適用されない
カスタムリソースによるアーカイブルールの適用
テンプレート
- CloudFormationテンプレート
  - ダミーパラメータを設定する
  - ServiceTimeoutを利用する
カスタムリソースのLambda関数のコード
デプロイ手順
終わりに

はじめに

こんにちは、藤本です。
この度、「2025 Japan AWS Jr. Champions」、「2025 Japan All AWS Certifications Engineers」に選出いただきました。「2025 Japan AWS Jr. Champions」の選出の経緯は、ブログでまとめているので、よかったら見てください!

tech.nri-net.com

今回は、以前執筆したブログ「マルチアカウント環境における AWS IAM Access Analyzer を用いた権限管理」をベースに、カスタムリソースによるIAM Access Analyzerのアーカイブルールの自動適用についてまとめてみました。 IAM Access Analyzerとは、IAMロールやリソースに対するアクセス許可を分析するサービスです。詳しい概要については、こちらのブログでまとめているので、ご参照ください。

アーカイブルールとは

アーカイブルール概要

アーカイブルール (Archive rules) とは、IAM Access Analyzerが検出した外部からのアクセスが可能であるリソースに対して、特定の条件に合致するアクセスを自動的にアーカイブするための設定です。検出結果として確認不要なアクセスをアーカイブルールによって、フィルタリングすることで検出結果のノイズを減らします。例えば、一時的または検証目的で作成されるようなテスト環境用のリソースや監視目的のSaaSで許可されているリソース等がアーカイブ対象となります。
アーカイブルールを多用しすぎると、意図しないアクセスを見逃す可能性があるため、アーカイブルールの取り扱いには注意が必要です。そのため定期的に、アーカイブルールを見直すような運用が必要となります。

アーカイブルールを作成しても既存の結果には適用されない

アーカイブルールの注意点として、アーカイブルールの作成時に既存の検出結果には適用されない点があります。新規に検出されたアクセスに対しては自動でアーカイブされますが、作成前に検知済みであるアクセスに関してはアーカイブされません。また、アーカイブルールを変更した場合でも、既存の結果には再適用が行われません。そのため、アーカイブルール作成・更新後には、明示的に適用を行う必要があります。アーカイブルールの適用を実行するにはマネジメントコンソール、CLI、カスタムリソースによる適用のいずれかの対応が必要になります。

マネジメントコンソール上からアーカイブルールを適用する場合は、アーカイブルール作成時に「アクティブな検出結果を作成およびアーカイブ」を選択することで、既存の分析結果にも適用されます。「ルールを作成」を選択すると、既存の結果には適用されず、アーカイブルールの作成のみが行われます。

CLIによるアーカイブルール適用の場合は、apply-archive-ruleコマンドを実行します。作成されたアーカイブルールの一覧を確認したい場合は、list-archive-rulesコマンドを実行します。

aws accessanalyzer apply-archive-rule \
  --analyzer-arn <アナライザーARN> \
  --rule-name <アーカイブルール名>

apply-archive-rule — AWS CLI 2.27.61 Command Reference

aws accessanalyzer list-archive-rules \
    --analyzer-name <アナライザー名>

list-archive-rules — AWS CLI 2.27.61 Command Reference

カスタムリソースによる適用に関しては、次の章で解説します。

カスタムリソースによるアーカイブルールの適用

カスタムリソースとは

カスタムリソースとは、AWS CloudFormationで定義されていないリソースや複雑な処理を、AWS LambdaやAmazon SNSなどを使って拡張的に実行する仕組みのことです。 CloudFormationが対応していない特定のリソースのAPI呼び出しや外部のサービスとの連携を行う際などに利用されます。 CloudFormationでIAM AccessAnalyzerを展開する場合、IAM Access Analyzerのアーカイブルール適用は、 CloudFormationでは対応していない操作になるため、アーカイブルールの作成と適用を同時に行いたい場合は、Lambdaによるカスタムリソースの実装が必要になります。

カスタムリソースの実行タイミング

カスタムリソースは、CloudFormationスタックの作成・更新・削除のタイミングで実行されます。これらの操作に応じて、CloudFormationはカスタムリソースに対してリクエスト（イベントタイプ）を送信します。

イベントタイプ	発生のタイミング
`Create`	CloudFormationスタックの新規作成時
`Update`	CloudFormationスタックの更新時
`Delete`	CloudFormationスタックの削除時

Lambda関数では、イベントタイプごとに処理を分けて実装することが推奨されており、分岐を行わないと意図しない動作やスタックエラーの原因となる可能性があります。例えば、 IAM Access Analyzerの場合、CloudFormationの作成・更新時にのみアーカイブルールを適用し、削除時にはカスタムリソースのLambda関数による処理を行わないように制御します。イベントタイプは、Lambda関数のeventのRequestTypeに含まれています。

def lambda_handler(event, context):
    request_type = event["RequestType"]
    
    if request_type == "Create":
        # スタックの新規作成時の処理
    elif request_type == "Update":
        # スタック更新時の処理
    elif request_type == "Delete":
        # スタック削除時の処理

CloudFormationに対してレスポンスが必要

Lambda関数の処理が完了したら、CloudFormationに対して、Lambda関数の処理結果（SUCCESSまたはFAILED）を通知する必要があります。この通知は、CloudFormationから渡されるリクエスト内に含まれている ResponseURL（署名付きS3のURL）に対して、HTTP PUTリクエストでレスポンスを送信することで行います。レスポンスが行われない場合、CloudFormationはLambda関数からの応答をデフォルトのタイムアウト時間である1時間（3600秒）待機し、それでも応答がない場合はスタック操作を失敗（FAILED）として終了します。タイムアウト時間が経過するまでは、スタックへの操作が実行できません。

レスポンスの処理を簡単に行う方法として、cfnresponse-sendモジュールがあります。これは、Lambda関数を呼び出したカスタムリソースへの応答の送信を簡素化するAWS公式のライブラリのことです。

docs.aws.amazon.com

cfnresponse-sendモジュールは、CloudFormationテンプレート内のZipFileプロパティを使用して、Pythonコードをインラインで記述する場合にのみ利用可能です。つまり、テンプレート内にPythonコードを直接埋め込む形式で利用することが前提となります。
ただし、インライン記述にはいくつかのデメリットがあります。

保守性が低い
コードの可読性や再利用性が下がり、バージョン管理もしにくくなります。
デバッグが困難
問題がCloudFormation側にあるのか、Pythonコード側にあるのか切り分けが難しくなります。そのため、CloudFormation テンプレートと Lambdaコードは分離した方が望ましいです。
コードサイズ制限
ZipFileプロパティには、4096Byte（4KB）までというサイズ制限があります。

これらの理由から、一般的には Lambda関数のコードをS3バケットに格納し、そこから参照する方法が採用されています。インライン記述を採用する場合は、ごく単純な処理に限定して使用します。

S3バケットに格納されたコードでは、cfnresponse-sendモジュールは使用できません。これは、cfnresponse-sendモジュールがLambdaランタイムに標準で含まれておらず、ModuleNotFoundErrorが発生するためです。そのため、S3上のコードでCloudFormationに応答を返す必要がある場合は、レスポンス処理を独自に関数として実装する必要があります。
今回は、小西さんが以前執筆されたブログ「AWS LambdaカスタムリソースでAWS Cloudformationスタックを別リージョンにデプロイする」の内容とAWSのドキュメントを参考にレスポンス処理を独自の関数で作成しました。

テンプレート

CloudFormationテンプレート

CloudFormationテンプレートでは、主にIAM Access Analyzerのアナライザーとアーカイブルール適用のためのLambdaを作成しています。

CloudFormationテンプレート

AWSTemplateFormatVersion: '2010-09-09'
Description: Create IAM Access Analyzer and apply archive rule automatically via Lambda (Custom Resource)
Parameters:
  AnalyzerName:
    Description: The name of the Analyzer.
    Type: String
    Default: IAMAccessAnalyzer
  CustomResourceUpdateToken:
    Description: Execute a single CustomResource.
    Type: String
    Default: 0
  LambdaCodeBucket:
    Description: S3 bucket containing Lambda source code (zip).
    Type: String
  LambdaCodeKey:
    Description: S3 key (path) to Lambda deployment package.
    Type: String

Resources:
  Analyzer:
    Type: AWS::AccessAnalyzer::Analyzer
    Properties:
      AnalyzerName: !Sub "${AnalyzerName}-${AWS::AccountId}-${AWS::Region}"
      Type: ACCOUNT
      ArchiveRules:
        - Filter:
          - Property: resourceType
            Eq:
            - AWS::IAM::Role
          - Property: resource
            Contains:
              - test
          RuleName: PrefixIAMRoleRule

  RunCustomResourceApplyArchiveRule:
    Type: 'Custom::ApplyArchiveRule'
    DependsOn: ApplyArchiveRuleLambda
    Properties:
      ServiceToken: !GetAtt ApplyArchiveRuleLambda.Arn
      ServiceTimeout: 60
      UpdateDummyInput: {Ref: CustomResourceUpdateToken}

  ApplyArchiveRuleLambda:
    Type: AWS::Lambda::Function
    DependsOn:
      - ApplyArchiveRuleRole
      - Analyzer
    Properties:
      FunctionName: ApplyArchiveRule
      Description : apply_archive_rule
      Runtime: python3.12
      MemorySize: 1024
      Timeout: 900
      Role: !GetAtt ApplyArchiveRuleRole.Arn
      Handler: index.lambda_handler
      Code:
        S3Bucket: !Ref LambdaCodeBucket
        S3Key: !Ref LambdaCodeKey
      Environment:
        Variables:
          ANALYZER_ARN: !GetAtt Analyzer.Arn
          ANALYZER_NAME: !Sub "${AnalyzerName}-${AWS::AccountId}-${AWS::Region}"

  ApplyArchiveRuleRole: 
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Sub 'ApplyArchiveRule-${AWS::Region}'
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - lambda.amazonaws.com
            Action:
              - sts:AssumeRole
      Policies:
        - PolicyName: !Sub 'IAMPolicy-ApplyArchiveRule-${AWS::Region}'
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Sid: ListRules
                Effect: Allow
                Action:
                  - access-analyzer:ListArchiveRules
                Resource: "*"
              - Sid: ApplyRule
                Effect: Allow
                Action:
                  - access-analyzer:ApplyArchiveRule
                Resource: !GetAtt Analyzer.Arn
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole

CloudFormationによるIAM Access Analyzer作成の詳細は、AWS CloudFormationユーザーガイド AccessAnalyzerをご確認ください。

CloudFormationテンプレート作成において、Tipsを紹介します。

ダミーパラメータを設定する

CloudFormationは、カスタムリソースのPropertiesに変更がない限り、スタック更新時にLambda関数を再実行しません。カスタムリソースの処理を再実行したい場合、見かけ上の変更を加える必要があります。そのために、ダミーパラメータ（今回の場合は、CustomResourceUpdateTokenと定義）を使用します。このダミーパラメータの値を変更し、CloudFormationを更新させることで、カスタムリソースを強制的に再実行します。今回のケースだと、マネジメントコンソールなどでアーカイブルール作成のみが行われた後、ダミーパラメータ（CustomResourceUpdateToken）の値を変更することで、アーカイブルールの適用を別途実行できます。

`ServiceTimeout`を利用する

「CloudFormationに対してレスポンスが必要」の章で説明した通り、CloudFormationのカスタムリソースでは、Lambda関数からのレスポンスがない場合、デフォルトで最大1時間（3600秒）スタック操作が待機状態になります。タイムアウト時間が経過するまでは、他の操作も実行できないため、開発・検証時には大きなボトルネックになります。この問題を回避するために、カスタムリソースにServiceTimeoutを設定します。

docs.aws.amazon.com

ServiceTimeoutは、カスタムリソースのタイムアウト時間を秒単位で明示的に指定できるプロパティです。 ServiceTimeout: 300と設定すれば、Lambda関数がエラーになった場合でもCloudFormationは5分後に処理を失敗（FAILED）として終了し、不要な待機時間を防ぐことができます。

  RunCustomResourceApplyArchiveRule:
    Type: 'Custom::ApplyArchiveRule'
    DependsOn: ApplyArchiveRuleLambda
    Properties:
      ServiceToken: !GetAtt ApplyArchiveRuleLambda.Arn
      ServiceTimeout: 60 # タイムアウト時間設定
      UpdateDummyInput: {Ref: CustomResourceUpdateToken} # ダミーパラメータ

カスタムリソースのLambda関数のコード

今回作成したカスタムリソースのLambda関数では、IAM Access Analyzerの名前やARNをPythonコード内でハードコーディングせず、CloudFormationの環境変数（Environment:）から取得するようにしています。これにより、可読性や保守性が向上し、CloudFormation側の値を変更するだけで、柔軟に対応できるようになります。

カスタムリソースのLambda関数のコード

import os
import json
import boto3
import botocore
import logging
import urllib.request

# ロガー設定
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# 環境情報
region = os.environ.get('AWS_REGION')
sts_client = boto3.client("sts", region_name=region)
accountId = sts_client.get_caller_identity()["Account"]

# CloudFormationの環境変数を受け取る
ANALYZER_ARN = os.environ["ANALYZER_ARN"]
ANALYZER_NAME = os.environ.get("ANALYZER_NAME") or ANALYZER_ARN.split("/")[-1]

def lambda_handler(event, context):
    logger.info(f"Received event: {json.dumps(event)}")

    try:
        request_type = event['RequestType']

        if request_type == 'Delete': # スタック削除時
            logger.info("Delete event received. No action needed.")
            cfn_response_send(event, context, 'SUCCESS', {}, reason='Delete request - no action')
            return

        elif request_type in ['Create', 'Update']: # スタック作成・更新時
            archive_rule_names = get_list_archiverules()
            if archive_rule_names:
                apply_archiverules(archive_rule_names)
                message = f"Applied archive rules: {archive_rule_names}"
            else:
                message = "No archive rules found"
            cfn_response_send(event, context, 'SUCCESS', {'Message': message})
            return

        cfn_response_send(event, context, 'FAILED', {}, reason=f"Unexpected RequestType: {request_type}")

    except Exception as e:
        logger.error(f"Unhandled exception: {e}")
        cfn_response_send(event, context, 'FAILED', {}, reason=str(e))

# アーカイブルールの一覧を取得
def get_list_archiverules():
    accessanalyzer_client = boto3.client('accessanalyzer')
    archive_rule_names = []

    try:
        paginator = accessanalyzer_client.get_paginator('list_archive_rules')
        for page in paginator.paginate(analyzerName=ANALYZER_NAME):
            archive_rule_names.extend(rule['ruleName'] for rule in page.get('archiveRules', []))
    except botocore.exceptions.ClientError as e:
        logger.error(f"Failed to list archive rules: {e}")

    return archive_rule_names

# アーカイブルールを適用
def apply_archiverules(archive_rule_names):
    accessanalyzer_client = boto3.client('accessanalyzer')
    for rule_name in archive_rule_names:
        try:
            accessanalyzer_client.apply_archive_rule(
                analyzerArn=ANALYZER_ARN,
                ruleName=rule_name
            )
            logger.info(f"Applied archive rule: {rule_name}")
        except botocore.exceptions.ClientError as e:
            logger.error(f"Error applying archive rule {rule_name}: {e}")
            continue

# 独自のレスポンス処理
def cfn_response_send(event, context, response_status, response_data,
                      physical_resource_id=None, no_echo=False, reason=None):
    response_url = event['ResponseURL']
    logger.info(f"Response URL: {response_url}")

    response_body = {
        'Status': response_status,
        'Reason': reason or f"See the details in CloudWatch Log Stream: {context.log_stream_name}",
        'PhysicalResourceId': physical_resource_id or context.log_stream_name,
        'StackId': event['StackId'],
        'RequestId': event['RequestId'],
        'LogicalResourceId': event['LogicalResourceId'],
        'NoEcho': no_echo,
        'Data': response_data
    }

    json_response_body = json.dumps(response_body)
    logger.info(f"Response body: {json_response_body}")

    headers = {
        'Content-Type': '',
        'Content-Length': str(len(json_response_body))
    }

    try:
        request = urllib.request.Request(
            response_url,
            data=json_response_body.encode('utf-8'),
            headers=headers,
            method='PUT'
        )
        with urllib.request.urlopen(request) as response:
            logger.info(f"CloudFormation response sent. Status code: {response.status}")
    except Exception as e:
        logger.error(f"Failed to send CloudFormation response: {e}")

デプロイ手順

共有したカスタムリソースのLambda関数のコードとCloudFormationテンプレートを用いて、デプロイを行います。

1.Lambda関数のコードをS3にアップロード
Lambda関数のコード（index.py）をzip化した上で、S3バケットにアップロードします。 S3バケットは別途作成します。

2.Cloud Shellにて、CloudFormationコマンド実行
Cloud Shell上にCloudFormationテンプレート(YAMLファイル)をアップロードした上で、以下のコマンドを実行します。

aws cloudformation deploy \
  --template-file <CloudFormationテンプレート名> \
  --stack-name <スタック名> \
  --capabilities CAPABILITY_NAMED_IAM \
  --parameter-overrides \
      LambdaCodeBucket=<アップロード先のS3バケット名>   \
      LambdaCodeKey=<S3バケットにアップロードしたzipファイル名>

終わりに

アーカイブルールの自動適用というニッチな内容でしたが、いかがでしたでしょうか。本ブログで、CloudFormationにおけるカスタムリソースの使い方・アーカイブルール自動適用の参考にしていただければと思います。また、前のブログでも触れましたが、IAM Access Analyzerではリソースの意図しない共有を検出することはできますが、アナライザー自体が自動的に修正を行うことはありません。そのため、検出された内容に対する対応方針や運用プロセスを事前に検討しておくことが重要です。