DMARCレポートの可視化ダッシュボードを作りました

はじめに

インフラGの鈴木です。ガールズケイリンアニメことリンカイ！の放映が近くなってきましたね。
最近小倉にギャンブル旅行にいったのですが、北九州競輪には等身大パネルがありました。本気(マジ)度が伝わってきます。アニメの放映日が楽しみです。
ところで、今回はDMARCの可視化基盤を作った話をします。なかなか大変¹でしたので、共有したいと思います。

かっこいいダッシュボード

そもそもDMARCって何？

DMARCは一言で言うと、メールの送信元を認証し、送信元メールアドレス（ドメイン）のなりすましを防ぐ技術です。
この仕様そのものがまず難しいです。ここでは詳しく解説しませんが、DMARCそのものについての理解や、DMARCの仕様にあるアライメントについての理解が求められました。これらについてはリンク先の記事が一番わかりやすかったです。
理解のし辛さを解説すると、DMARCの前提知識としてSPFとDKIMがあります。そうなるとまずメールの基本となる以下の理解が必要です。

• ヘッダーFROM
• エンベローブFROM

さらに、SPFとDKIMでは保証できない部分をDMARCの以下の仕様で保証するという仕組みです。

• SPFアライメント
• DKIMアライメント

複雑な仕様の理由は、メールの仕様(SMTP)が古く、悪意のあるメールを送ることが簡単にできてしまうため、その対策として色々な仕様が生まれたという経緯があります。

Googleの発表によってDMARC対応が必要に

去年の10月にGoogleがメールを大量に送信²する事業者に対してDMARC対応を求めると発表しました。ちゃんと設定しないとGmailにメールが届かなくなるらしく、ビジネスへのインパクトが大きいです。
SPFとDKIMの設定はメールを送っている事業者なら大体設定していると思いますが、DMARCは当時まだマイナーだった印象です。実際リブセンスにおいても、Googleの発表があるまでは、DMARCの設定はしていませんでした。なのでこれを機に全てのメールを送っているドメインを確認し、必要なドメインに対してDMARCの設定を行いました。
DMARCレコードを設定することによって、XMLレポートが送られてくるようになります。モノによっては500万行以上あるXMLもあり、かなりの頻度で送られてくるので、人間で見るのは不可能です。
そこで、このレポートの可視化基盤を作ることにしました。これを作ることによってわかることは以下です。

• ドメイン名を悪用し、悪意のあるメールを送っている送信元を特定できる
• ドメインの認証結果(SPF/DKIMの認証がPassしているかどうか)が確認できる

これらを確認し、問題があった場合、対策を打つことでメールの信頼性を向上させることができます。

SaaSの検討

レポートを可視化するために、SaaSを使うことを検討しました。Googleのガイドラインに該当するドメインは社内で5つでしたので、5ドメインとして価格を試算しました。
検討段階で出てきたSaaSをいくつか紹介³します。

製品名	ドメイン数	価格	月のレポート受信数上限
dmarcian	15ドメインまで	月$499	5,000,000
PowerDMARC	5ドメインまで	月$8 – $250	2,000,000
SimpleDMARC	1ドメイン	月$99	無制限

いくつかのSaaSを試してみましたが、どれもグラフィカルで使いやすかったです。SaaSによってはBIMIの設定を支援してくれるサービスもあり、メールがビジネスの根幹を占める企業にとっては、非常に便利なサービスだと思います。
SaaSの導入も、XMLレポートをSaaS側で受け取ることができれば良いので、実際の作業はDMARCレコードにレポートを送るためにメールアドレスを設定するだけでした。簡単でいいですね。
しかし、一つ問題がありました。月のXMLレポート受信数上限がSaaSの上限より多いという問題です。無制限のサービスもありますが、複数ドメイン取り扱うと高くつきます。こうなっては自前で構築する方が安いと判断しました。

OSSの検討・選定

いくつかのOSSを調べてピックアップしましたが、開発が鈍化しているものが多く、継続開発がなされていそうなparsedmarcに着目しました。
その結果、parsedmarcを使用しているdmarc-visualizerというDocker Composeで動くDMARCの可視化ツールをベースとして、以下のOSSを組み合わせて構築しました。
技術選定は以下のようにしています。

やりたいこと	選定した方法
XMLを貯める	XMLレポートを受信する専用のGoogle Workspaceのアドレスを作り、GASで添付ファイルをGoogle Driveに日次で格納する
パースしてデータベースに入れる	Google Driveに格納したXMLファイルをparsedmarcでOpenSearchに格納する
データベースから取り出して可視化する	ECSにホストしたGrafanaからOpenSearchのデータを可視化する

構成

構成図は以下となります。

動作

GmailからGoogle Driveへ格納する

