はじめに

アプリケーションの中でAWS S3にデータをアップロードする、ダウンロードするなどの機能を実装する時に、署名付きURLを利用することがあると思います。
署名付きURLは、AWS S3のオブジェクトに対して一時的なアクセス権を付与するためのURLです。
このURLを使用すると、特定のオブジェクトに対して、指定した期間だけアクセスできるようになります。

これにより、バケットを公開することなく、特定のユーザーに対してオブジェクトへのアクセスを許可することができます。

ローカル環境での開発時にはMinIOが便利です。
MinIOは、AWS S3互換のオブジェクトストレージを提供するオープンソースのソフトウェアです。

今回は、MinIOを使用して署名付きURLを生成する方法について説明します。

MinIOの設定

docker composeを使用してMinIOを起動させるのが楽なので、設定を行います。

docker-compose.yml

services:
  minio:
    image: minio/minio:latest
    ports:
      - 9000:9000 # S3-compatible API
      - 9001:9001 # Web Console
    environment:
      # .envファイルからアクセスキーとシークレットキーを読み込む
      MINIO_ROOT_USER: ${AWS_ACCESS_KEY_ID}
      MINIO_ROOT_PASSWORD: ${AWS_SECRET_ACCESS_KEY}
    command: ["server", "/data", "--console-address", ":9001"]
    volumes:
      - minio_data:/data
    healthcheck:
      # MinIOサーバーが正常に起動したかを確認
      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
      interval: 3s
      timeout: 30s
      retries: 10

  mc:
    # このサービスはバケット作成のためだけに初回起動時に実行される想定
    profiles: ["bucket-setup"] # 通常の`up`では起動しないようにプロファイル指定
    image: minio/mc
    depends_on:
      minio:
        condition: service_healthy
    volumes:
      - minio_data:/data
    entrypoint: >
      /bin/sh -c "
      mc alias set my_minio http://minio:9000 ${AWS_ACCESS_KEY_ID} ${AWS_SECRET_ACCESS_KEY};
      mc mb --ignore-existing my_minio/${UPLOAD_BUCKET_NAME};
      mc anonymous set public my_minio/${UPLOAD_BUCKET_NAME};
      "

volumes:
  minio_data:

MinIOはAWS_ACCESS_KEY_IDとAWS_SECRET_ACCESS_KEYを使用して認証を行います。boto3などのライブラリを使用して、MinIOに接続する際には、これらの値を使用します。

また、MinIOはバケットを作成するために、mcコマンドを使用します。

mcはMinIOのコマンドラインツールで、AWS S3互換のAPIを使用して、バケットの作成やオブジェクトのアップロードなどを行うことができます。今回は、mcを使用して、バケットを作成しています。
パブリックアクセスを許可していますが、実際の運用では、セキュリティ上の理由から、パブリックアクセスを許可しない方が良いでしょう。

.env

PROJECT_NAME=minio-sample
PUBLIC_ENV=local

# MinIO/S3接続情報
AWS_ACCESS_KEY_ID=minioadmin
AWS_SECRET_ACCESS_KEY=minioadmin
AWS_ENDPOINT=http://localhost:9000
AWS_REGION=ap-northeast-1

# バケット名
UPLOAD_BUCKET_NAME=${PROJECT_NAME}-${PUBLIC_ENV}-upload

以下のようなコマンドでMinIOの設定を行います。

docker compose --profile bucket-setup up

--profile bucket-setupを指定することで、バケット作成用のサービスのみを起動します。
バケットが作成されたら、mcサービスは終了します。
その後、通常のサービスを起動します。

docker compose up -d

署名付きURLの生成

署名付きURLを生成するためには、boto3ライブラリを使用します。
通常はAWSサービスに接続するために、AWS_ACCESS_KEY_IDとAWS_SECRET_ACCESS_KEYを使用しますが、MinIOの場合は、これらの値を使用して接続します。
まずはクライアントを作成します

minio_client.py

import boto3
import os
from dotenv import load_dotenv


# .envファイルから環境変数を読み込む
load_dotenv()


