GuardDuty Malware Protection for S3 でアップロードされたファイルを自動スキャンする：単独運用・スキャン結果ゲーティング・S3 Protection との違いを実コードで

「ユーザーがアップロードしたファイル、マルウェアじゃないって誰が保証してます？」——アップロード機能を持つ SaaS のセキュリティを相談される場で、私が最初に投げる質問です。

たいてい返ってくるのは沈黙か、「拡張子で弾いてます」「フロントで MIME を見てます」です。けれどそれは入口の名札チェックであって、中身のスキャンではありません。攻撃者は拡張子も Content-Type も偽れます。アップロードされた PDF が実は実行ファイルで、それを別のユーザーがダウンロードして開いたら——あなたのアプリはマルウェアの配布経路になります。そして恐ろしいのは、それが正規のユーザー導線を通って起きることです。

この記事は、GuardDuty Malware Protection for S3 で「S3 にアップロードされたオブジェクトを自動でマルウェアスキャンし、清浄と確認できたものだけを下流に流す」仕組みを、本番品質で設計・実装するための実装ガイドです。題材として、私がマルチアカウント AWS 上のサーバーレス決済プラットフォームで IAM・可観測性・DR を横断実装し、実際の金銭を扱う基盤で「同じイベントを2回受けても副作用は1回」を冪等性で担保した経験——その発想は、at-least-once なスキャン結果イベントを安全に捌く設計とまったく同じ——も交えます。

この記事のルール：仕様・タグ値・上限・料金・EventBridge イベント構造は AWS 公式ドキュメント（2026年6月時点） に基づきます。上限・料金・対応リージョンは改定されるため、本番投入前に必ず公式の最新値（quotas／pricing ページ）を確認してください。そしてもう一つ——GuardDuty Malware Protection for S3 は「フルの AV／EDR」ではありません。既知・一部の未知マルウェアをスキャンエンジンで検出する一層であり、入力検証・最小権限の IAM・暗号化・WAF を代替しません。スキャン結果を受けた自動処理は 「冪等・スコープを絞る・取り消せる」 を満たすものから始めてください。

0. メンタルモデル：これは「アップロードの検疫所」であって「常時監視カメラ」ではない

設計を始める前に、混同されがちな2つの S3 セキュリティ機能を一行で切り分けます。これを最初に固定しないと、要件と機能がすれ違います。

S3 Protection ＝ S3 への「アクセス（API 操作）」を監視して不審な振る舞いを検知する（CloudTrail データイベント）。Malware Protection for S3 ＝ S3 にアップロードされた「ファイルの中身」をマルウェアスキャンする。前者は『誰が何をしたか』、後者は『何が入ってきたか』。

ここから3つの帰結が出ます。これが設計判断の土台です。

1. 見ているものが違う。 GuardDuty の脅威検知に含まれる S3 Protection は、CloudTrail の S3 データイベント（GetObject / PutObject / ListObjects / DeleteObject など）を解析し、「正規の認証情報を使った不審なアクセスやデータ持ち出し」を検知します。一方 Malware Protection for S3 は、新しくアップロードされたオブジェクトの中身をダウンロードしてマルウェアスキャンします。前者は振る舞いを、後者は中身を見る——別物です（1 章で表にして決着させます）。
2. Malware Protection for S3 は GuardDuty 本体なしで動く。 これが本記事の主役機能の最大の特徴です。GuardDuty サービスを有効化せず、この機能だけを単独（independent）で使えます。ただし単独モードではアカウントに detector ID が無いため、マルウェアを検知しても GuardDuty の finding は生成されません。結果は EventBridge のデフォルトバス・CloudWatch・（任意の）オブジェクトタグにだけ出ます（2 章）。
3. 検知だけでは安全にならない。「検疫 → 隔離 → 昇格」の配管が要る。 スキャンして「マルウェアです」と分かっても、その汚染オブジェクトがまだ下流から読める場所にあるなら意味がありません。本記事の山場は、汚染オブジェクトを下流から物理的に読めなくし、清浄なものだけを流す配管（5 章のセキュアアップロードパイプライン＋ TBAC）です。

この3点を押さえると、やるべきことは 「①2機能を取り違えない → ②保護プランを正しく有効化する（単独 or 本体込み）→ ③スキャン結果を冪等な配管で『隔離／昇格』に変える」 の3つだと分かります。順に作ります。

1. 取り違えないための決着表：S3 Protection vs Malware Protection for S3

名前が似ているせいで、現場で最も多い事故が**「やりたいことと有効化した機能のミスマッチ」**です。まず正面から決着をつけます。

公式の表現を借りると、S3 Protection は "helps you detect potential security risks for data, such as data exfiltration and destruction"——つまりデータへの脅威（持ち出し・破壊）の検知です。Malware Protection for S3 は "helps you detect potential presence of malware by scanning newly uploaded objects"——新規アップロードのマルウェアスキャンです。

