GuardDuty の finding を EventBridge で自動インシデント対応（SOAR）にする：本番設計の全体像を Terraform / Step Functions / Python で

GuardDuty finding
   │  (source=aws.guardduty / detail-type="GuardDuty Finding" / detail.severity=数値)
   ▼
EventBridge ルール（severity の数値マッチ + 型ルーティング + アーカイブ除外）
   ├─▶ [常に] SNS → Slack / メール       … 通知は広く（人間トリアージを速く）
   └─▶ Step Functions ステートマシン       … 対応は厳選（決める配管）
          1. ENRICH   finding を補強（タグ・所有者・関連リソース・既知IoC）
          2. DECIDE   重大度 × 型 × 許可リストで「通知のみ / 封じ込め / 要承認」を決定
          3. CONTAIN  リソース種別ごとのプレイブック Lambda（冪等・取り消し可能）
          4. APPROVE  破壊的操作だけ task token で人間承認を待つ（Slack/メール）
          5. NOTIFY+TICKET  対応結果を通知し、Jira/ServiceNow にチケット起票
          └─ 失敗時 → DLQ（SQS）＋ リトライ。取りこぼさない

{
  "version": "0",
  "id": "cd2d702e-ab31-411b-9344-793ce56b1bc7",
  "detail-type": "GuardDuty Finding",
  "source": "aws.guardduty",
  "account": "111122223333",
  "region": "ap-northeast-1",
  "time": "2026-06-27T02:00:00Z",
  "detail": {
    "id": "1ab23c...",
    "type": "UnauthorizedAccess:EC2/MaliciousIPCaller.Custom",
    "severity": 7.5,
    "accountId": "111122223333",
    "region": "ap-northeast-1",
    "updatedAt": "2026-06-27T02:00:00.000Z",
    "service": { "archived": false },
    "resource": { "resourceType": "Instance", "instanceDetails": { "instanceId": "i-0abc..." } }
  }
}

# ── High 以上(severity >= 7) かつ未アーカイブの GuardDuty finding を捕捉 ──
resource "aws_cloudwatch_event_rule" "gd_high" {
  name        = "guardduty-high-to-soar"
  description = "Route GuardDuty findings (severity >= 7, not archived) to the SOAR pipeline"
  event_pattern = jsonencode({
    source        = ["aws.guardduty"]
    "detail-type" = ["GuardDuty Finding"]
    detail = {
      severity = [{ numeric = [">=", 7] }] # 数値マッチ。High(7.0-8.9) と Critical(9.0-10.0)
      service  = { archived = [false] }    # 抑制で自動アーカイブされたものは元々来ないが、二重に弾く
    }
  })
}

# ① 人間向け：必ず SNS（→ Slack/メール）へ。通知は広く。
resource "aws_cloudwatch_event_target" "to_sns" {
  rule      = aws_cloudwatch_event_rule.gd_high.name
  target_id = "notify-sns"
  arn       = aws_sns_topic.security_alerts.arn
}

# ② 機械向け：オーケストレーションする Step Functions ステートマシンへ。
resource "aws_cloudwatch_event_target" "to_sfn" {
  rule      = aws_cloudwatch_event_rule.gd_high.name
  target_id = "soar-state-machine"
  arn       = aws_sfn_state_machine.responder.arn
  role_arn  = aws_iam_role.eventbridge_invoke_sfn.arn # EventBridge が SFN を起動するためのロール

  # 一過性の失敗に備えてリトライ＆DLQ（取りこぼさない）。
  retry_policy {
    maximum_event_age_in_seconds = 3600 # 最大1時間まで再試行（古すぎる対応は意味がない）
    maximum_retry_attempts       = 4
  }
  dead_letter_config { arn = aws_sqs_queue.soar_dlq.arn }
}

# 配送に失敗したイベントを溜める DLQ。ここが空でないことを必ずアラート対象にする。
resource "aws_sqs_queue" "soar_dlq" {
  name                      = "soar-eventbridge-dlq"
  message_retention_seconds = 1209600 # 14日。調査の猶予を最大に取る
}