# MinIO接続情報
AWS_ACCESS_KEY_ID = os.getenv("AWS_ACCESS_KEY_ID")
AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY")
AWS_ENDPOINT = os.getenv("AWS_ENDPOINT")
AWS_REGION = os.getenv("AWS_REGION")  # boto3に渡すがMinIO自体はリージョンレス


def get_minio_client():
    """MinIOクライアントを取得する関数"""
    if not all([AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_ENDPOINT, AWS_REGION]):
        print(
            "エラー: 環境変数 (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_ENDPOINT, AWS_REGION) が設定されていません。"
        )
        return None

    try:
        client = boto3.client(
            "s3",
            aws_access_key_id=AWS_ACCESS_KEY_ID,
            aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
            endpoint_url=AWS_ENDPOINT,
            region_name=AWS_REGION,
            config=boto3.session.Config(signature_version="s3v4"),
        )

        # 接続テスト
        client.list_buckets()
        print("MinIOへの接続に成功しました。")
        return client
    except Exception as e:
        print(f"MinIOクライアント作成中に予期せぬエラーが発生しました: {e}")
        return None


if __name__ == "__main__":
    print("MinIOクライアント接続テストを開始します...")
    minio_client = get_minio_client()
    if minio_client:
        print("テスト完了: MinIO client is ready.")
    else:
        print("テスト完了: Failed to create MinIO client.")

コマンドを実行して、MinIOに接続できるか確認します。

❯ uv run python minio_client.py
MinIOクライアント接続テストを開始します...
MinIOへの接続に成功しました。
テスト完了: MinIO client is ready.

接続できたら、次にアップロード用の署名付きURLを生成します。

presigned_put.py

from botocore.exceptions import ClientError


def generate_presigned_put_url(client, bucket_name, object_name, expiration=3600):
    """ファイルアップロード(PUT)用の署名付きURLを生成する関数"""
    if not bucket_name:
        print("エラー: バケット名が設定されていません。")
        return None
    if not client:
        print("エラー: MinIOクライアントが提供されていません。")
        return None

    try:
        response = client.generate_presigned_url(
            ClientMethod="put_object",
            Params={
                "Bucket": bucket_name,
                "Key": object_name,
            },
            ExpiresIn=expiration,
            HttpMethod="PUT",
        )
        print(
            f"アップロード用署名付きURLを生成しました: Key={object_name}, Bucket={bucket_name}"
        )
        return response
    except ClientError as e:
        print(
            f"署名付きPUT URLの生成に失敗しました (Bucket: {bucket_name}, Key: {object_name}): {e}"
        )
        return None
    except Exception as e:
        print(f"予期せぬエラー (PUT URL生成): {e}")
        return None

次に、ダウンロード用の署名付きURLを生成します。

presigned_get.py

from botocore.exceptions import ClientError


def generate_presigned_get_url(client, bucket_name, object_name, expiration=3600):
    """ファイルダウンロード(GET)用の署名付きURLを生成する関数"""
    if not bucket_name:
        print("エラー: バケット名が設定されていません。")
        return None
    if not client:
        print("エラー: MinIOクライアントが提供されていません。")
        return None

    try:
        response = client.generate_presigned_url(
            ClientMethod="get_object",
            Params={"Bucket": bucket_name, "Key": object_name},
            ExpiresIn=expiration,
            HttpMethod="GET",
        )
        print(
            f"ダウンロード用署名付きURLを生成しました: Key={object_name}, Bucket={bucket_name}"
        )
        return response
    except ClientError as e:
        print(
            f"署名付きGET URLの生成に失敗しました (Bucket: {bucket_name}, Key: {object_name}): {e}"
        )
        return None
    except Exception as e:
        print(f"予期せぬエラー (GET URL生成): {e}")
        return None

それらを組み合わせて、署名付きURLを生成するメインのスクリプトを作成します。

main.py

import os
import requests
import time
from minio_client import get_minio_client
from presigned_put import generate_presigned_put_url
from presigned_get import generate_presigned_get_url