設計の指針：この2つは競合せず、補完します。「アップロードを受け付ける」かつ「機密データを置く」バケットなら、両方有効化するのが理想です。Malware Protection for S3 が「入ってくる悪いもの」を、S3 Protection が「正規のアクセスに紛れた持ち出し」を見ます。なお S3 Protection は GuardDuty 本体の保護プランなので、有効化方法はピラー記事の S3_DATA_EVENTS feature を参照してください。本記事はここから先、Malware Protection for S3 に絞ります。

さらに紛らわしい第3の存在に注意：「Malware Protection for EC2」という別機能もあります。これは EC2/コンテナにアタッチされた EBS ボリュームをエージェントレスでスキャンするもので、本記事の for S3 とは対象がまったく違います。「Malware Protection」だけで会話すると三者が混線するので、必ず for S3 / for EC2 を付けて話すのが事故防止です。

2. Malware Protection for S3 の仕組み：単独運用と「本体込み」の決定的な違い

2.1 スキャンオンアップロードのモデル

仕組みはシンプルです。保護を有効にしたバケット（公式では "protected bucket"）に、オブジェクトが新規アップロードされる（または既存オブジェクトの新バージョンが上がる）と、GuardDuty が自動でマルウェアスキャンを開始します。

スキャンを起動するのは、S3 の Object Created 系イベント——PutObject / POST Object / CopyObject / CompleteMultipartUpload——です。GuardDuty は対象オブジェクトを AWS PrivateLink 経由でダウンロードし、同一リージョンの隔離環境（インターネット非接続の VPC）で復号・読み取り・スキャンします。スキャン中の一時コピーは KMS で暗号化され、スキャン完了後にダウンロードしたコピーは削除されます。つまりあなたのデータがスキャンのために外に出ることはなく、結果のメタデータだけが残ります。

2.2 2つの有効化アプローチ——ここが本記事の核心

Malware Protection for S3 には、有効化のしかたが2通りあります。この違いが運用設計を分けます。

公式が明確に書いている単独モードの決定的な性質を、正確に押さえてください：

"When you enable Malware Protection for S3 independently in an account, that account will not have an associated detector ID. ... when an S3 malware scan detects the presence of malware, no GuardDuty finding will get generated in your AWS account because all GuardDuty findings are associated with a detector ID."

つまり——単独モードでは「マルウェアを見つけても GuardDuty のダッシュボードには何も出ない」。これは欠陥ではなく設計です。GuardDuty の finding は detector ID に紐づくため、detector が無ければ finding も無い。代わりに結果は EventBridge のデフォルトイベントバスと CloudWatch のメトリクス、そして有効化していればオブジェクトタグに出ます。

だから単独モードの設計では、EventBridge とタグが「検知の唯一の出口」になります。「finding が来るだろう」という前提でアラートを組むと、単独モードでは永遠にアラートが鳴りません。後述のパイプラインは、この EventBridge イベントを軸に組みます。

2.3 トレードオフ：どちらを選ぶか

• 既に GuardDuty を組織で運用しているなら → (a) 本体込み。 finding として既存のインシデント対応配管（EventBridge → 自動対応）に自然に乗ります。マルウェア検知が他の脅威シグナルと同じ土俵に並ぶ価値は大きい。
• 「このアップロードバケットだけスキャンしたい」「GuardDuty 全体は今は要らない」なら → (b) 単独。 最小コスト・最小権限で、ピンポイントに機能を導入できます。後から GuardDuty 本体を有効化すれば finding も出るようになります。

私の推奨：まずは単独モードで小さく始め、スキャン結果を EventBridge で受けて配管を作る。GuardDuty 本体の全社展開はそれ自体が大きな意思決定なので、「アップロードの検疫」という1つの要件をそれに人質に取らせない。後から (a) に昇格しても、EventBridge を軸に組んだ配管はそのまま再利用できます（finding 経路が増えるだけ）。

3. 有効化の構成要素：Malware Protection plan・IAM ロール・プレフィックス・上限

3.1 何ができて、何ができないか（制約を先に固定）

• 自アカウントのバケットのみ。 委任 GuardDuty 管理者でも、メンバーアカウントのバケットには有効化できません（同一アカウント内に閉じる）。
• 同一リージョンのみ。 クロスリージョンのバケットは対象外。
• バケット単位の "Malware Protection plan" リソースが作られ、固有の plan ID が振られます。GuardDuty は DO-NOT-DELETE-AmazonGuardDutyMalwareProtectionS3* という名前の EventBridge マネージドルールを自動作成・管理します（手で消さないこと）。
• プレフィックスでスコープを絞れる。 バケット全体ではなく、特定の**オブジェクトプレフィックス（最大5つ）**だけをスキャン対象にできます。「アップロード受け口は uploads/ 配下だけ」という設計に有効。
• KMS 暗号化バケットに対応（スキャン環境内で復号）。ただしSSE-C（顧客提供キー）のオブジェクトはスキャンできません（後述の ACCESS_DENIED 理由 SSE_C_ENCRYPTED_OBJECT）。

