green snake
Photo by Pixabay on Pexels.com

FastAPIで安全にファイルアップロードを実装する:UploadFile・サイズ制限・ウイルス対策・S3互換ストレージ・署名付きURLの実務パターン


要約(最初に全体像)

  • FastAPIのファイルアップロードは UploadFile を使うとメモリ効率が良く、実務の土台になります。
  • 安全にする鍵は「サイズ制限」「拡張子ではなくMIME/実体の検査」「保存先の分離」「ファイル名の無害化」「権限チェック」「ログと監査」です。
  • 本番では、APIサーバがファイル本体を配り続けない設計が安定します。S3互換ストレージに置き、APIは署名付きURLを発行する役に寄せるのが定番です。
  • 画像のサムネイル生成やPDF変換などの重い処理は、アップロード後にバックグラウンドジョブへ渡すとレスポンスが軽くなります。
  • テストでは「成功」「サイズ超過」「不正MIME」「権限なし」「署名URLの期限切れ」などを最小セットで守ると、事故が減ります。

誰が読んで得をするか

  • 個人開発・学習者の方:プロフィール画像や添付ファイルを付けたいけれど、どこまで対策すればよいか迷っている方。
  • 小規模チームの方:管理画面の添付やCSVアップロードが増え、サイズ制限や拡張子偽装などの不安が出てきた方。
  • SaaS開発チームの方:マルチインスタンス運用でファイル配信が重く、ストレージ分離や署名付きURL、監査ログを整えたい方。

アクセシビリティ評価

  • 見出しを細かく分け、手順を番号付きで提示しています。必要な情報を探しやすい構成です。
  • 専門語は初出で短く補足し、同じ用語を繰り返して混乱を減らしています。
  • コードは短いブロックに分け、コメントは必要最低限にしています。
  • 目標レベルはAA相当です。

1. よくある失敗パターン:まず「危ない形」を知る

ファイルアップロードは、動かすだけなら簡単です。でも、実務で事故になりやすいポイントがたくさんあります。

  • サイズ無制限で受け取ってしまい、メモリやディスクが枯渇する
  • 拡張子だけで判定し、実体が違うファイル(例:.jpgに見せた実行形式)を通してしまう
  • ユーザーの指定したファイル名をそのまま保存して、パス・トラバーサルや文字化け、意図しない上書きを招く
  • APIサーバがファイル配信まで担当してしまい、ピーク時にCPU/ネットワークが詰まる
  • 認可(誰がどのファイルを見てよいか)が曖昧で、他人の添付が見える

この記事は、このあたりを一つずつ丁寧に避ける「型」を作ることを目標にします。


2. 最小のアップロード:UploadFileを使う理由

FastAPIのアップロードは、UploadFile を基本にすると扱いやすいです。bytes で受け取る方法もありますが、サイズが大きいとメモリを圧迫しやすくなります。

# app/routers/uploads.py
from fastapi import APIRouter, UploadFile, File

router = APIRouter(prefix="/uploads", tags=["uploads"])

@router.post("")
async def upload(file: UploadFile = File(...)):
    return {"filename": file.filename, "content_type": file.content_type}

ここで押さえておきたいことは2つです。

  • file.filename は信用しない(表示用の参考にはできるが保存名には使わない)
  • file.content_type も完全には信用しない(クライアントが偽装できる)

安全にするには、この先のバリデーションと保存戦略が重要になります。


3. サイズ制限:アプリとプロキシの二段構え

サイズ制限は、APIで頑張る前に「入口で落とす」ほど効果が高いです。

3.1 リバースプロキシ(Nginxなど)での制限

本番でNginxを使うなら、まず client_max_body_size で上限を決めます。これで巨大リクエストがアプリまで届きにくくなります。

3.2 アプリ側での制限(読み取り中に止める)

プロキシだけだと環境差が出ることもあるので、アプリ側でも制限を入れておくと安心です。UploadFile のストリームを小分けに読みながら、一定サイズを超えたらエラーにします。