UPLOAD_BUCKET_NAME = os.getenv("UPLOAD_BUCKET_NAME")
TEST_OBJECT_KEY = f"e2e-test/test-{int(time.time())}.txt"
LOCAL_UPLOAD_FILE = "upload_test.txt"
LOCAL_DOWNLOAD_FILE = "download_test.txt"
FILE_CONTENT = "これは署名付きURLのテスト用ファイルです。"
PUT_URL_EXPIRATION = 600
GET_URL_EXPIRATION = 300


def run_e2e_test():
    """署名付きURLを使ったE2Eテストを実行"""
    print("--- E2Eテスト開始 ---")

    # === ステップ1: MinIOクライアント取得 ===
    print("\n[ステップ1] MinIOクライアントを取得中...")
    minio_client = get_minio_client()
    if not minio_client:
        print("テスト中止: MinIOクライアントを取得できませんでした。")
        return
    if not UPLOAD_BUCKET_NAME:
        print("テスト中止: バケット名(UPLOAD_BUCKET_NAME)が設定されていません。")
        return
    print(f"使用するバケット: {UPLOAD_BUCKET_NAME}")

    # === ステップ2: テスト用ローカルファイル作成 ===
    print(f"\n[ステップ2] テスト用ファイル ({LOCAL_UPLOAD_FILE}) を作成中...")
    try:
        with open(LOCAL_UPLOAD_FILE, "w", encoding="utf-8") as f:
            f.write(FILE_CONTENT)
        print("テスト用ファイルを作成しました。")
    except IOError as e:
        print(f"テスト中止: テスト用ファイルの作成に失敗しました: {e}")
        return

    # === ステップ3: アップロード用署名付きURL発行 ===
    print(f"\n[ステップ3] オブジェクト '{TEST_OBJECT_KEY}' のPUT URLを発行中...")
    put_url = generate_presigned_put_url(
        minio_client, UPLOAD_BUCKET_NAME, TEST_OBJECT_KEY, expiration=PUT_URL_EXPIRATION
    )
    if not put_url:
        print("テスト中止: PUT URLの発行に失敗しました。")
        cleanup_local_files()
        return
    print(f"PUT URL発行成功 (有効期間: {PUT_URL_EXPIRATION}秒)")

    # === ステップ4: ファイルをアップロード ===
    print(
        f"\n[ステップ4] ファイル '{LOCAL_UPLOAD_FILE}' を署名付きURLでアップロード中..."
    )
    try:
        print(f"PUTリクエスト送信先: {put_url[:80]}...")  # URLが長いので一部表示
        with open(LOCAL_UPLOAD_FILE, "rb") as data:
            headers = {"Content-Type": "text/plain"}  # ContentTypeを指定
            response = requests.put(put_url, data=data, headers=headers)
            response.raise_for_status()  # ステータスコードが200番台でなければ例外発生
        print(f"アップロード成功！ ステータスコード: {response.status_code}")
    except requests.exceptions.RequestException as e:
        print(f"テスト中止: アップロード中にエラーが発生しました: {e}")
        if hasattr(e, "response") and e.response is not None:
            print(f"レスポンス内容: {e.response.text}")
        cleanup_local_files()
        return
    except Exception as e:
        print(f"テスト中止: アップロード中に予期せぬエラー: {e}")
        cleanup_local_files()
        return

    # === ステップ5: ダウンロード用署名付きURL発行 ===
    print(f"\n[ステップ5] オブジェクト '{TEST_OBJECT_KEY}' のGET URLを発行中...")
    get_url = generate_presigned_get_url(
        minio_client, UPLOAD_BUCKET_NAME, TEST_OBJECT_KEY, expiration=GET_URL_EXPIRATION
    )
    if not get_url:
        print("テスト中止: GET URLの発行に失敗しました。")
        cleanup_local_files()
        return
    print(f"GET URL発行成功 (有効期間: {GET_URL_EXPIRATION}秒)")

    # === ステップ6: ファイルをダウンロード ===
    print(f"\n[ステップ6] ファイルを署名付きURLでダウンロード中...")
    try:
        print(f"GETリクエスト送信先: {get_url[:80]}...")
        response = requests.get(get_url, stream=True)
        response.raise_for_status()
        with open(LOCAL_DOWNLOAD_FILE, "wb") as f:
            for chunk in response.iter_content(chunk_size=8192):
                f.write(chunk)
        print(
            f"ダウンロード成功！ ファイル '{LOCAL_DOWNLOAD_FILE}' として保存しました。"
        )
    except requests.exceptions.RequestException as e:
        print(f"テスト中止: ダウンロード中にエラーが発生しました: {e}")
        if hasattr(e, "response") and e.response is not None:
            print(f"レスポンス内容: {e.response.text}")
        cleanup_local_files()
        return
    except Exception as e:
        print(f"テスト中止: ダウンロード中に予期せぬエラー: {e}")
        cleanup_local_files()
        return

    # === ステップ7: 内容検証 (任意) ===
    print(f"\n[ステップ7] アップロード内容とダウンロード内容を比較中...")
    try:
        with open(LOCAL_DOWNLOAD_FILE, "r", encoding="utf-8") as f:
            downloaded_content = f.read()

        if downloaded_content == FILE_CONTENT:
            print(
                "検証成功: ダウンロードしたファイルの内容は、アップロードしたものと一致します。"
            )
        else:
            print("検証失敗: ダウンロードしたファイルの内容が異なります。")
            print(f"  期待値: '{FILE_CONTENT}'")
            print(f"  実際値: '{downloaded_content}'")
    except IOError as e:
        print(f"検証エラー: ダウンロードファイルの読み込みに失敗しました: {e}")
    except Exception as e:
        print(f"検証中に予期せぬエラー: {e}")

    # === 後処理 ===
    cleanup_local_files()
    print("\n--- E2Eテスト終了 ---")