3.2 IAM ロール：GuardDuty に「あなたの代わり」をさせる最小権限

Malware Protection for S3 は、GuardDuty があなたのアカウントでスキャンを実行するための IAM ロールを要求します。このロールに必要な権限は、おおむね次の3カテゴリです：

1. 新規アップロードの通知を受ける（EventBridge マネージドルール経由）
2. 対象オブジェクトを読む・復号する（s3:GetObject ＋ 必要なら kms:Decrypt）
3. （任意）スキャン後にタグを付ける（s3:PutObjectTagging）

ロールの信頼ポリシーは、GuardDuty の Malware Protection サービスプリンシパルが sts:AssumeRole できるようにします。公式の IAM ポリシーテンプレートに従うのが安全で、バケットを増やすときは同じロールに対象バケット名を足していく運用が推奨です。最小権限の原則そのままに、「このバケットの、このプレフィックスだけ」に Resource を絞ります（データ層の最小権限の考え方と同型）。

3.3 上限（公式の正確な値・要・最新確認）

これらを超えると、スキャンはスキップされ、結果は UNSUPPORTED（理由例：OBJECT_SIZE_LIMIT_EXCEEDED / EXTRACTED_FILE_LIMIT_EXCEEDED / EXTRACTED_LEVEL_LIMIT_EXCEEDED / EXTRACTION_RATIO_LIMIT_EXCEEDED）になります。**「スキャンできなかった ≠ 安全」**である点を、後段の配管で必ず扱います（5 章）。

3.4 オンデマンドスキャン：既存オブジェクト・再スキャン用

自動スキャンは新規アップロードに対して走りますが、保護を有効化する前から存在したオブジェクトや、一度スキャンしたものを再スキャンしたい場合は、オンデマンドスキャンを使います。

# 既存オブジェクト（最新バージョン）をオンデマンドでスキャン。
# 事前条件: 対象バケットで Malware Protection for S3 が有効 + 呼び出し元に
#           AWS マネージドポリシー AmazonGuardDutyFullAccess_v2 が付与されていること。
aws guardduty send-object-malware-scan \
  --s3-object '{"Bucket": "my-upload-landing", "Key": "uploads/legacy-file.pdf"}'

# 特定バージョンを指定してスキャンする場合は VersionId を渡す。
aws guardduty send-object-malware-scan \
  --s3-object '{"Bucket": "my-upload-landing", "Key": "uploads/legacy-file.pdf", "VersionId": "d41d8cd9...EXAMPLE"}'

注意点：オンデマンドスキャンはplan のプレフィックス設定を上書きし（プレフィックス外でも対象にできる）、上限と料金は自動スキャンと同じく適用されます。そして重要——オンデマンドは無料枠の対象外です。「成功応答＝スキャン完了」ではなく受理されただけなので、結果は必ず EventBridge / タグ / CloudWatch で確認します。

4. スキャン結果を読む：タグ・状態値・EventBridge イベント

自動処理を組むには、結果の構造を正確に読める必要があります。出口は3つ——オブジェクトタグ・EventBridge イベント・CloudWatch メトリクスです。

4.1 スキャン結果タグ（任意・有効化は「アップロード前」必須）

タグ付けを有効にしておくと、スキャン後に GuardDuty がオブジェクトへ定義済みタグを付けます。キーと値は公式に固定です：

キー:  GuardDutyMalwareScanStatus
値:    NO_THREATS_FOUND | THREATS_FOUND | UNSUPPORTED | ACCESS_DENIED | FAILED

致命的な落とし穴：タグ付けはオブジェクトがアップロードされる「前」に有効化しておかないと、後から有効化してもそのオブジェクトにはタグが付きません。だから「バケットを作る → 保護＋タグ付けを有効化 → それからアップロードを受け付ける」の順序が鉄則です。また、オブジェクトに付けられるタグは最大10個で、枠が埋まっていると GuardDuty はタグを付けられず、代わりに「post-scan tag failure」イベントが EventBridge に出ます。

4.2 EventBridge のスキャン結果イベント（自動処理の主役）

GuardDuty はスキャン結果を必ずデフォルトの EventBridge イベントバスに publishします（単独モードでも、本体込みでも）。これが自動処理の入口です。detail-type は GuardDuty Malware Protection Object Scan Result、source は aws.guardduty。

NO_THREATS_FOUND のイベント（公式スキーマ、抜粋）：