# app/services/upload_validator.py
from fastapi import HTTPException, status, UploadFile

MAX_BYTES = 10 * 1024 * 1024  # 10MB

async def enforce_size_limit(file: UploadFile) -> bytes:
    total = 0
    chunks: list[bytes] = []

    while True:
        chunk = await file.read(1024 * 1024)  # 1MBずつ
        if not chunk:
            break
        total += len(chunk)
        if total > MAX_BYTES:
            raise HTTPException(
                status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
                detail="file too large",
            )
        chunks.append(chunk)

    # この例は最後にまとめていますが、実務ではディスク/ストレージに逐次書き込みが推奨です
    return b"".join(chunks)

上の例は「考え方」を見せるために最後に結合しています。実務では、次の章のように逐次保存する設計がより安全です。


4. ファイル名の無害化:保存名はサーバ側で発行する

保存名は、ユーザーが指定したものではなく、サーバ側で一意なIDを発行して使うのが基本です。

  • 表示名:ユーザーのファイル名(必要ならDBに保存)
  • 実体名:UUIDなどで生成した安全なキー

例として、オブジェクトキーを uploads/{user_id}/{uuid}.{ext} のようにします。

# app/services/upload_naming.py
import uuid
from pathlib import Path

ALLOWED_EXT = {"png", "jpg", "jpeg", "pdf"}

def safe_object_key(user_id: int, original_filename: str) -> str:
    ext = Path(original_filename).suffix.lower().lstrip(".")
    if ext not in ALLOWED_EXT:
        ext = "bin"
    uid = uuid.uuid4().hex
    return f"uploads/{user_id}/{uid}.{ext}"

ここでのポイントは、拡張子は「参考」程度に扱い、後段で実体検査をすることです。


5. 種類の検証:拡張子ではなく実体を見る

セキュリティ上の基本として、拡張子だけで許可・拒否を決めない方が安全です。

5.1 最低限のMIMEチェック

まずは content_type を見て弾くのは有効ですが、偽装は可能です。ここは「第一関門」として使います。

# app/services/content_type.py
from fastapi import HTTPException, status, UploadFile

ALLOWED_CT = {
    "image/png",
    "image/jpeg",
    "application/pdf",
}

def enforce_content_type(file: UploadFile) -> None:
    if file.content_type not in ALLOWED_CT:
        raise HTTPException(
            status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
            detail="unsupported content type",
        )

5.2 実体検査(簡易)

本格的にはファイルシグネチャ(マジックナンバー)検査を入れたいところですが、環境により使うライブラリが変わります。ここでは「最小の自前チェック」例だけ示します。

# app/services/magic_check.py
from fastapi import HTTPException, status

def looks_like_png(head: bytes) -> bool:
    return head.startswith(b"\x89PNG\r\n\x1a\n")

def looks_like_jpeg(head: bytes) -> bool:
    return head.startswith(b"\xff\xd8\xff")

def looks_like_pdf(head: bytes) -> bool:
    return head.startswith(b"%PDF")

def enforce_magic(head: bytes, content_type: str) -> None:
    ok = False
    if content_type == "image/png":
        ok = looks_like_png(head)
    elif content_type == "image/jpeg":
        ok = looks_like_jpeg(head)
    elif content_type == "application/pdf":
        ok = looks_like_pdf(head)

    if not ok:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="file content does not match content-type",
        )

実務では、より堅牢な判定ライブラリを使うことも多いです。ただ、どれを採用するにせよ「拡張子だけで判断しない」という姿勢が重要です。


6. 保存戦略:ローカル保存とS3互換ストレージの使い分け

6.1 ローカル保存(学習・小規模向け)

単一サーバで運用し、ファイルが小さめならローカル保存も成立します。ただし、複数インスタンスにすると共有が難しくなります。

# app/services/local_storage.py
from pathlib import Path
from fastapi import UploadFile

