はじめに

所定のレポジトリ内に、ブランチが作成されるたび、特定のポリシーの権限を自動更新するLambdaを書いてみました。
結局この内容は使用されなかったのですが、せっかく書いたので、記事にします。

やりたいこととしては
①以下の記事を参考にして、所定のブランチに対してアクセス制限をかけるIAMポリシーを作成する

➁新しいブランチが作成されるたびに、そのブランチ名を上記で作成したポリシーに追加する

です。

目次

フローの説明
コードの解説
終わりに

フローの説明

今回の流れは以下となります。
※うち➁～⑥がLambdaの処理
①新しいブランチの作成がトリガーとなり、Lambdaにeventが渡される
➁渡されたeventから必要情報を抽出
③更新対象のポリシーの必要情報を抽出
④新しいVersionのポリシーを、既存ポリシーをコピーして作成
⑤新しく作成されたブランチ名を、更新対象のポリシーに反映
⑥最も古いVersionのポリシーを削除

コードの解説

以下が実際のコードになります。
この後詳細を説明していきます。

import boto3
import json
import copy

def get_new_policy_document(old_document, branch_name):
    new_document = copy.deepcopy(old_document)
    new_document["Statement"][0]["Condition"]["StringEquals"]["codecommit:References"].append(f'refs/heads/{branch_name}')
    return(new_document)

def lambda_handler(event, context):
    # CodeCommitからのイベント情報を取得
    record = event['Records'][0]
    repository_name = record['eventSourceARN'].split(':')[5]
    ref = event['Records'][0]['codecommit']['references'][0]['ref']
    branch_name = ref.split('/')[-1]
    print(f"New branch {branch_name} was created in {repository_name} repository.")
    
    # IAMポリシーの更新

    iam = boto3.client('iam')
    policy_arn = 'arn:aws:iam::133285731447:policy/MusakaTestForDevelopmentMember'
    response = iam.list_policy_versions(PolicyArn=policy_arn)
    latest_version = response['Versions'][0]['VersionId']
    policy = iam.get_policy_version(PolicyArn=policy_arn, VersionId=latest_version)
    
    policy_version = iam.create_policy_version(
        PolicyArn=policy_arn,
        PolicyDocument=json.dumps(get_new_policy_document(policy["PolicyVersion"]["Document"], branch_name)),
        SetAsDefault=True
    )
    iam.delete_policy_version(
        PolicyArn=policy_arn, 
        VersionId=response['Versions'][-1]['VersionId']
        )

    print(f"IAM policy {policy_arn} was updated with new policy version {policy_version['PolicyVersion']['VersionId']}.")

ではフローの順番に説明します。

新しいブランチの作成がトリガーとなり、Lambdaにeventが渡される

コンソール上からLambdaにアクセスし、「トリガーを追加」を選択します。
その後、CodeCommitを選択し、必要な項目を埋めていきます。
今回は「ブランチまたはタグを作成する」がリッスンするイベントになります。

渡されたeventから必要情報を抽出

ここまでで、ブランチが作成されるたびにeventがLambdaに渡されるフローが完成しました。
ここから、Lambdaの処理について説明していきます。
ゴールとしては、渡されたeventからレポジトリ名とブランチ名を取得して、以下をプリントすることです。

"New branch {branch_name} was created in {repository_name} repository.

import boto3
import json
import copy

def get_new_policy_document(old_document, branch_name):
    new_document = copy.deepcopy(old_document)
    new_document["Statement"][0]["Condition"]["StringEquals"]["codecommit:References"].append(f'refs/heads/{branch_name}')
    return(new_document)

def lambda_handler(event, context):
    # CodeCommitからのイベント情報を取得
    record = event['Records'][0]
    repository_name = record['eventSourceARN'].split(':')[5]
    ref = event['Records'][0]['codecommit']['references'][0]['ref']
    branch_name = ref.split('/')[-1]
    print(f"New branch {branch_name} was created in {repository_name} repository.")

・必要なモジュールのインポート
今回は以下のモジュールが必要なのでインポートします。

import boto3
import json
import copy

※コードの順番的に、def get_new_policy_document(old_document, branch_name):関数がありますが、こちらの説明は後回しにします。