{
  "detail-type": "GuardDuty Malware Protection Object Scan Result",
  "source": "aws.guardduty",
  "account": "111122223333",
  "region": "us-east-1",
  "resources": ["arn:aws:guardduty:us-east-1:111122223333:malware-protection-plan/b4c7f464ab3a4EXAMPLE"],
  "detail": {
    "schemaVersion": "1.0",
    "scanStatus": "COMPLETED",
    "resourceType": "S3_OBJECT",
    "s3ObjectDetails": {
      "bucketName": "amzn-s3-demo-bucket",
      "objectKey": "uploads/report.pdf",
      "eTag": "ASIAI44QH8DHBEXAMPLE",
      "versionId": "d41d8cd98f00b204e9800998eEXAMPLE",
      "s3Throttled": false
    },
    "scanResultDetails": {
      "scanResultStatus": "NO_THREATS_FOUND",
      "threats": null,
      "statusReasons": null
    }
  }
}

THREATS_FOUND のときは scanResultDetails.threats に検出名が入ります（既定では最初に検出した1件を報告し、scanStatus は COMPLETED）：

{
  "detail": {
    "scanStatus": "COMPLETED",
    "s3ObjectDetails": { "bucketName": "amzn-s3-demo-bucket", "objectKey": "uploads/evil.bin", "versionId": "..." },
    "scanResultDetails": {
      "scanResultStatus": "THREATS_FOUND",
      "threats": [ { "name": "EICAR-Test-File (not a virus)" } ],
      "statusReasons": null
    }
  }
}

スキャンがスキップされたときは scanStatus: "SKIPPED" で、scanResultStatus が UNSUPPORTED か ACCESS_DENIED、さらに statusReasons に具体理由（PASSWORD_PROTECTED・SSE_C_ENCRYPTED_OBJECT・OBJECT_SIZE_LIMIT_EXCEEDED など）が入ります。

at-least-once（必読）：公式が明言しています——"GuardDuty uses at-least-once delivery, which means you might receive multiple scan results for the same object. We recommend designing your applications to handle duplicate results." つまり同じオブジェクトのスキャン結果が複数回届き得ます。決済基盤で二重課金を防ぐのと同じ発想で、結果ハンドラは冪等でなければなりません（5 章の Lambda はそこを作り込みます）。なお課金は重複してもオブジェクトあたり1回です。

4.3 状態モデルの注意：状態（scanStatus）と結果（scanResultStatus）は別物

混同しやすいので明示します。scanStatus は「スキャンの状態」（COMPLETED / SKIPPED / FAILED）、scanResultStatus は「結果」（上の5値）。COMPLETED でも結果は THREATS_FOUND か NO_THREATS_FOUND のどちらか。「SKIPPED／FAILED／UNSUPPORTED／ACCESS_DENIED は『安全』を意味しない」——スキャンできなかっただけです。設計の鉄則は 「NO_THREATS_FOUND だけを明示的に許可（allowlist）し、それ以外はすべて隔離側に倒す」。これが次章の配管の核になります。

5. 高付加価値な応用：セキュアアップロードパイプライン（landing → 清浄 / 隔離）

ここが本記事の山場です。スキャンするだけでは安全になりません。「汚染されているかもしれないオブジェクトを、下流が絶対に読めない状態に保ち、清浄と確認できたものだけを流す」——その配管を作ります。

5.1 全体像：3バケット＋イベント駆動の昇格

ユーザー
  │  アップロード（署名付き URL など）
  ▼
landing バケット（Malware Protection for S3 有効・タグ付け ON）
  │  ・下流は読めない（TBAC: 清浄タグが無いオブジェクトの GetObject を DENY）
  │  ・GuardDuty が自動スキャン
  ▼
EventBridge（detail-type = "GuardDuty Malware Protection Object Scan Result"）
  ▼
スキャン結果 Lambda（冪等）
  ├─ NO_THREATS_FOUND → clean バケットへ「昇格」（コピー）。下流はここだけ読む
  ├─ THREATS_FOUND    → quarantine バケットへ隔離 + セキュリティ通知（人間へ）
  └─ それ以外(UNSUPPORTED/ACCESS_DENIED/FAILED/SKIPPED) → quarantine + 要調査通知
                                                          （「スキャン不可 ≠ 安全」）
下流の消費者
  └─ clean バケットだけを読む（landing は TBAC で物理的に読めない）

設計の急所は 2段の防御です：

1. EventBridge 駆動の昇格——清浄なものだけ clean に移す（下流は clean しか見ない）。
2. TBAC（タグベースアクセス制御）——万一 landing を誰かが読もうとしても、NO_THREATS_FOUND タグの無いオブジェクトの GetObject を S3 バケットポリシーで DENY。アプリのロジックではなくストレージ層の境界で封じるので、コードのバグでは破れません（least privilege をデータ層で効かせる発想と同じ）。

5.2 TBAC バケットポリシー：清浄タグが無ければ読ませない

