AssumeRoleをECS&Digdag環境で利用する

やりたいこと

• AWS アカウント A

  • S3 に日次でファイルが置かれる

• AWS アカウント B

  • アカウント A の S3 のファイルを取得したい
  • 実行環境は ECS 上で動いている Digdag サーバー

AssumeRole は一時的なクレデンシャルを生成して、他の AWS アカウントからのアクセスを許可できるので、以下のようなセキュリティ的に避けたい設定を行わなくて済みます

• IAM ユーザーを作成し、AWS アクセスキー発行
• S3 の公開設定

AWS アカウント A - 参照先

ロールを作成

1. AWS コンソールからロールを作成
2. 信頼されたエンティティの種類を「別の AWS アカウント」にして、アカウント B のアカウント ID を設定
   • この段階では、アカウント B の root に権限が付与される（ここは後で変えます）
3. アカウント B 用のポリシーを作成して、ロールにアタッチ
   以下は今回のやりたいことを実現する最低限のポリシーです

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Sid": "VisualEditor0",
         "Effect": "Allow",
         "Action": "s3:GetObject",
         "Resource": ["arn:aws:s3:::[バケット名]/*"]
       }
     ]
   }

AWS アカウント B - 参照元

AssumeRole のポリシーを作成

1. AWS コンソールからポリシーを作成

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Sid": "VisualEditor0",
         "Effect": "Allow",
         "Action": "sts:AssumeRole",
         "Resource": "arn:aws:iam::[アカウントAのID]:role/[さっき作ったロール名]"
       }
     ]
   }

1. ECS タスクのロールにこのポリシーをアタッチ
   ECS タスクで使われているロールは、コンソールの ECS タスク定義から、「タスク実行ロール」で確認できます

AWS アカウント A - 参照先

作成したロールの信頼関係タブから、以下のように編集

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "arn:aws:iam::[アカウントBのID]:role/ECSタスクロール名"
      },
      "Action": "sts:AssumeRole",
      "Condition": {}
    }
  ]
}

さきほどは、アカウント B の root に権限を与えていましたが、ECS タスクロールに権限を与えるように変更しました

動作確認

AWS CLI

Digdag のタスクが sh で、AWS CLI を使う場合

ECS タスクで使う Docker などで以下を用意する

~/.aws/configを作成

[profile assume]
role_arn = arn:aws:iam::[アカウントAのID]:role/[ロール名]
credential_source = EcsContainer

または

echo "[profile assume]" > ~/.aws/config
echo "role_arn = arn:aws:iam::[アカウントAのID]:role/[ロール名]" >> ~/.aws/config
echo "credential_source = EcsContainer" >> ~/.aws/config

aws s3 cp [コピー元] [コピー先] --recursive --profile assume

Python

Digdag のタスクが Python の場合

from sts import STS

# role_arn: arn:aws:iam::[アカウントAのID]:role/[ロール名]
# bucket: 参照先のバケット名
# key: 参照先のオブジェクトキー
# save_file_path: 保存先（Digdagコンテナー内）
class S3(object):
    def get_by_sts(self, role_arn, bucket, key, save_file_path):
        resource = STS().get_resource(role_arn, "s3")

        resource.Bucket(bucket).download_file(
            key,
            save_file_path,
        )

import boto3


class STS(object):
    sts_client = boto3.client("sts")

    def get_resource(self, role_arn, resource_name):
        role_obj = self.sts_client.assume_role(
            RoleArn=role_arn,
            RoleSessionName="AssumeRoleSession",
        )

        resource = boto3.resource(
            resource_name,
            aws_access_key_id=role_obj["Credentials"]["AccessKeyId"],
            aws_secret_access_key=role_obj["Credentials"]["SecretAccessKey"],
            aws_session_token=role_obj["Credentials"]["SessionToken"],
        )

        return resource

boto3.client("sts").assume_roleに RoleArn と任意のセッション名を渡して、一時的なクレデンシャルを発行しています