🤺

S3の署名付きurlを自動で無効化したい

2023/11/20に公開

はじめに

S3の署名付きURL機能を使用することで、S3のブロックパブリックアクセスを維持したまま、ファイルを特定のユーザのみに公開することができます。またAWSのIAMユーザをもたないユーザに対して一時的にファイルアップロードを可能にすることもできます。
しかし、万が一署名付きURLが漏洩してしまった場合どのような対処をするべきか考えたことはありますでしょうか。
今回はS3の署名付きURLを事実上無効化する為の方法を共有いたします。

忙しい人向けのまとめ

有効期限を迎える前に署名付きURLそのものを無効化したり、削除することはできない。代替案としてバケットポリシーを工夫して署名付きURLからのアクセスを無効化することで対応できる
アップロード検知後に無効化したい場合はS3イベント通知でLambdaを実行する
ダウンロード検知後に無効化したい場合はCloudWawtch logsのサブスクリプションフィルター経由でLambdaを実行する(S3イベント通知では検知できないことに注意)

やってみる

今回は以下のようなケースを想像しています。

AWS管理コンソールにログインできない(IAMユーザを持たない)ユーザにS3のファイルダウンロードとファイルアップロードを許可したい。
だが、署名付きURLの漏洩をできる限り防ぐために、ユーザがファイルダウンロード・アップロードしたら該当の署名付きURLを無効化したい

署名付きURLを発行するLambda関数の作成

ダウンロード用の署名付きURLは管理コンソールからの操作でも作成可能ですが、アップロード用の署名付きURLは管理コンソールからは作成することはできません。別途SDKなどを使用してプログラムから呼び出す必要があります。
参考:署名付き URL を使用したオブジェクトのアップロード

そのため、署名付きURLを取得してSNS経由で通知するLambda関数を用意しちゃいます。

ソースコードはこちら(折りたたんであります)

import boto3
import os
import json
from botocore.exceptions import ClientError
from datetime import datetime
import pytz

def lambda_handler(event, context):
    # 環境変数からS3バケット名とSNSトピック名を取得
    bucket_name = os.environ['S3_BUCKET_NAME']
    sns_topic_name = os.environ['SNS_TOPIC_NAME']

    # JSTタイムゾーンを設定
    jst = pytz.timezone('Asia/Tokyo')

    # 現在の日時をJSTで取得してフォーマット（例：TestFile_20211118123045）
    timestamp = datetime.now(jst).strftime("TestFile_%Y%m%d%H%M%S")
    object_key = timestamp

    # S3クライアントの生成
    s3_client = boto3.client('s3')
    # S3バケットポリシーを空にする
    try:
        s3_client.delete_bucket_policy(Bucket=bucket_name)
    except ClientError as e:
        return {
            'statusCode': 500,
            'body': json.dumps(f"Error clearing bucket policy: {str(e)}")
        }
    try:
        # アップロード用の署名付きURLの生成
        presigned_url = s3_client.generate_presigned_url('put_object',
                                                         Params={'Bucket': bucket_name, 'Key': object_key},
                                                         ExpiresIn=3600)
    except ClientError as e:
        return {
            'statusCode': 500,
            'body': json.dumps(f"Error generating presigned URL: {str(e)}")
        }

    # SNSクライアントの生成
    sns_client = boto3.client('sns')

    try:
        # SNSを通じて署名付きURLをメールで通知
        sns_client.publish(
            TopicArn=f"arn:aws:sns:{context.invoked_function_arn.split(':')[3]}:{context.invoked_function_arn.split(':')[4]}:{sns_topic_name}",
            Message=f"Here is your presigned URL for uploading to S3: {presigned_url}",
            Subject='S3 Upload Presigned URL Notification'
        )
    except ClientError as e:
        return {
            'statusCode': 500,
            'body': json.dumps(f"Error sending notification via SNS: {str(e)}")
        }

    return {
        'statusCode': 200,
        'body': json.dumps('Presigned URL generated and notified successfully')
    }