# detector に対し、再発 finding の EventBridge 配信頻度を 15 分に。
# 注意: この頻度を変えられるのは（委任）管理者アカウントのみ。
#       メンバーアカウントは自分用に変更できず、管理者の設定が全メンバーに波及する。
resource "aws_guardduty_detector" "this" {
  enable                       = true
  finding_publishing_frequency = "FIFTEEN_MINUTES" # 自動対応の反応速度を上げる
}

{
  "Comment": "GuardDuty finding -> automated incident response (SOAR)",
  "StartAt": "Enrich",
  "States": {
    "Enrich": {
      "Comment": "finding を補強し、対応レベル(notify/contain/approve)を決定する",
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Parameters": {
        "FunctionName": "${EnrichRouterFunctionArn}",
        "Payload.$": "$"
      },
      "ResultSelector": { "decision.$": "$.Payload.decision", "finding.$": "$.Payload.finding" },
      "Retry": [{ "ErrorEquals": ["States.TaskFailed"], "IntervalSeconds": 2, "MaxAttempts": 3, "BackoffRate": 2.0 }],
      "Next": "RouteByDecision"
    },
    "RouteByDecision": {
      "Comment": "decision に応じて分岐。notify-only はそのまま通知へ",
      "Type": "Choice",
      "Choices": [
        { "Variable": "$.decision.mode", "StringEquals": "contain", "Next": "RouteByResourceType" },
        { "Variable": "$.decision.mode", "StringEquals": "approve", "Next": "WaitForHumanApproval" }
      ],
      "Default": "NotifyAndTicket"
    },
    "RouteByResourceType": {
      "Comment": "リソース種別ごとのプレイブックへ。封じ込めは冪等・取り消し可能なものだけ",
      "Type": "Choice",
      "Choices": [
        { "Variable": "$.decision.resourceType", "StringEquals": "Instance",   "Next": "ContainEC2" },
        { "Variable": "$.decision.resourceType", "StringEquals": "S3Bucket",   "Next": "ContainS3" },
        { "Variable": "$.decision.resourceType", "StringEquals": "Kubernetes", "Next": "ContainEKS" }
      ],
      "Default": "NotifyAndTicket"
    },
    "ContainEC2": { "Type": "Task", "Resource": "${ContainEc2FunctionArn}", "Next": "NotifyAndTicket",
      "Catch": [{ "ErrorEquals": ["States.ALL"], "Next": "NotifyAndTicket" }] },
    "ContainS3":  { "Type": "Task", "Resource": "${ContainS3FunctionArn}",  "Next": "NotifyAndTicket",
      "Catch": [{ "ErrorEquals": ["States.ALL"], "Next": "NotifyAndTicket" }] },
    "ContainEKS": { "Type": "Task", "Resource": "${ContainEksFunctionArn}", "Next": "NotifyAndTicket",
      "Catch": [{ "ErrorEquals": ["States.ALL"], "Next": "NotifyAndTicket" }] },

    "WaitForHumanApproval": {
      "Comment": "破壊的操作(鍵無効化・終了)は task token で人間の承認を待つ。承認まで実行しない",
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke.waitForTaskToken",
      "Parameters": {
        "FunctionName": "${RequestApprovalFunctionArn}",
        "Payload": { "finding.$": "$.finding", "taskToken.$": "$$.Task.Token" }
      },
      "TimeoutSeconds": 3600,
      "Next": "ExecuteDestructive",
      "Catch": [{ "ErrorEquals": ["States.Timeout"], "Next": "NotifyAndTicket" }]
    },
    "ExecuteDestructive": {
      "Comment": "承認済みのときだけ破壊的アクションを実行（鍵無効化等）",
      "Type": "Task",
      "Resource": "${ExecuteDestructiveFunctionArn}",
      "Next": "NotifyAndTicket"
    },

    "NotifyAndTicket": {
      "Comment": "結果を通知し、チケットを起票して終了（常にここを通る）",
      "Type": "Task",
      "Resource": "${NotifyTicketFunctionArn}",
      "End": true
    }
  }
}

