Lambda関数からWorkload Identity Federationを使用してGoogle DriveからS3にファイルをアップロードしてみる

gcloud services enable drive.googleapis.com \
  --project=YOUR_GCP_PROJECT_ID

gcloud iam service-accounts create YOUR_SA_NAME \
  --display-name="Google Drive to S3 Transfer" \
  --project=YOUR_GCP_PROJECT_ID

gcloud iam workload-identity-pools create "aws-s3-pool" \
  --project=YOUR_GCP_PROJECT_ID \
  --location="global" \
  --display-name="AWS S3 Transfer Pool"

gcloud iam workload-identity-pools providers create-aws "aws-provider" \
  --project=YOUR_GCP_PROJECT_ID \
  --location="global" \
  --workload-identity-pool="aws-s3-pool" \
  --display-name="AWS Provider" \
  --account-id="YOUR_AWS_ACCOUNT_ID" \
  --attribute-mapping="google.subject=assertion.arn,attribute.aws_role=assertion.arn.extract('assumed-role/{role}/')" \
  --attribute-condition="assertion.arn.startsWith('arn:aws:sts::YOUR_AWS_ACCOUNT_ID:assumed-role/')"

# プロジェクト番号を取得
PROJECT_NUMBER=$(gcloud projects describe YOUR_GCP_PROJECT_ID --format="value(projectNumber)")

gcloud iam service-accounts add-iam-policy-binding \
  YOUR_SA_NAME@YOUR_GCP_PROJECT_ID.iam.gserviceaccount.com \
  --project=YOUR_GCP_PROJECT_ID \
  --role="roles/iam.workloadIdentityUser" \
  --member="principal://iam.googleapis.com/projects/${PROJECT_NUMBER}/locations/global/workloadIdentityPools/aws-s3-pool/subject/arn:aws:sts::YOUR_AWS_ACCOUNT_ID:assumed-role/YOUR_LAMBDA_ROLE_NAME"

aws lambda get-function-configuration \
  --function-name YOUR_FUNCTION_NAME \
  --query "Role" --output text
# 出力例: arn:aws:iam::YOUR_AWS_ACCOUNT_ID:role/YOUR_LAMBDA_ROLE_NAME

gcloud iam workload-identity-pools create-cred-config \
  "projects/${PROJECT_NUMBER}/locations/global/workloadIdentityPools/aws-s3-pool/providers/aws-provider" \
  --service-account="YOUR_SA_NAME@YOUR_GCP_PROJECT_ID.iam.gserviceaccount.com" \
  --aws \
  --output-file="credential-config.json"

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "lambda.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "s3:PutObject"
      ],
      "Resource": "arn:aws:s3:::YOUR_S3_BUCKET/*"
    },
    {
      "Effect": "Allow",
      "Action": "sts:GetCallerIdentity",
      "Resource": "*"
    }
  ]
}

# 作業ディレクトリを作成
mkdir -p lambda-layer/python
cd lambda-layer

# Lambda実行環境に合わせたパッケージをインストール
pip3 install \
  google-auth \
  google-api-python-client \
  --platform manylinux2014_x86_64 \
  --target=python/ \
  --python-version 3.12 \
  --only-binary=:all:

# zipファイルを作成
zip -r google-drive-layer.zip python/

# Lambdaレイヤーとして発行
aws lambda publish-layer-version \
  --layer-name google-drive-layer \
  --zip-file fileb://google-drive-layer.zip \
  --compatible-runtimes python3.12 \
  --region ap-northeast-1

https://drive.google.com/drive/folders/#この部分#

import io
import json
import os
import tempfile
import boto3
import google.auth
from googleapiclient.discovery import build
from googleapiclient.http import MediaIoBaseDownload

def get_google_credentials():
    """
    Workload Identity Federationの認証構成を使ってGoogle認証情報を取得する
    """
    # 環境変数からcredential configを読み取り、一時ファイルに書き出す
    cred_config = os.environ.get("GOOGLE_APPLICATION_CREDENTIALS_JSON")
    if not cred_config:
        raise ValueError("GOOGLE_APPLICATION_CREDENTIALS_JSON is not set")

    # 一時ファイルに認証構成を書き出す
    cred_file = os.path.join(tempfile.gettempdir(), "cred_config.json")
    with open(cred_file, "w") as f:
        f.write(cred_config)

    # 環境変数にファイルパスを設定してADCで読み込む
    os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = cred_file

    credentials, project = google.auth.default(
        scopes=["https://www.googleapis.com/auth/drive.readonly"]
    )
    return credentials