最初Gmail APIを叩いてXMLを取得しようとしましたが、色々検証した結果、GoogleのものはGoogleのもので取得した方が後々の運用が楽だと判断しました。
添付ファイルをGASで取得しようとしたところ、クラウドワークスさんのブログにやりたいことそのものがあったので、参考にしてGASを設定しました。感謝です。
このスクリプトを日次のトリガーで実行し、前日のXMLをGoogle Driveに格納することができるようにしました。

XMLをパースしてOpenSearchに格納する

EventBridge Schedulerを使用して日次でECSタスクをバッチ実行しました。バッチは前日のXMLをGoogle Driveから取得してパース、格納といった動きです。

Google Driveからコンテナ内にダウンロードする

CLIからGoogle Driveを操作するにはgdriveやskickaが有名ですが、リポジトリを見たところ両方アーカイブ化されており、頭を抱えました。
gdriveは後継が存在しますが、これらを使って欲しい機能を実現すると、シェル芸しそうな雰囲気を感じたのでPythonを書きました。

▼ Python

import os
import io
import os.path
import time
import boto3
import sys
import json
from googleapiclient.http import MediaIoBaseDownload
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
from datetime import datetime, timedelta

SCOPES = ['https://www.googleapis.com/auth/drive.readonly']
GOOGLE_DRIVE_FOLDER_ID = os.environ['GOOGLE_DRIVE_FOLDER_ID']
GOOGLE_DRIVE_ACCESS_TOKEN = json.loads(os.environ['GOOGLE_DRIVE_ACCESS_TOKEN'])
GOOGLE_DRIVE_ACCESS_TOKEN_SSM_PATH = os.environ['GOOGLE_DRIVE_ACCESS_TOKEN_SSM_PATH']

# local debug
if len(sys.argv) > 1:
    my_session = boto3.Session(profile_name=sys.argv[1])
    ssm_client = my_session.client('ssm')
else:
    ssm_client = boto3.client('ssm')

def download_file(service, file_id, file_name):
    # 保存場所を指定
    download_dir = '/app/inputs'
    file_path = os.path.join(download_dir, file_name)
    # ダウンロード処理
    request = service.files().get_media(fileId=file_id)
    fh = io.FileIO(file_path, 'wb')
    downloader = MediaIoBaseDownload(fh, request)
    done = False
    while done is False:
        status, done = downloader.next_chunk()
        print(f'Download progress for file {file_name}: {int(status.progress() * 100)}%')
    # ダウンロードしたファイルの存在チェック
    if os.path.exists(file_path):
        print(f'Download completed: {file_name}')
        return file_path  # return path of downloaded file
    else:
        print(f'Download failed: {file_name}')
        time.sleep(5)


def get_access_token():
    # https://developers.google.com/drive/api/quickstart/python?hl=ja
    # クイックスタートの認証を参考にした
    # トークンをSSMパラメータストア経由でECSから取得する
    creds = None
    creds = Credentials.from_authorized_user_info(GOOGLE_DRIVE_ACCESS_TOKEN, SCOPES)
    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
            # リフレッシュした場合パラメータストアを更新する
            response = ssm_client.put_parameter(
                Name=GOOGLE_DRIVE_ACCESS_TOKEN_SSM_PATH,
                Value=creds.to_json(),
                Type='SecureString',
                Overwrite=True
            )
            print(response)
    return creds


def get_target_folder(creds):
    service = build('drive', 'v3', credentials=creds)
    results = service.files().list(
        pageSize=1000, fields="nextPageToken, files(id, name)",
        q=f"'{GOOGLE_DRIVE_FOLDER_ID}' in parents").execute()
    items = results.get('files', [])
    if not items:
        print('No dmarc folder found.')
        sys.exit(1)
    return items

def get_yesterday_dmarc_report(creds, items):
    # 昨日のDMARCレポートを取得する
    service = build('drive', 'v3', credentials=creds)
    get_day = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
    print(f'google drive folders : {items}')
    for item in items:
        if item['name'] in get_day:
            target_folder_id = item['id']
    if not target_folder_id:
        print('No folder.')
        sys.exit(1)
    results = service.files().list(
        pageSize=1000, fields="nextPageToken, files(id, name)",
        q=f"'{target_folder_id}' in parents").execute()
    for prefix, i in enumerate(results.get('files', [])):
        print(f"file : {i['name']} id:{i['id']}")
        time.sleep(1)
        download_file(service, i['id'], f'{prefix}_{i["name"]}')

def main():
    creds = get_access_token()
    items = get_target_folder(creds)
    get_yesterday_dmarc_report(creds, items)


if __name__ == '__main__':
    main()