BASE = Path("./data")

async def save_to_local(path: str, file: UploadFile) -> None:
    full = BASE / path
    full.parent.mkdir(parents=True, exist_ok=True)

    with full.open("wb") as f:
        while True:
            chunk = await file.read(1024 * 1024)
            if not chunk:
                break
            f.write(chunk)

6.2 S3互換ストレージ(本番向けの定番)

本番では、ストレージをAPIサーバから分離すると安定します。S3互換ストレージに保存し、APIは「アップロードの受付」と「参照URLの発行」に寄せるのが扱いやすいです。

考え方としては次のどちらかです。

  1. API経由アップロード:APIが受け取り、ストレージに転送する
  2. 直接アップロード:APIは署名付きURLを発行し、ブラウザがストレージへ直接PUTする(負荷が軽い)

運用が大きくなるほど、2が強いです。次章で署名付きURLのパターンを紹介します。


7. 署名付きURL:APIサーバを配信係から外す

署名付きURLは「一定時間だけ有効な、限定的なアクセスURL」を発行する仕組みです。

  • アップロード用URL(PUT)
  • ダウンロード用URL(GET)

この2つを分けると、権限管理が明確になります。

7.1 ざっくり設計

  • POST /files/presign-upload:アップロード先URLとキーを返す
  • GET /files/{file_id}/download:ダウンロード用URLを返す
  • DBには file_id, owner_id, object_key, content_type, size, original_name などを保存

FastAPI側のレスポンス例(あくまで形の例):

from pydantic import BaseModel

class PresignUploadResponse(BaseModel):
    file_id: str
    object_key: str
    upload_url: str
    expires_in: int

7.2 セキュリティ上の要点

  • URLの有効期限は短く(数分〜十数分)
  • ダウンロードURLの発行時は必ず「所有者」や「権限」をチェック
  • Content-Type やサイズを条件に含められるなら含める(ストレージ側での条件付与)

クラウドごとのSDKで実装が変わるため、ここでは「API設計とチェック項目」を中心に押さえておくのがおすすめです。


8. ウイルス対策とサニタイズ:現実的な落としどころ

ファイルを受け入れる以上、悪意あるファイルが混ざる可能性はゼロではありません。

  • 外部に公開するサービスほど、ウイルススキャンやサンドボックスを検討する価値が上がります
  • ただし、最初から完璧にやろうとすると重くなるので、段階導入が現実的です

段階的な案:

  1. サイズ制限・種類制限・保存名の無害化(必須)
  2. 画像なら再エンコード(アップロードされた画像をそのまま配らない)
  3. スキャン基盤(ClamAVなど)を非同期で回す
  4. 検疫(quarantine)バケットを用意し、スキャンOKのものだけ本番バケットへ移す

ここでのコツは「アップロード直後に即公開しない」ことです。とくに外部公開の添付は、検疫フローがあるだけで安全性が上がります。


9. 画像サムネイルなどの後処理:バックグラウンドへ渡す

画像サムネイル生成、PDFのプレビュー生成、OCRなどは、同期レスポンスに混ぜない方が快適です。

流れの例:

  1. アップロード受付 → DBに status=processing を保存
  2. ジョブキューへ file_id を投入
  3. ワーカーで処理 → status=ready に更新、派生ファイルのキーも保存
  4. フロントは status を見て表示を切り替える

既に扱った Celery/Redis のパターンと相性が良い領域です。


10. APIの実装例:安全なアップロードの最小形

ここまでの部品をまとめ、ローカル保存の最小例を作ります。実務ではこの部分をストレージ実装に差し替えます。

# app/routers/uploads.py
from fastapi import APIRouter, UploadFile, File, Depends
from app.services.content_type import enforce_content_type
from app.services.magic_check import enforce_magic
from app.services.upload_naming import safe_object_key
from app.services.local_storage import save_to_local

router = APIRouter(prefix="/uploads", tags=["uploads"])