※こちらのソースコードは生成系AI(ChatGPT)を使用して作成しています。業務での使用を検討されている方は十分注意してください

各ソースコードの処理概要は以下です

S3バケットポリシーを空にする
※後述の処理で署名付きURLを拒否するバケットポリシーを使用するため
アップロード用の署名付きURLの作成
※ユーザが署名付きURLでアップロードしたときのファイル名は"TestFile_yyyymmddhhmmss"で固定
ダウンロード用の署名付きURLの作成
※ダウンロードファイルは"test.txt"で固定
SNSへのメッセージ送付

署名付きURLを事実上無効化するS3バケットポリシーの作成

忙しい人向けのまとめでも記載しましたが、一度発行してしまった署名付きURLを有効期限前に無効化することは残念ながらできません。そこでS3のバケットポリシーを工夫して、署名付きURLを使用した場合はS3にアクセスできないようにすることで事実上署名付きURLを無効化する方針で対応したいと思います。

Amazon S3 Signature Version 4 Authentication Specific Policy Keys
のドキュメントにも記載がある通りS3のバケットポリシーの条件キーとして"s3:authType"というものがあります。
そして、Authenticating Requests (AWS Signature Version 4)のドキュメントにも記載がある通り、署名付きURLを指定してS3にアクセスする際はAuthTypeとしてQuery string parametersを使用することがわかります。詳しい解説はClassmethodさんが署名付き URL (presigned URL) のときだけ IAM ポリシーで条件を制御してみたの記事にまとめていますので参照してみてください

以上より設定するべきバケットポリシーは以下になります

BucketPolicyはこちら(折りたたんであります)

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "DenyNonRESTHeaderAuthAndSpecificUser",
            "Effect": "Deny",
            "Principal": "*",
            "Action": [
                "s3:GetObject"
            ],
            "Resource": [
                "arn:aws:s3:::test-for-presigined-url",
                "arn:aws:s3:::test-for-presigined-url/*"
            ],
            "Condition": {
                "StringNotEquals": {
                    "aws:username": "handsonforbeginners"
                },
                "ForAnyValue:StringEquals": {
                    "s3:authType": "REST-QUERY-STRING"
                }
            }
        }
    ]
}

こちらは署名付きURLによるGetObject(つまりダウンロード)を拒否、IAMユーザ(今回の場合はhandsonforbeginners)によるダウンロードは継続して許可するようなバケットポリシーを設定しています

バケットポリシーを更新するLambdaの作成(ダウンロード拒否)

上記のバケットポリシーに自動で書き換えるためにLambdaを作成します。

ソースコードはこちら(折りたたんであります)

import boto3
import os
import json

def lambda_handler(event, context):
    # 環境変数からS3バケット名、SSMパラメータ名、SNSトピック名を取得
    bucket_name = os.environ['S3_BUCKET_NAME']
    ssm_parameter_name = os.environ['SSM_PARAMETER_NAME']
    sns_topic_name = os.environ['SNS_TOPIC_NAME']

    # SSMクライアントを初期化
    ssm_client = boto3.client('ssm')

    # SSMからバケットポリシーを取得
    parameter = ssm_client.get_parameter(Name=ssm_parameter_name, WithDecryption=True)
    bucket_policy = parameter['Parameter']['Value']

    # S3クライアントを初期化
    s3_client = boto3.client('s3')

    # S3バケットポリシーを更新
    s3_client.put_bucket_policy(Bucket=bucket_name, Policy=bucket_policy)

    # SNSクライアントを初期化
    sns_client = boto3.client('sns')

    # SNSトピックARNを取得
    response = sns_client.list_topics()
    topic_arn = next(topic['TopicArn'] for topic in response['Topics'] if sns_topic_name in topic['TopicArn'])

    # SNSを使用して通知を送信
    message = "オブジェクトに対する処理が確認されたため、バケットポリシーを修正し、署名付きURLからのアクセスを無効化しました。"
    sns_client.publish(TopicArn=topic_arn, Message=message)

    return {
        'statusCode': 200,
        'body': json.dumps('S3 bucket policy updated and notification sent')
    }