公式のテンプレートに沿った、landing バケットの**「NO_THREATS_FOUND でなければ読めない」**ポリシーです。{{...}} を自分の値に置き換えます。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "NoReadUnlessClean",
      "Effect": "Deny",
      "NotPrincipal": {
        "AWS": [
          "arn:aws:sts::555555555555:assumed-role/IAM-role-name/GuardDutyMalwareProtection",
          "arn:aws:iam::555555555555:role/IAM-role-name"
        ]
      },
      "Action": ["s3:GetObject", "s3:GetObjectVersion"],
      "Resource": "arn:aws:s3:::amzn-s3-demo-bucket/*",
      "Condition": {
        "StringNotEquals": {
          "s3:ExistingObjectTag/GuardDutyMalwareScanStatus": "NO_THREATS_FOUND"
        }
      }
    },
    {
      "Sid": "OnlyGuardDutyCanTagScanStatus",
      "Effect": "Deny",
      "NotPrincipal": {
        "AWS": [
          "arn:aws:sts::555555555555:assumed-role/IAM-role-name/GuardDutyMalwareProtection",
          "arn:aws:iam::555555555555:role/IAM-role-name"
        ]
      },
      "Action": "s3:PutObjectTagging",
      "Resource": "arn:aws:s3:::amzn-s3-demo-bucket/*",
      "Condition": {
        "ForAnyValue:StringEquals": {
          "s3:RequestObjectTagKeys": "GuardDutyMalwareScanStatus"
        }
      }
    }
  ]
}

このポリシーの読み解き：

• NoReadUnlessClean：s3:ExistingObjectTag/GuardDutyMalwareScanStatus が NO_THREATS_FOUND でないオブジェクトの読み取りを DENY。タグがまだ付いていない（＝スキャン未完了）オブジェクトも、当然 NO_THREATS_FOUND ではないので読めません。「スキャンが終わって清浄と確定するまで誰も読めない」をストレージ層で保証します。
• NotPrincipal で GuardDuty ロールだけ除外：スキャン実行ロール（と GuardDuty が assume する .../GuardDutyMalwareProtection）は、読み取り・タグ付けのために除外します。
• OnlyGuardDutyCanTagScanStatus：GuardDutyMalwareScanStatus タグを付けられるのは GuardDuty だけにする DENY。これが無いと、誰かが汚染オブジェクトに手で NO_THREATS_FOUND を付けてゲートをすり抜けられます。

組織での追加防御：AWS Organizations を使っているなら、**SCP で「GuardDutyMalwareScanStatus タグを改ざんできない」**を全社強制してください（公式が EC2 の例を s3 に置き換えて使うよう案内）。タグが信頼の根拠になるなら、タグの改ざん防止が前提条件です。ここを飛ばすと TBAC は「鍵のかかっていない金庫」になります。

5.3 スキャン結果 Lambda（Python・冪等）

EventBridge のスキャン結果を受け、NO_THREATS_FOUND だけを clean へ昇格、それ以外は quarantine へ隔離する Lambda です。at-least-once 配信を前提に冪等化します。

"""GuardDuty Malware Protection for S3 のスキャン結果に応答する Lambda。

設計原則:
  - 冪等: EventBridge は at-least-once。同じオブジェクトの結果を2回受けても副作用は1回分。
  - allowlist で安全側に倒す: 'NO_THREATS_FOUND' のときだけ昇格。それ以外は全て隔離。
    （UNSUPPORTED/ACCESS_DENIED/FAILED/SKIPPED は『スキャン不可』であって『安全』ではない）
  - 取り消し可能: landing からは消さずコピーで昇格／隔離。誤判定でも原本が残る。
  - 可観測: 構造化ログ。脅威名は通知に載せるが、オブジェクトの中身は読まない・出さない。
"""
from __future__ import annotations

import json
import logging
import os
from typing import Any, Final
from urllib.parse import unquote_plus

import boto3
from botocore.exceptions import ClientError

logger = logging.getLogger()
logger.setLevel(logging.INFO)

s3 = boto3.client("s3")
sns = boto3.client("sns")

CLEAN_BUCKET: Final[str] = os.environ["CLEAN_BUCKET"]
QUARANTINE_BUCKET: Final[str] = os.environ["QUARANTINE_BUCKET"]
ALERT_TOPIC_ARN: Final[str] = os.environ["ALERT_TOPIC_ARN"]

# 昇格を許す唯一の結果値。これ以外は全部隔離側へ倒す（fail-closed）。
CLEAN_STATUS: Final[str] = "NO_THREATS_FOUND"


