ASR 自動修復結果を Slack に通知するパイプラインを構築してみた

AWSTemplateFormatVersion: '2010-09-09'
Description: ASR Slack Notification Pipeline - Formatter Lambda + SNS + Amazon Q Developer

Parameters:
  SlackWorkspaceId:
    Type: String
    Description: >-
      Slack ワークスペースID。Amazon Q Developer コンソールで確認可能
  SlackChannelId:
    Type: String
    Description: >-
      通知先の Slack チャンネルID
  AsrTopicArn:
    Type: String
    Description: >-
      ASR Topic ARN。SSM Parameter /Solutions/SO0111/SNS_Topic_ARN から取得
  AsrAlarmTopicArn:
    Type: String
    Description: >-
      ASR Alarm Topic ARN。ASR Topic ARN の Topic名を置換して導出

Resources:
  # Amazon Q Developer用の通知中継SNS Topic
  NotificationTopic:
    Type: AWS::SNS::Topic
    Properties:
      TopicName: ASR-SlackNotifications
      DisplayName: ASR Remediation Notifications

  # Amazon Q Developer の IAM ロール（リソースタイプは旧名のまま）
  ChatbotRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: chatbot.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/AWSResourceExplorerReadOnlyAccess

  # Amazon Q Developer Slack チャンネル設定（リソースタイプは旧名のまま）
  SlackChannelConfig:
    Type: AWS::Chatbot::SlackChannelConfiguration
    Properties:
      ConfigurationName: ASR-Notifications
      SlackWorkspaceId: !Ref SlackWorkspaceId
      SlackChannelId: !Ref SlackChannelId
      IamRoleArn: !GetAtt ChatbotRole.Arn
      SnsTopicArns:
        - !Ref NotificationTopic
      LoggingLevel: INFO

  # フォーマッター Lambda の IAM ロール
  FormatterLambdaRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
      Policies:
        - PolicyName: SnsPublish
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action: sns:Publish
                Resource: !Ref NotificationTopic

  # フォーマッター Lambda
  FormatterLambda:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: ASR-NotificationFormatter
      Runtime: python3.13
      Handler: index.handler
      Timeout: 30
      MemorySize: 128
      Role: !GetAtt FormatterLambdaRole.Arn
      Environment:
        Variables:
          ASR_NOTIFICATION_TOPIC_ARN: !Ref NotificationTopic
      Code:
        ZipFile: |
          """ASR通知フォーマッターLambda."""

          import json
          import os
          import urllib.parse

          import boto3

          sns_client = boto3.client("sns")
          TOPIC_ARN = os.environ["ASR_NOTIFICATION_TOPIC_ARN"]


          def handler(event, context):
              processed = 0
              skipped = 0

              for record in event.get("Records", []):
                  raw_message = record["Sns"]["Message"]
                  if _process_message(raw_message):
                      processed += 1
                  else:
                      skipped += 1

              return {"processed": processed, "skipped": skipped}


          def _process_message(raw_message):
              parsed = json.loads(raw_message)

              if "AlarmName" in parsed:
                  return _handle_alarm(parsed)
              if "Finding" in parsed:
                  return _handle_native_remediation(parsed)
              if "finding" in parsed:
                  return _handle_remediation(parsed)
              return False


          def _handle_native_remediation(parsed):
              """実環境で発行されるPascal_Snake_Case形式を処理する."""
              status = parsed.get("Remediation_Status", "")
              if status == "SUCCESS":
                  emoji, label, keyword = ":white_check_mark:", "自動修復成功", "SUCCESS"
              elif status == "FAILED":
                  emoji, label, keyword = ":x:", "自動修復失敗", "FAILED"
              else:
                  return False

              finding = parsed.get("Finding", {})
              return _build_and_publish_remediation(
                  finding, parsed.get("Severity", ""), parsed.get("Message", ""),
                  emoji, label, keyword,
              )


          def _handle_remediation(parsed):
              """公式ドキュメント記載の小文字形式を処理する."""
              message = parsed.get("message", "")
              finding = parsed.get("finding", {})
              severity = parsed.get("severity", "")

              if "Remediation succeeded" in message:
                  emoji, label, keyword = ":white_check_mark:", "自動修復成功", "SUCCESS"
              elif "Remediation failed" in message:
                  emoji, label, keyword = ":x:", "自動修復失敗", "FAILED"
              else:
                  return False

              return _build_and_publish_remediation(
                  finding, severity, message, emoji, label, keyword,
              )


          def _build_and_publish_remediation(finding, severity, message, emoji, label, keyword):
              control = finding.get("standard_control", "")
              account = finding.get("account", "")
              region = finding.get("region", "")
              finding_arn = finding.get("finding_arn", "")
              encoded_arn = urllib.parse.quote(finding_arn, safe="")
              finding_url = (
                  f"https://{region}.console.aws.amazon.com"
                  f"/securityhub/home?region={region}"
                  f"#/findings?search=Id%3D%5Coperator%5CEQUALS%5C{encoded_arn}"
              )

              desc_lines = [
                  f"*{finding.get('title', '')}*",
                  "",
                  f">  アカウント: `{account}`",
                  f">  リージョン: `{region}`",
                  f">  コントロール: `{control}` ({finding.get('standard_name', '')} v{finding.get('standard_version', '')})",
              ]

              notification = {
                  "version": "1.0",
                  "source": "custom",
                  "content": {
                      "textType": "client-markdown",
                      "title": f"{emoji} ASR {label}: {control}",
                      "description": "\n".join(desc_lines),
                      "nextSteps": [f"<{finding_url}|Security Hub CSPM で確認>"],
                      "keywords": [control, account, keyword],
                  },
                  "metadata": {
                      "threadId": f"asr-{account}",
                      "summary": f"ASR {label}: {control} ({account})",
                      "additionalContext": {
                          "findingArn": finding_arn,
                          "severity": severity,
                      },
                  },
              }
              _publish(notification)
              return True


          def _handle_alarm(parsed):
              alarm_name = parsed.get("AlarmName", "")
              state = parsed.get("NewStateValue", "")

              if state == "ALARM":
                  emoji, label, keyword = ":rotating_light:", "ASR アラーム発生", "ALARM"
              elif state == "OK":
                  emoji, label, keyword = ":white_check_mark:", "ASR アラーム復旧", "OK"
              else:
                  emoji, label, keyword = ":warning:", "ASR アラーム データ不足", "INSUFFICIENT_DATA"

              desc_lines = [
                  f"*{alarm_name}*",
                  "",
                  f">  状態: `{state}`",
                  f">  理由: {parsed.get('NewStateReason', '')}",
              ]
              if parsed.get("AlarmDescription"):
                  desc_lines.append(f">  説明: {parsed['AlarmDescription']}")
              if parsed.get("AWSAccountId"):
                  desc_lines.append(f">  アカウント: `{parsed['AWSAccountId']}`")

              notification = {
                  "version": "1.0",
                  "source": "custom",
                  "content": {
                      "textType": "client-markdown",
                      "title": f"{emoji} {label}: {alarm_name}",
                      "description": "\n".join(desc_lines),
                      "nextSteps": [],
                      "keywords": [alarm_name, keyword],
                  },
                  "metadata": {
                      "threadId": "asr-alarm",
                      "summary": f"{label}: {alarm_name}",
                  },
              }
              _publish(notification)
              return True


          def _publish(notification):
              sns_client.publish(
                  TopicArn=TOPIC_ARN,
                  Message=json.dumps(notification),
              )

  # Lambda 呼び出し権限（SNS → Lambda）
  LambdaSnsPermission:
    Type: AWS::Lambda::Permission
    Properties:
      FunctionName: !Ref FormatterLambda
      Action: lambda:InvokeFunction
      Principal: sns.amazonaws.com

  # ASR Topic → Lambda サブスクリプション
  AsrTopicSubscription:
    Type: AWS::SNS::Subscription
    Properties:
      TopicArn: !Ref AsrTopicArn
      Protocol: lambda
      Endpoint: !GetAtt FormatterLambda.Arn

  AsrAlarmTopicSubscription:
    Type: AWS::SNS::Subscription
    Properties:
      TopicArn: !Ref AsrAlarmTopicArn
      Protocol: lambda
      Endpoint: !GetAtt FormatterLambda.Arn

Outputs:
  NotificationTopicArn:
    Value: !Ref NotificationTopic
    Description: Amazon Q Developer連携用SNS Topic ARN
  FormatterLambdaArn:
    Value: !GetAtt FormatterLambda.Arn
    Description: フォーマッターLambda ARN
  SlackChannelConfigArn:
    Value: !Ref SlackChannelConfig
    Description: Amazon Q Developer Slack チャンネル設定 ARN