※こちらのソースコードは生成系AI(ChatGPT)を使用して作成しています。業務での使用を検討されている方は十分注意してください
各ソースコードの処理概要は以下です

SSMパラメータストアからバケットポリシーを取得
※今回は更新するバケットポリシーの内容をパラメータストアに保存しています
S3のバケットポリシーを更新
SNS経由で更新した旨を通知

CloudTrailのログ出力設定

S3のイベントとの連携方法として代表的なものはS3イベント通知ですが、残念ながらS3:GetObjectは対応していません。回避策として、CloudTrailのデータイベントを取得し、そこにS3:GetObjectのイベントがログとして出力されたら上記のLambdaを実行するように設定します。
以下のようにCloudTrailの設定でS3のデータイベントをCloudWatchLogsへ出力します

CloudWatch Logsサブスクリプションフィルターの設定

サブスクリプションフィルターを使用することで、CloudWatchLogsに特定のログが出力されたらLambdaを実行することが可能です。
サブスクリプションフィルターで以下のように設定します。

※フィルターパターンを"{ $.eventName = "GetObject" }"とすることでGetObjectイベントを検知できます
これでファイルダウンロードが行われたらLambdaが自動でS3のバケットポリシーを上書きするので、(ポリシーを手動で元に戻さない限り)署名付きURL経由でのダウンロードはできなくなります

バケットポリシーを更新するLambdaの作成(アップロード拒否)

ダウンロード同様に、アップロードを検知したら自動でバケットポリシーを上書きするLambdaを作成します。
Lambdaのソースコード自体は先ほどと同様です。SSMパラメータストアに保存するバケットポリシーだけ以下のようにGetObjectとPutObject両方を拒否するようにしてください

BucketPolicyはこちら(折りたたんであります)

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "DenyNonRESTHeaderAuthAndSpecificUser",
            "Effect": "Deny",
            "Principal": "*",
            "Action": [
                "s3:GetObject","s3:PutObject"
            ],
            "Resource": [
                "arn:aws:s3:::test-for-presigined-url",
                "arn:aws:s3:::test-for-presigined-url/*"
            ],
            "Condition": {
                "StringNotEquals": {
                    "aws:username": "handsonforbeginners"
                },
                "ForAnyValue:StringEquals": {
                    "s3:authType": "REST-QUERY-STRING"
                }
            }
        }
    ]
}

S3イベント通知の設定

PutObjectについてはS3のイベント通知機能を使用できます。以下のように設定することで、ユーザがファイルをアップロードしたら上記のLambdaが自動でS3バケットポリシーを上書きするので、署名付きURLを経由したファイルアップロードは不可になります。

動作の確認

署名付きURLの発行

まず、署名付きURLを発行します。
署名付きURLを発行するLambda関数の作成で作成したLambdaのテストボタンを押下します
※事前にSNSを設定しメール送付先を登録しておいて下さい

以下のようにダウンロード・アップロード用の署名付きURLが届いていると思います

またS3のバケットポリシーも空になっていることがわかります

ダウンロード

署名付きURLを使用してファイルの内容を取得してみましょう

しばらくするとメールが届き、S3のバケットポリシーが変更されていることがわかります

また、再度同様の署名付きURLを使用してファイルを取得しようとするとアクセス拒否となることがわかります

アップロード

署名付きURLを使用してファイルをアップロードしてみましょう
Powershellの場合以下のようなコマンドでファイルをアップロードすることができます
$filePath = "test.txt"
$signedUrl = "署名付きURL"
Invoke-WebRequest -Uri $signedUrl -Method Put -InFile $filePath