"""GuardDuty finding を補強し、対応レベルを決定する enrich/route ディスパッチャ。

設計原則:
  - 副作用なし: 判断だけを返す純粋寄りの Lambda。封じ込めは別 Lambda に委譲。
  - 冪等: finding.id を冪等キーに、updatedAt で『古い再送』を弾く（DynamoDB 条件付き書き込み）。
  - スコープを絞る: 自動封じ込めは CONTAIN_ALLOWLIST に載った型 + High 以上のみ。
  - 破壊的は承認待ちへ: DESTRUCTIVE_ALLOWLIST は 'approve' を返し、自動実行しない。
  - 可観測: 構造化ログ。機密値(認証情報・PII)は出さない。
"""
from __future__ import annotations

import json
import logging
import os
import time
from typing import Any, Final, TypedDict

import boto3
from botocore.exceptions import ClientError

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

ddb = boto3.client("dynamodb")
IDEMPOTENCY_TABLE: Final[str] = os.environ["IDEMPOTENCY_TABLE"]
IDEMPOTENCY_TTL_SECONDS: Final[int] = 7 * 24 * 3600  # 7日で自動失効（TTL 属性で削除）

# 自動封じ込め（冪等・取り消し可能）を許す finding 型。
CONTAIN_ALLOWLIST: Final[frozenset[str]] = frozenset({
    "UnauthorizedAccess:EC2/MaliciousIPCaller.Custom",
    "CryptoCurrency:EC2/BitcoinTool.B!DNS",
    "Backdoor:EC2/C&CActivity.B!DNS",
    "Policy:S3/BucketPublicAccessGranted",
    "Policy:S3/BucketAnonymousAccessGranted",
})
# 破壊的操作（鍵無効化等）= 人間承認を挟む型。自動実行しない。
DESTRUCTIVE_ALLOWLIST: Final[frozenset[str]] = frozenset({
    "UnauthorizedAccess:IAMUser/InstanceCredentialExfiltration.OutsideAWS",
    "CredentialAccess:IAMUser/AnomalousBehavior",
})


class Decision(TypedDict):
    mode: str          # "notify" | "contain" | "approve"
    resourceType: str  # "Instance" | "S3Bucket" | "Kubernetes" | "AccessKey" | ...
    reason: str


def handler(event: dict[str, Any], _context: object) -> dict[str, Any]:
    detail = event["detail"]
    finding_id: str = detail["id"]
    updated_at: str = detail.get("updatedAt", detail.get("createdAt", ""))
    finding_type: str = detail["type"]
    severity: float = float(detail["severity"])
    resource_type: str = detail.get("resource", {}).get("resourceType", "Unknown")

    log = {"finding_id": finding_id, "type": finding_type,
           "severity": severity, "resourceType": resource_type}

    # ── 冪等ガード ──
    # 同じ finding_id を、同じ(以前の)updatedAt で2回処理しない。
    # at-least-once 配信や Step Functions の再試行で再入しても副作用は1回分。
    if not _claim_finding(finding_id, updated_at):
        logger.info(json.dumps({**log, "decision": "duplicate-skip"}))
        return {"finding": _safe_finding(detail), "decision": Decision(
            mode="notify", resourceType=resource_type, reason="duplicate (already processed)")}

    # ── 判断 ──
    if finding_type in DESTRUCTIVE_ALLOWLIST and severity >= 7.0:
        decision = Decision(mode="approve", resourceType=resource_type,
                            reason="destructive remediation requires human approval")
    elif finding_type in CONTAIN_ALLOWLIST and severity >= 7.0:
        decision = Decision(mode="contain", resourceType=resource_type,
                            reason="allowlisted reversible containment")
    else:
        # 許可リスト外・Medium 以下・型不明 → 通知＋チケットのみ。
        decision = Decision(mode="notify", resourceType=resource_type,
                            reason="notify-only (not in allowlist or below threshold)")

    logger.info(json.dumps({**log, "decision": decision["mode"], "reason": decision["reason"]}))
    return {"finding": _safe_finding(detail), "decision": decision}