def handler(event: dict[str, Any], _context: object) -> dict[str, str]:
    detail = event["detail"]
    obj = detail["s3ObjectDetails"]
    src_bucket: str = obj["bucketName"]
    # S3 のイベントキーは URL エンコードされ得るのでデコードする。
    key: str = unquote_plus(obj["objectKey"])
    version_id: str | None = obj.get("versionId")
    result: str = detail.get("scanResultDetails", {}).get("scanResultStatus", "FAILED")
    threats = detail.get("scanResultDetails", {}).get("threats")

    log = {"bucket": src_bucket, "key": key, "version": version_id, "result": result}

    if result == CLEAN_STATUS:
        dest = CLEAN_BUCKET
        disposition = "promoted"
    else:
        dest = QUARANTINE_BUCKET
        disposition = "quarantined"

    # ── 昇格／隔離（冪等）──
    # 同一バージョンを宛先キーに含めることで、再配信されても同じ宛先に上書きコピー＝
    # 何度実行しても結果は同じ（at-least-once に対する冪等性）。
    dest_key = f"{key}" if version_id is None else f"{key}"
    moved = _idempotent_copy(src_bucket, key, version_id, dest, dest_key)

    # 脅威・スキャン不可は人間に通知（fail-closed の確認とトリアージ）。
    if result != CLEAN_STATUS:
        _alert(src_bucket, key, result, threats, disposition)

    logger.info(json.dumps({**log, "disposition": disposition, "copied": moved}))
    return {"disposition": disposition, "result": result}


def _idempotent_copy(
    src_bucket: str, key: str, version_id: str | None, dest_bucket: str, dest_key: str
) -> bool:
    """landing から dest へコピー（冪等）。既に同じ版がコピー済みなら no-op。

    冪等キー: 宛先に '元の versionId' をメタデータとして書き、再実行時に一致したらスキップ。
    landing の原本は消さない（誤判定からの復旧余地を残す＝取り消し可能）。
    """
    # 既にコピー済みかを確認（同じ source version なら 2 回目はスキップ）。
    try:
        head = s3.head_object(Bucket=dest_bucket, Key=dest_key)
        if head.get("Metadata", {}).get("source-version-id") == (version_id or ""):
            return False  # 既に同じ版を処理済み → no-op
    except ClientError as exc:
        if exc.response["Error"]["Code"] not in ("404", "NoSuchKey"):
            raise  # 想定外のエラーは握りつぶさない

    copy_source: dict[str, str] = {"Bucket": src_bucket, "Key": key}
    if version_id:
        copy_source["VersionId"] = version_id

    s3.copy_object(
        Bucket=dest_bucket,
        Key=dest_key,
        CopySource=copy_source,
        # 元バージョンを冪等キーとして残す。MetadataDirective=REPLACE で確実に書く。
        Metadata={"source-version-id": version_id or ""},
        MetadataDirective="REPLACE",
    )
    return True


def _alert(
    bucket: str, key: str, result: str, threats: list[dict[str, str]] | None, disposition: str
) -> None:
    """脅威・スキャン不可をセキュリティ担当へ通知。中身は読まない・載せない。"""
    threat_names = ", ".join(t.get("name", "?") for t in (threats or [])) or "n/a"
    sns.publish(
        TopicArn=ALERT_TOPIC_ARN,
        Subject=f"[S3 Malware][{result}] {bucket}/{key}",
        Message="\n".join(
            [
                f"bucket: {bucket}",
                f"key: {key}",
                f"scanResultStatus: {result}",
                f"threats: {threat_names}",
                f"disposition: {disposition}",
                "note: 'NO_THREATS_FOUND' 以外は安全とみなさず隔離済み。要トリアージ。",
            ]
        ),
    )

このコードの設計判断を明示します。

• fail-closed（安全側に倒す）：昇格するのは NO_THREATS_FOUND のただ一つだけ。UNSUPPORTED・ACCESS_DENIED・FAILED、そして万一未知の値が来ても、すべて隔離に倒れます。「判定できなかったものは危険とみなす」——セキュリティの既定値はこれです。
• 冪等：宛先に元の versionId をメタデータで残し、2回目以降は head_object で検出して no-op。at-least-once で同じ結果が複数回来ても、コピーは1回分。これは決済基盤で同じイベントを2回受けても課金は1回にする発想と同型です。
• 取り消し可能：landing の原本は消さずにコピーで昇格／隔離します。誤判定（false positive）が後で分かっても、原本が残っているので戻せます。
• 最小権限：この Lambda の実行ロールは、landing の s3:GetObject*、clean/quarantine の s3:PutObject* ＋ head_object、sns:Publish に限定し、Resource を各バケット ARN に絞ります。バケット間を跨ぐので、各バケットのバケットポリシー側でもこのロールを明示的に許可します。
• 中身を扱わない：Lambda はオブジェクトのバイト列を読みません（コピーは S3 サーバーサイド copy_object）。脅威名は通知に載せますが、ファイル内容はログにも通知にも出しません（可観測性とセキュリティの両立）。