・渡されたeventから必要な情報の抽出
では渡されたeventから、レポジトリ名とブランチ名を取得していきます。
そもそもeventがどんな形で渡されるかというと、こんな感じのJSONで渡されます。
※以下はあくまで一例で、実際の値ではありません。こちらはLambdaのテストで以下を選択することで確認可能です。

{
  "Records": [
    {
      "awsRegion": "us-east-1",
      "codecommit": {
        "references": [
          {
            "commit": "5c4ef1049f1d27deadbeeff313e0730018be182b",
            "ref": "refs/heads/master"
          }
        ]
      },
      "customData": "this is custom data",
      "eventId": "5a824061-17ca-46a9-bbf9-114edeadbeef",
      "eventName": "TriggerEventTest",
      "eventPartNumber": 1,
      "eventSource": "aws:codecommit",
      "eventSourceARN": "arn:aws:codecommit:us-east-1:123456789012:my-repo",
      "eventTime": "2016-01-01T23:59:59.000+0000",
      "eventTotalParts": 1,
      "eventTriggerConfigId": "5a824061-17ca-46a9-bbf9-114edeadbeef",
      "eventTriggerName": "my-trigger",
      "eventVersion": "1.0",
      "userIdentityARN": "arn:aws:iam::123456789012:root"
    }
  ]
}

そのためには、以下のコードが必要になります。

    record = event['Records'][0]
    repository_name = record['eventSourceARN'].split(':')[5]
    ref = event['Records'][0]['codecommit']['references'][0]['ref']
    branch_name = ref.split('/')[-1]

JSONで必要な値にアクセスするには順番に下っていけばよいので
①「Event」の中の「Records」配列の0番目を変数recordに格納
➁「Records配列」0番目の中の「eventSourceARN」を:で割った5番目を変数repository_nameに格納
③「Records配列」0番目の中の「codecommit」の中の「reference配列」の0番目の「ref」を変数refに格納
④refに格納されたrefs/heads/masterの中の後ろから一つ目

という順番で地道に必要な情報にアクセスしています。
はい、ここまで問題なければ以下がプリントされることになります。

"New branch {branch_name} was created in {repository_name} repository.

更新対象のポリシーの必要情報を抽出

次は以下のコードを用いて、更新対象のポリシーの情報を取得していきます。

iam = boto3.client('iam')
    policy_arn = 'arn:aws:iam::133285731447:policy/MusakaTestForDevelopmentMember'
    response = iam.list_policy_versions(PolicyArn=policy_arn)
    latest_version = response['Versions'][0]['VersionId']
    policy = iam.get_policy_version(PolicyArn=policy_arn, VersionId=latest_version)