def _claim_finding(finding_id: str, updated_at: str) -> bool:
    """冪等トークンを確保。新規 or より新しい updatedAt のときだけ True を返す。

    条件付き書き込みで『未処理 or 受信した updatedAt がより新しい』場合のみ上書き。
    既に同じ/より新しい版を処理済みなら False（＝重複）。
    """
    now = int(time.time())
    try:
        ddb.put_item(
            TableName=IDEMPOTENCY_TABLE,
            Item={
                "finding_id": {"S": finding_id},
                "updated_at": {"S": updated_at},
                "ttl": {"N": str(now + IDEMPOTENCY_TTL_SECONDS)},
            },
            # 未登録、または登録済みでも今回の updated_at の方が新しいときだけ書く。
            ConditionExpression="attribute_not_exists(finding_id) OR updated_at < :u",
            ExpressionAttributeValues={":u": {"S": updated_at}},
        )
        return True
    except ClientError as exc:
        if exc.response["Error"]["Code"] == "ConditionalCheckFailedException":
            return False  # 同じ/古い再送 → スキップ
        raise


def _safe_finding(detail: dict[str, Any]) -> dict[str, Any]:
    """下流に渡す finding から、機密になり得るフィールドを落とす（ログ・通知に乗せない）。"""
    keep = ("id", "type", "severity", "accountId", "region", "title", "updatedAt")
    slim = {k: detail.get(k) for k in keep}
    res = detail.get("resource", {})
    slim["resourceType"] = res.get("resourceType")
    slim["instanceId"] = res.get("instanceDetails", {}).get("instanceId")
    slim["bucketName"] = (res.get("s3BucketDetails") or [{}])[0].get("name")
    return slim

# EC2 封じ込め Lambda の実行ロール。EC2 の隔離に必要な action だけ。
data "aws_iam_policy_document" "contain_ec2" {
  statement {
    sid       = "DescribeForQuarantine"
    actions   = ["ec2:DescribeInstances"]
    resources = ["*"] # Describe は arn 単位で絞れないため condition で補う
  }
  statement {
    sid    = "QuarantineActions"
    actions = [
      "ec2:ModifyNetworkInterfaceAttribute", # ENI を隔離SGに付け替え
      "ec2:CreateTags",                       # 冪等ガード用の隔離タグ
    ]
    resources = ["*"]
    condition {
      test     = "StringEquals"
      variable = "aws:RequestedRegion"
      values   = [var.region] # 対象リージョンに限定
    }
  }
  # 重要: terminate / delete 系は与えない。破壊的操作は別ロール＋承認後に限る。
}

# 各 finding 型のサンプルを発火させ、SOAR パイプラインが想定通り動くか検証する。
aws guardduty create-sample-findings \
  --detector-id "$DETECTOR_ID" \
  --finding-types "UnauthorizedAccess:EC2/MaliciousIPCaller.Custom" \
                  "Policy:S3/BucketPublicAccessGranted" \
                  "UnauthorizedAccess:IAMUser/InstanceCredentialExfiltration.OutsideAWS"

{
  "version": "0",
  "detail-type": "Security Hub Findings - Custom Action",
  "source": "aws.securityhub",
  "account": "111122223333",
  "region": "ap-northeast-1",
  "resources": [
    "arn:aws:securityhub:ap-northeast-1:111122223333:action/custom/quarantine"
  ],
  "detail": {
    "actionName": "quarantine",
    "actionDescription": "Quarantine the affected resource",
    "findings": [ { "...": "ASFF 形式の finding 1件" } ]
  }
}

# Security Hub の手動カスタムアクション（"Quarantine"）を拾い、同じ SOAR 配管へ。
resource "aws_cloudwatch_event_rule" "sh_custom_action" {
  name = "securityhub-quarantine-custom-action"
  event_pattern = jsonencode({
    source        = ["aws.securityhub"]
    "detail-type" = ["Security Hub Findings - Custom Action"]
    resources     = ["arn:aws:securityhub:${var.region}:${var.account_id}:action/custom/quarantine"]
  })
}

resource "aws_cloudwatch_event_target" "sh_to_sfn" {
  rule     = aws_cloudwatch_event_rule.sh_custom_action.name
  arn      = aws_sfn_state_machine.responder.arn
  role_arn = aws_iam_role.eventbridge_invoke_sfn.arn
}