EventBridge ルールと Lambda の配線：detail-type = "GuardDuty Malware Protection Object Scan Result" を拾うルールを作り、ターゲットに Lambda を置き、リトライ＋ DLQを付けます（自動対応記事の retry_policy ＋ dead_letter_config と同じパターン）。DLQ が空でないこと＝結果を処理できなかったオブジェクトがある＝危険な沈黙なので、必ずアラート対象にします。

5.4 Terraform：Malware Protection plan を landing バケットに付ける

aws_guardduty_malware_protection_plan リソースで、landing バケットの uploads/ プレフィックスだけを保護し、結果タグ付けを有効化します。

# landing バケットに Malware Protection for S3 を有効化する。
# role = GuardDuty が assume してスキャン・タグ付けするための IAM ロール ARN。
resource "aws_guardduty_malware_protection_plan" "landing" {
  role = aws_iam_role.gd_malware_s3.arn

  protected_resource {
    s3_bucket {
      bucket_name = aws_s3_bucket.landing.id
      # スキャン対象をアップロード受け口に限定（最大5プレフィックス）。
      # 受け口を絞ることで、無関係なオブジェクトのスキャン課金を避ける。
      object_prefixes = ["uploads/"]
    }
  }

  # スキャン結果をオブジェクトタグ(GuardDutyMalwareScanStatus)に書く。
  # 5.2 の TBAC ポリシーはこのタグに依存するので ENABLED 必須。
  actions {
    tagging {
      status = "ENABLED"
    }
  }

  tags = { ManagedBy = "terraform", Purpose = "upload-malware-scan" }
}

単独運用の補足：この aws_guardduty_malware_protection_plan は、aws_guardduty_detector が無くても作れます——これが「単独モード」の Terraform 表現です。GuardDuty 本体込みにしたいなら、別途 aws_guardduty_detector（とお好みで MALWARE_PROTECTION 系 feature）を有効化すれば、検知が finding としても出るようになります。role に渡す IAM ロールの信頼ポリシーと権限（s3:GetObject / kms:Decrypt / s3:PutObjectTagging ＋ EventBridge マネージドルール用）は、3.2 のとおり公式テンプレートに沿って最小化してください。

6. コスト：従量課金・無料枠・単独 vs 本体込み

6.1 課金モデル（要・公式最新確認）

Malware Protection for S3 の料金は、他の保護プランと異なる従量課金です。概念を押さえると予算が読めます。

公式の無料枠は 「各アカウント・各リージョンで、月 1,000 リクエスト ＋ 1 GB データスキャンまで無料」。これを超えた分から従量課金が始まります。オンデマンドスキャンとタグ付けは無料枠の対象外である点に注意してください。

金額の数値は必ず公式の最新値を確認：本記事では具体的な単価（USD/GB、USD/1,000 オブジェクト等）をあえて断定しません。料金はリージョンで異なり改定されるため、GuardDuty pricing ページで自分のリージョンの最新値を見積もるのが正解です。US East（バージニア北部）の数値を見るときも、それは**参考（要確認）**として扱ってください。コスト最適化の話はGuardDuty のコスト最適化記事で別途深掘りしています。

6.2 単独 vs 本体込みのコスト含意

• 単独モード：払うのは Malware Protection for S3 の従量課金だけ。GuardDuty 本体（基盤検知・他の保護プラン）の費用は発生しません。「アップロード検疫だけ」という要件に最小コストで応えられます。
• 本体込み：Malware Protection for S3 の従量課金に加え、GuardDuty 本体と有効化した他プランの費用が乗ります。ただしマルウェア検知が finding として既存のインシデント対応・相関に乗る価値が付きます。

コスト設計の勘所：プレフィックスでスコープを絞る（uploads/ だけ）こと、そしてスキャン対象を「本当にユーザーが上げる受け口」に限定することが、そのまま課金の最小化になります。バケット全体を無造作に保護すると、内部処理で生成される一時オブジェクトまでスキャンしてオブジェクト評価数が膨らむことがあります。「資産に比例した課金」を作るのは、ここでも設計の仕事です。

7. まとめ：Malware Protection for S3 本番チートシート

迷ったときの早見表です。