コメントにもありますが、認証周りはGoogle Workspace APIクイックスタートのコードがほぼそのまま使えたので、これを改変しました。 取得したアクセストークンをECSタスク定義のsecretsに登録し、環境変数として読み込むようにしました。更新したアクセストークンをSSMに再格納することでステートレスにしてます。

パースと格納

XMLレポートをダウンロードした後にparsedmarcを使ってパースし、OpenSearchに格納します。ちなみにOpenSearchの対応は最近でした。感謝です。

github.com

▼ 設定類

parsedmarcの設定

[general]
save_aggregate = True
save_forensic = True
output = /output/
debug = True
log_file = /dev/stdout

[opensearch]
hosts = https://hoehoge.hoge #実際はAWSに払い出されたEndpoint
ssl = False

Dockerfile

FROM python:3.11.8-alpine3.19

WORKDIR /app

RUN apk add --update --no-cache libxml2-dev libxslt-dev
RUN apk add --update --no-cache --virtual .build_deps build-base libffi-dev

COPY ./requirements.txt /app/requirements.txt
RUN pip install -r requirements.txt

COPY parsedmarc.ini /etc/parsedmarc.ini
COPY get_dmarc_reports.py /tmp/get_dmarc_reports.py
COPY run_commands.sh /app/run_commands.sh
RUN mkdir -p /app/inputs/ && \
    chmod +x /tmp/get_dmarc_reports.py && \
    chmod +x /app/run_commands.sh 

CMD ["/app/run_commands.sh"]

/app/run_commands.sh

#!/bin/sh
if [ -z "${GOOGLE_DRIVE_ACCESS_TOKEN}" ]; then
  echo "GOOGLE_DRIVE_ACCESS_TOKENが設定されていません。"
  exit 1
fi

if [ -z "${GOOGLE_DRIVE_FOLDER_ID}" ]; then
  echo "GOOGLE_DRIVE_FOLDER_IDが設定されていません。"
  exit 1
fi

python /tmp/get_dmarc_reports.py && parsedmarc -c /etc/parsedmarc.ini /app/inputs/*

可視化

OpenSearchに貯めたデータをECSにホストしたGrafanaを使って可視化しました。ダッシュボードはparsedmarcのリポジトリに既に作られているものがあったのでこちらを使いました。 認証に関してはGoogleのOAuth2認証を選択しました。これに関してGrafanaのドキュメントがとても丁寧で助かりました。ただroot_url(環境変数だとGF_SERVER_ROOT_URL)を正しく設定しないとうまくいかないのにハマりました。

You may have to set the root_url option of [server] for the callback URL to be correct. For example in case you are serving Grafana behind a proxy.

と、サラッと記載がありますが、これを設定しないとダメで、exampleコピペだけじゃ動かないので注意です。

苦労した点

Gmailの仕様とparsedmarcの相性が悪い

parsedmarcにはIMAP連携があります。これを使うとID/Passを設定するだけでXMLレポートを受信し次第、OpenSearchに自動で格納できます。しかしGoogleは今年の9月末にサードパーティとの連携についてIMAPをやめてOAuth2のみにすると発表しています。これでIMAP連携については詰みました。
代わりにGmail APIオプションもありますが、コンテナ環境においてOAuth2の認証は至難の業(ブラウザ開いてGoogle認証)です。
Docker環境においてparsedmarcを使うリポジトリでも同じ問題の会話がされており、Gmail APIを使う方法は諦めました。もしかしたらtokenを使い回すことでうまくできるのかも知れません。しかしDMARCレポートは最低1日1回送られてくる仕様であり、リアルタイム性は求めていないので、日次バッチの方式にしました。

OpenSearch突然データが全部消えた

シングルノードのOpenSearchで一週間運用したところ、突如データが全て揮発しました。

データが消滅

同一事象を調べたところ、シングルノードだとノードの更新や負荷がかかるタイミングでデータが揮発することがある様です。
データ再投入やスナップショットからの復元でリカバリーできるかと思い試しましたが、検証の結果、手間だったのでノードを2つにしました。 blog.ecbeing.tech

作ってみてよかったこと

SPF/DKIMの設定を最適化することができました。DMARCレポートが可視化されたことによって、SPFやDKIMの不合格がドメインごとに明確にわかるようになり、するべきことが浮き彫りになりました。これを展開することによって全社的にセキュアなメール設定をすることができました。

今後の運用

せっかく作ったものを腐らせてはもったいないので、ダッシュボードを積極的に活用する方法を考えていきたいです。まずアラートを作るところから始めていきたいと思います。

---

1. ブログを書くときのBGMは非常に大切ですので、今回は星街すいせいさんを聴きながら書きました↩
2. 1 日あたり 5,000 件以上のメールを送信する場合の話です↩
3. 2024/03/19現在の価格です↩