def list_files_in_folder(service, folder_id):
    """
    Google Driveの指定フォルダ内のファイル一覧を取得する
    """
    query = f"'{folder_id}' in parents and trashed = false"
    results = service.files().list(
        q=query,
        fields="files(id, name, mimeType)",
        includeItemsFromAllDrives=True,
        supportsAllDrives=True,
        corpora="allDrives",
    ).execute()

    return results.get("files", [])

def download_from_google_drive(service, file_id, file_name, mime_type):
    """
    Google Driveからファイルをダウンロードして一時ファイルパスと最終ファイル名を返す
    Google Workspace形式のファイルはエクスポートして変換する
    """
    # Google Workspace形式のMIMEタイプとエクスポート先の対応表
    export_mime_map = {
        "application/vnd.google-apps.spreadsheet": {
            "mime": "text/csv",
            "ext": ".csv",
        },
        "application/vnd.google-apps.document": {
            "mime": "application/pdf",
            "ext": ".pdf",
        },
        "application/vnd.google-apps.presentation": {
            "mime": "application/pdf",
            "ext": ".pdf",
        },
    }

    if mime_type in export_mime_map:
        # Google Workspace形式はexport_mediaでエクスポート
        export_info = export_mime_map[mime_type]
        request = service.files().export_media(
            fileId=file_id,
            mimeType=export_info["mime"],
        )
        # 拡張子を変換後のものに変更（例: レポート → レポート.csv）
        file_name = f"{file_name}{export_info['ext']}"
    else:
        # バイナリファイルはそのままダウンロード
        request = service.files().get_media(
            fileId=file_id,
            supportsAllDrives=True,
        )

    tmp_path = os.path.join(tempfile.gettempdir(), file_name)
    with open(tmp_path, "wb") as f:
        downloader = MediaIoBaseDownload(f, request)
        done = False
        while not done:
            _, done = downloader.next_chunk()

    return tmp_path, file_name

def upload_to_s3(bucket, key, file_path):
    """
    S3にファイルをアップロードする
    """
    s3_client = boto3.client("s3")
    s3_client.upload_file(file_path, bucket, key)

def lambda_handler(event, context):
    """
    メインハンドラ
    環境変数またはeventで指定したフォルダ内の全ファイルをS3に転送する
    """
    bucket = event.get("bucket", os.environ.get("S3_BUCKET"))
    s3_prefix = event.get("s3_prefix", "")
    folder_id = event.get("folder_id", os.environ.get("GOOGLE_DRIVE_FOLDER_ID"))

    if not bucket:
        return {
            "statusCode": 400,
            "body": json.dumps({"error": "bucket is required"})
        }

    if not folder_id:
        return {
            "statusCode": 400,
            "body": json.dumps({"error": "folder_id is required"})
        }

    try:
        # 1. Google認証情報を取得
        creds = get_google_credentials()
        service = build("drive", "v3", credentials=creds)

        # 2. フォルダ内のファイル一覧を取得
        files = list_files_in_folder(service, folder_id)

        transferred_files = []

        for file in files:
            # フォルダはスキップ
            if file["mimeType"] == "application/vnd.google-apps.folder":
                continue

            file_name = file["name"]

            # 3. Google Driveからファイルをダウンロード（Workspace形式は自動エクスポート）
            tmp_path, actual_file_name = download_from_google_drive(
                service, file["id"], file_name, file["mimeType"]
            )

            # 4. S3にアップロード
            s3_key = f"{s3_prefix}{actual_file_name}" if s3_prefix else actual_file_name
            upload_to_s3(bucket, s3_key, tmp_path)

            # 5. 一時ファイルを削除
            os.remove(tmp_path)
            transferred_files.append({"name": actual_file_name, "s3_key": s3_key})

        return {
            "statusCode": 200,
            "body": json.dumps({
                "message": f"{len(transferred_files)} file(s) transferred successfully",
                "files": transferred_files,
            })
        }

    except Exception as e:
        return {
            "statusCode": 500,
            "body": json.dumps({"error": str(e)})
        }

{
  "s3_prefix": "from-drive/"
}

{
  "statusCode": 200,
  "body": "{\"message\": \"1 file(s) transferred successfully\", \"files\": [{\"name\": \"test_s3_from_googledrive.txt\", \"s3_key\": \"from-drive/test_s3_from_googledrive.txt\"}]}"
}