• 2機能を取り違えない：S3 Protection ＝ CloudTrail の S3 データイベントで不審アクセス／持ち出しを検知（GuardDuty 必須・feature S3_DATA_EVENTS）。Malware Protection for S3 ＝ 新規アップロードの中身をマルウェアスキャン。「誰が何をしたか」vs「何が入ってきたか」。さらに for EC2（EBS スキャン）とも別物。
• 単独運用できる：GuardDuty 本体なしで Malware Protection for S3 だけ有効化可。ただしdetector ID が無い＝マルウェア検知時に GuardDuty finding は生成されない。結果は EventBridge デフォルトバス＋CloudWatch＋（任意の）タグだけ。「finding 前提」のアラートは単独モードでは鳴らない。
• 仕組み：バケット単位の Malware Protection plan。アップロード（PutObject 等）で自動スキャン。自アカウント・同一リージョンのみ、委任管理者でもメンバーのバケットは不可。プレフィックスで最大5つにスコープ、KMS 対応（SSE-C は不可）。既存/再スキャンは SendObjectMalwareScan（オンデマンド・無料枠対象外）。
• 上限（要・最新確認）：最大オブジェクト 100 GB・抽出ファイル 100,000・ネスト 100・保護バケット 25/アカウント/リージョン。超過は UNSUPPORTED。
• 結果：タグ GuardDutyMalwareScanStatus ＝ NO_THREATS_FOUND / THREATS_FOUND / UNSUPPORTED / ACCESS_DENIED / FAILED。タグ付けはアップロード前に有効化必須。EventBridge は detail-type="GuardDuty Malware Protection Object Scan Result"・at-least-once → 冪等必須。scanStatus（状態）と scanResultStatus（結果）は別物。
• 安全なパイプライン：landing（保護＋タグ）→ EventBridge → 冪等な Lambda が NO_THREATS_FOUND だけ clean へ昇格・それ以外は quarantine（fail-closed）。下流は clean だけ読む。**TBAC バケットポリシーで「清浄タグが無いオブジェクトの GetObject を DENY」**し、タグ改ざんは GuardDuty 以外に禁止（＋組織なら SCP）。
• コスト（要・最新確認）：スキャン GB ＋ オブジェクト評価数の従量。月 1,000 リクエスト＋1 GB 無料（オンデマンド・タグ付けは対象外）。プレフィックスでスコープを絞るのがそのまま節約。単独モードは本体費用ゼロ。

Malware Protection for S3 は「箱に入れたら勝手に安全」ではなく、**「スキャン結果（特に NO_THREATS_FOUND 以外）を、冪等で・取り消せて・fail-closed な配管に変えられるか」**で価値が決まります。最大のレバレッジは検知そのものより、汚染を下流から物理的に遮断し、清浄だけを流す境界（EventBridge ＋ TBAC）の設計にあります。

私はマルチアカウントのサーバーレス決済プラットフォームで、実際の金銭・カーボンクレジット・地域通貨を扱う基盤の IAM・可観測性・DR を横断実装し、「正しさ」を運用の注意深さではなくコードの構造と冪等性で担保してきました——at-least-once な世界で同じイベントを2回受けても副作用は1回にする、その発想はそのままスキャン結果の処理に転用できます。特定のクライアント案件で Malware Protection for S3 を運用した、と主張するつもりはありません。けれど、この「セキュアアップロードパイプライン（単独運用・TBAC ゲーティング・冪等な昇格／隔離）」は、上記の実経験に基づいて設計・実装してお渡しできます。

「自社のアップロード機能に、マルウェアスキャンの検疫所をどう組み込むか——単独で始めるか GuardDuty 本体に乗せるか、TBAC でどう下流を守るか、コストをどう最小化するか」。要件の整理段階から、Terraform / Lambda / バケットポリシーの実装まで、一人 ×生成AI（Claude Code）で速く・安全に伴走できます。 お気軽にご相談ください。

参考（公式ドキュメント）

• GuardDuty Malware Protection for S3 — 機能概要・2つの有効化アプローチ（本体込み／単独）・単独モードで finding が生成されない理由・自アカウント＆同一リージョン制約
• How does Malware Protection for S3 work? — Malware Protection plan・IAM ロール・プレフィックス・KMS 復号・タグキー GuardDutyMalwareScanStatus・at-least-once 配信
• Monitoring S3 object scans in Malware Protection for S3 — scanStatus と scanResultStatus の正確な値・statusReasons の一覧
• Monitoring S3 object scans with Amazon EventBridge — detail-type="GuardDuty Malware Protection Object Scan Result" の完全な JSON スキーマ
• Using tag-based access control (TBAC) with Malware Protection for S3 — NO_THREATS_FOUND でなければ DENY する S3 バケットポリシーの正式テンプレート・タグ改ざん防止
• On-demand S3 malware scan in GuardDuty — SendObjectMalwareScan API・既存/再スキャン・無料枠対象外
• Quotas in Malware Protection for S3 — 最大オブジェクト 100 GB・抽出ファイル 100,000・ネスト 100・保護バケット 25
• Pricing and usage cost for Malware Protection for S3 — 無料枠（1,000 リクエスト＋1 GB/月）・タグ付け／オンデマンドは対象外
• GuardDuty S3 Protection — CloudTrail S3 データイベント監視（Malware Protection for S3 とは別物）
• Terraform: aws_guardduty_malware_protection_plan — role / protected_resource { s3_bucket { bucket_name, object_prefixes } } / actions { tagging { status } }