def get_current_user_id() -> int:
    # 実際はJWTなどから取得
    return 1

@router.post("")
async def upload_file(
    file: UploadFile = File(...),
    user_id: int = Depends(get_current_user_id),
):
    enforce_content_type(file)

    head = await file.read(16)
    enforce_magic(head, file.content_type)

    # 先頭を読んだので、保存時に先頭も書き込む必要がある
    object_key = safe_object_key(user_id, file.filename)

    # 先頭を書いてから残りを逐次保存
    # local_storage.save_to_local は file.read で続きから読むので、このパターンで整合します
    # 先頭分は別途書き足します
    from pathlib import Path
    base = Path("./data")
    full = base / object_key
    full.parent.mkdir(parents=True, exist_ok=True)

    with full.open("wb") as f:
        f.write(head)
        while True:
            chunk = await file.read(1024 * 1024)
            if not chunk:
                break
            f.write(chunk)

    return {
        "file_id": object_key,  # 本当はDB採番IDを返すのが良いです
        "content_type": file.content_type,
    }

この例で守れていること:

  • content_type の第一関門
  • マジックナンバーによる簡易実体チェック
  • 保存名をサーバ側で生成
  • ストリームで逐次書き込み

これに「サイズ制限」「DBでの所有者管理」「ダウンロード時の認可」「署名付きURL」などを足していくと、実務の形に近づきます。


11. ダウンロード設計:認可が最重要

ダウンロードはアップロード以上に「誰が見られるか」が重要です。

  • file_id からDBで owner_id を引く
  • 現在ユーザーが owner_id と一致するか、または管理者権限かを確認
  • S3互換ストレージなら署名付きGET URLを発行して返す
  • ローカルなら FileResponse を返すが、APIサーバ負荷が増える点に注意

「ファイルIDさえ知っていれば取れる」状態は事故になりやすいので、ここは必ずテストで守るのがおすすめです。


12. テストの最小セット:事故を減らす5本

最初に揃えると効果が大きいテストです。

  • 正常:許可タイプ+小さいサイズ → 200/201
  • サイズ超過:上限を超える → 413
  • 不正MIME:許可外のcontent-type → 415
  • 偽装:content-typeは許可だがマジック不一致 → 400
  • 認可:他人の file_id でダウンロードできない → 403/404

テストの数は少なくても、意図が明確なものを置くと守りが強くなります。


13. 導入ロードマップ(少しずつで大丈夫)

  1. UploadFile で最小アップロードを作る
  2. サイズ制限、保存名の無害化、content-typeの制限を入れる
  3. 実体検査を追加(簡易でよいので「偽装が通らない」状態へ)
  4. DBで所有者とファイルメタ情報を管理し、認可付きダウンロードを作る
  5. S3互換ストレージへ移し、署名付きURLでアップロード/ダウンロードを分離
  6. 画像やPDFはバックグラウンド処理へ(検疫やサムネイル生成)
  7. 監査ログ、アラート、容量・失敗率のメトリクスも整える

参考リンク


まとめ

  • FastAPIのファイルアップロードは UploadFile を軸にすると、メモリ効率と拡張性が良くなります。
  • 実務の要点は、サイズ制限、実体検査、保存名の無害化、所有者ベースの認可です。
  • 本番では、ファイル本体をAPIサーバに持たせず、ストレージへ分離し、署名付きURLで受け渡しする設計が安定します。
  • 重い後処理(サムネイルやスキャン)はバックグラウンドへ渡し、APIは軽く保つと運用が楽になります。

次の記事の候補としては、この流れと相性が良い「マルチテナント設計(テナント境界と認可)」「監査ログ設計(誰が何をしたかを残す)」あたりが続編として書きやすいです。必要なら、そのテーマで続けますね。

投稿者 greeden

コメントを残す

メールアドレスが公開されることはありません。 が付いている欄は必須項目です

日本語が含まれない投稿は無視されますのでご注意ください。(スパム対策)