まずiamという変数にboto3.client('iam'）で取得した値を格納しています。
そもそもboto3ってなに？という説明ですと、要するにPythonを使用してAWSリソースを使用することができるSDKの一種となります。

例えばS3サービスのアクションを実行したい場合は、以下のように利用することが出来ます。

import boto3

s3 = boto3.client('s3')

# S3クライアント上の 'list_buckets' メソッドを呼び出す
response = s3.list_buckets()

# バケットの一覧を表示
print(response['Buckets'])

今回必要なアクションはIAMですので、boto3.client('IAM')をIAMという変数に代入することを最初に実施しているわけです。
以下公式ドキュメントです。

このコードでは
①更新ポリシーのARNを指定
➁list_policy_versionsで対象ポリシーのversionをリストアップ
③取得されたポリシーのversionのうち最も新しいものを変数latest_versionに格納
④get_policy_versionで更新ポリシー最新versionの情報を取得
という流れの処理が書かれています。

➁番以降を説明します。

以下を見ると、必要な引数を確認することができます。

以下がレスポンスとして返ってきますので、その中から必要情報を変数latest_versionに格納します。
そもそもlatest_versionが必要だった理由ですが、次のget_policy_versionの引数としてVersionIdが必要だったからです。

{
    'Versions': [
        {
            'Document': 'string',
            'VersionId': 'string',
            'IsDefaultVersion': True|False,
            'CreateDate': datetime(2015, 1, 1)
        },
    ],
    'IsTruncated': True|False,
    'Marker': 'string'
}

この後、get_policy_versionでARNで指定したポリシーの最新Versionの詳細を取得します。
以下は取得できる例です。

この後中にある「Document」に対して更新をかけるので、ここで更新対象のポリシーの情報をまとめて取得しています。

{
    "PolicyVersion": {
        "Document": {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Sid": "RestrictBranches",
                    "Effect": "Deny",
                    "Action": [
                        "codecommit:GitPush",
                        "codecommit:DeleteBranch",
                        "codecommit:PutFile",
                        "codecommit:DeleteFile",
                        "codecommit:MergeBranchesByFastForward",
                        "codecommit:MergeBranchesBySquash",
                        "codecommit:MergeBranchesByThreeWay",
                        "codecommit:MergePullRequestByFastForward",
                        "codecommit:MergePullRequestBySquash",
                        "codecommit:MergePullRequestByThreeWay"
                    ],
                    "Resource": "arn:aws:codecommit:*:*:test",
                    "Condition": {
                        "StringEqualsIfExists": {
                            "codecommit:References": [
                                "refs/heads/develop"
                            ]
                        },
                        "Null": {
                            "codecommit:References": False
                        }
                    }
                }
            ]
        },
        "VersionId": "v5",
        "IsDefaultVersion": true,
        "CreateDate": "2023-03-10T06:41:18.000000+00:00"
    },
    "ResponseMetadata": {
        "RequestId": "d097edca-8b6e-4c66-9fff-6883d15d9a47",
        "HTTPStatusCode": 200,
        "HTTPHeaders": {
            "x-amzn-requestid": "d097edca-8b6e-4c66-9fff-6883d15d9a47",
            "content-type": "text/xml",
            "content-length": "2673",
            "date": "Sun, 12 Mar 2023 06:16:26 GMT"
        },
        "RetryAttempts": 0
    }
}

新しいVersionのポリシーを、既存ポリシーをコピーして作成

ここでは、一つ上のセクションで取得したポリシーをコピーして、さらにその中の「Document」に変更を加えていきます。


def get_new_policy_document(old_document, branch_name):
    new_document = copy.deepcopy(old_document)
    new_document["Statement"][0]["Condition"]["StringEquals"]["codecommit:References"].append(f'refs/heads/{branch_name}')
    return(new_document)

 policy_version = iam.create_policy_version(
        PolicyArn=policy_arn,
        PolicyDocument=json.dumps(get_new_policy_document(policy["PolicyVersion"]["Document"], branch_name)),
        SetAsDefault=True
    )

まずは関数の解説から
引数としては、old_documentとbranch_nameを受け取ります。

①受け取ったold_documentをコピーし、new_documentという変数に格納
➁new_document内のdocumentに引数として渡されたbranch_nameを追加
③new_documentを返す

以上で関数の解説は終了です。

新しく作成されたブランチ名を、更新対象のポリシーに反映

上記で解説した関数の処理を、create_policy_versionの中で実施します。
create_policy_versionの詳細は以下より確認できます。

必要な引数としては以下になります。

response = client.create_policy_version(
    PolicyArn='string',
    PolicyDocument='string',
    SetAsDefault=True|False
)

そのためPolicyDocumentの指定で、 get_new_policy_document(old_document, branch_name)関数で取得したnew_documentをjson形式に変換しています。
補足すると、この関数の中で引数として渡しているのは、
①policy = version = response['Versions'][0]['VersionId']
policy = iam.get_policy_version(PolicyArn=policy_arn, VersionId=latest_version)

➁branch_name = ref.split('/')[-1]

になります。
はい、これで新しいブランチが作成されるたび、そのブランチ名をポリシーに自動追加する
が実装できました。

ただしこれではIAMポリシーのversionが無限に増え続けてしまうので、以下の記述で最も古いIAMポリシーを削除します。

    iam.delete_policy_version(
        PolicyArn=policy_arn, 
        VersionId=response['Versions'][-1]['VersionId']
        )

以上コードの解説を終了します。

終わりに

頑張ってコードを書いたのですが、所定のブランチに対してアクセス権限を管理したいのであれば
①PowerUserポリシーをアタッチ
➁アクセスを制限したい箇所に対して、denyをかける

で実装できたので、なんというか空回りしてしまった感じがあります。
とはいえ、Boto3のよい勉強になったので結果的にはよかったかと思います。