def cleanup_local_files():
    """テストで使用したローカルファイルを削除する"""
    print("後処理: ローカルテストファイルを削除します...")
    for f in [LOCAL_UPLOAD_FILE, LOCAL_DOWNLOAD_FILE]:
        if os.path.exists(f):
            try:
                os.remove(f)
                print(f" - {f} を削除しました。")
            except OSError as e:
                print(f" - {f} の削除に失敗しました: {e}")


if __name__ == "__main__":
    run_e2e_test()

実行結果は以下のようになります。

❯ uv run python main.py
--- E2Eテスト開始 ---

[ステップ1] MinIOクライアントを取得中...
MinIOへの接続に成功しました。
使用するバケット: minio-sample-local-upload

[ステップ2] テスト用ファイル (upload_test.txt) を作成中...
テスト用ファイルを作成しました。

[ステップ3] オブジェクト 'e2e-test/test-1745160645.txt' のPUT URLを発行中...
アップロード用署名付きURLを生成しました: Key=e2e-test/test-1745160645.txt, Bucket=minio-sample-local-upload
PUT URL発行成功 (有効期間: 600秒)

[ステップ4] ファイル 'upload_test.txt' を署名付きURLでアップロード中...
PUTリクエスト送信先: http://localhost:9000/minio-sample-local-upload/e2e-test/test-1745160645.txt?X-A...
アップロード成功！ ステータスコード: 200

[ステップ5] オブジェクト 'e2e-test/test-1745160645.txt' のGET URLを発行中...
ダウンロード用署名付きURLを生成しました: Key=e2e-test/test-1745160645.txt, Bucket=minio-sample-local-upload
GET URL発行成功 (有効期間: 300秒)

[ステップ6] ファイルを署名付きURLでダウンロード中...
GETリクエスト送信先: http://localhost:9000/minio-sample-local-upload/e2e-test/test-1745160645.txt?X-A...
ダウンロード成功！ ファイル 'download_test.txt' として保存しました。

[ステップ7] アップロード内容とダウンロード内容を比較中...
検証成功: ダウンロードしたファイルの内容は、アップロードしたものと一致します。
後処理: ローカルテストファイルを削除します...
 - upload_test.txt を削除しました。
 - download_test.txt を削除しました。

--- E2Eテスト終了 ---

まとめ

MinIOを使用して、署名付きURLを生成する方法について説明しました！
署名付きURLを使用することで、特定のオブジェクトに対して一時的なアクセス権を付与することができます。

MinIOはを利用すると、ローカル環境での開発やテストに非常に便利です。

サンプルコードも置いておくので、ぜひ活用してみてください！