Duo 認証ログを収集する

以下でサポートされています。

Google SecOps SIEM

このドキュメントでは、Duo 認証ログを Google Security Operations に取り込む方法について説明します。パーサーは、JSON 形式のメッセージからログを抽出します。未加工のログデータを統合データモデル（UDM）に変換し、ユーザー、デバイス、アプリケーション、ロケーション、認証の詳細などのフィールドをマッピングします。また、さまざまな認証要素と結果を処理して、セキュリティイベントを分類します。また、パーサーは、データ品質と整合性を確保するために、データのクレンジング、型変換、エラー処理も行います。

次の 2 つの収集方法から選択します。

オプション 1: サードパーティ API を使用した直接取り込み
オプション 2: Cloud Run 関数と Google Cloud Storage を使用してログを収集する

始める前に

次の前提条件を満たしていることを確認します。

Google SecOps インスタンス
Duo 管理パネルへの特権アクセス（Admin API アプリケーションを作成するにはオーナーロールが必要）
オプション 2 を使用している場合は、GCP への特権アクセス

オプション 1: サードパーティ API を使用して Duo 認証ログを取り込む

Duo の前提条件（API 認証情報）を収集する

オーナー、管理者、アプリケーションマネージャーのいずれかのロールを持つ管理者として Duo 管理パネルにログインします。
[Applications] > [Application Catalog] に移動します。
カタログで Admin API のエントリを見つけます。
[+ 追加] をクリックして、アプリケーションを作成します。
次の詳細をコピーして安全な場所に保存します。
- 統合キー
- 秘密鍵
- API ホスト名（例: api-XXXXXXXX.duosecurity.com）
[権限] セクションに移動します。
[読み取りログの付与] 以外のすべての権限オプションの選択を解除します。
[変更を保存] をクリックします。

Duo 認証ログを取り込むように Google SecOps でフィードを構成する

[SIEM 設定] > [フィード] に移動します。
[+ 新しいフィードを追加] をクリックします。
[フィード名] フィールドに、フィードの名前を入力します（例: Duo Authentication Logs）。
[ソースタイプ] として [サードパーティ API] を選択します。
[ログタイプ] として [Duo Auth] を選択します。
[次へ] をクリックします。
次の入力パラメータの値を指定します。
- Username: Duo の統合キーを入力します。
- Secret: Duo のシークレットキーを入力します。
- API Hostname: API ホスト名（例: api-XXXXXXXX.duosecurity.com）を入力します。
- アセットの名前空間: 省略可。アセットの名前空間。
- 取り込みラベル: 省略可。このフィードのイベントに適用されるラベル。
[次へ] をクリックします。
[Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。

オプション 2: Google Cloud Storage を使用して Duo 認証ログを取り込む

Duo Admin API の認証情報を収集する

Duo 管理パネルにログインします。
[Applications] > [Application Catalog] に移動します。
アプリケーションカタログで Admin API を見つけます。
[+ 追加] をクリックして、Admin API アプリケーションを追加します。
次の値をコピーして保存します。
- インテグレーションキー（ikey）
- 秘密鍵（skey）
- API ホスト名（例: api-XXXXXXXX.duosecurity.com）
[権限] で、[読み取りログの付与] を有効にします。
[変更を保存] をクリックします。

Google Cloud Storage バケットを作成する

Google Cloud Console に移動します。
プロジェクトを選択するか、新しいプロジェクトを作成します。
ナビゲーションメニューで、[Cloud Storage > バケット] に移動します。
[バケットを作成] をクリックします。

次の構成情報を提供してください。

設定	値
バケットに名前を付ける	グローバルに一意の名前（`duo-auth-logs` など）を入力します。
ロケーションタイプ	ニーズに基づいて選択します（リージョン、デュアルリージョン、マルチリージョン）。
ロケーション	ロケーションを選択します（例: `us-central1`）。
ストレージクラス	Standard（頻繁にアクセスされるログにおすすめ）
アクセス制御	均一（推奨）
保護ツール	省略可: オブジェクトのバージョニングまたは保持ポリシーを有効にする

[作成] をクリックします。

Cloud Run functions のサービスアカウントを作成する

Cloud Run 関数には、GCS バケットに書き込む権限を持つサービスアカウントが必要です。

サービスアカウントの作成

GCP Console で、[IAM と管理>サービスアカウント] に移動します。
[サービスアカウントを作成] をクリックします。
次の構成の詳細を指定します。
- サービスアカウント名: 「duo-auth-collector-sa」と入力します。
- サービスアカウントの説明: 「Service account for Cloud Run function to collect Duo authentication logs」と入力します。
[作成して続行] をクリックします。
[このサービスアカウントにプロジェクトへのアクセスを許可する] セクションで、次のロールを追加します。
1. [ロールを選択] をクリックします。
2. [ストレージオブジェクト管理者] を検索して選択します。
3. [+ 別のロールを追加] をクリックします。
4. [Cloud Run 起動元] を検索して選択します。
5. [+ 別のロールを追加] をクリックします。
6. [Cloud Functions 起動元] を検索して選択します。
[続行] をクリックします。
[完了] をクリックします。

これらのロールは、次の目的で必要です。

Storage オブジェクト管理者: ログを GCS バケットに書き込み、状態ファイルを管理する
Cloud Run 起動元: Pub/Sub が関数を呼び出すことを許可する
Cloud Functions 起動元: 関数の呼び出しを許可する

GCS バケットに対する IAM 権限を付与する

GCS バケットに対する書き込み権限をサービスアカウントに付与します。

[Cloud Storage] > [バケット] に移動します。
バケット名をクリックします。
[権限] タブに移動します。
[アクセス権を付与] をクリックします。
次の構成の詳細を指定します。
- プリンシパルを追加: サービスアカウントのメールアドレス（例: duo-auth-collector-sa@PROJECT_ID.iam.gserviceaccount.com）を入力します。
- ロールを割り当てる: [Storage オブジェクト管理者] を選択します。
[保存] をクリックします。

Pub/Sub トピックの作成

Cloud Scheduler がパブリッシュし、Cloud Run functions がサブスクライブする Pub/Sub トピックを作成します。

GCP Console で、[Pub/Sub> トピック] に移動します。
[トピックを作成] をクリックします。
次の構成の詳細を指定します。
- トピック ID: 「duo-auth-trigger」と入力します。
- その他の設定はデフォルトのままにします。
[作成] をクリックします。

ログを収集する Cloud Run 関数を作成する

Cloud Run 関数は、Cloud Scheduler からの Pub/Sub メッセージによってトリガーされ、Duo Admin API からログを取得して GCS に書き込みます。

GCP Console で、[Cloud Run] に移動します。
[サービスを作成] をクリックします。
[関数] を選択します（インラインエディタを使用して関数を作成します）。

[構成] セクションで、次の構成の詳細を指定します。

設定	値
サービス名	`duo-auth-collector`
リージョン	GCS バケットと一致するリージョンを選択します（例: `us-central1`）。
ランタイム	[Python 3.12] 以降を選択します。

[トリガー（省略可）] セクションで、次の操作を行います。
1. [+ トリガーを追加] をクリックします。
2. [Cloud Pub/Sub] を選択します。
3. [Cloud Pub/Sub トピックを選択してください] で、トピック duo-auth-trigger を選択します。
4. [保存] をクリックします。
[認証] セクションで、次の操作を行います。
1. [認証が必要] を選択します。
2. Identity and Access Management（IAM）を確認します。
注: Pub/Sub は、関数を呼び出すときに認証を自動的に処理します。
下にスクロールして、[コンテナ、ネットワーキング、セキュリティ] を開きます。
[セキュリティ] タブに移動します。
- サービスアカウント: サービスアカウント duo-auth-collector-sa を選択します。

[コンテナ] タブに移動します。

[変数とシークレット] をクリックします。
環境変数ごとに [+ 変数を追加] をクリックします。

変数名	値の例
`GCS_BUCKET`	`duo-auth-logs`
`GCS_PREFIX`	`duo/auth/`
`STATE_KEY`	`duo/auth/state.json`
`DUO_IKEY`	`DIXYZ...`
`DUO_SKEY`	`****************`
`DUO_API_HOSTNAME`	`api-XXXXXXXX.duosecurity.com`
`LIMIT`	`500`

[変数とシークレット] タブで [リクエスト] まで下にスクロールします。
- リクエストのタイムアウト: 600 秒（10 分）を入力します。
[コンテナ] の [設定] タブに移動します。
- [リソース] セクションで次の操作を行います。
  - メモリ: 512 MiB 以上を選択します。
  - CPU: [1] を選択します。
- [完了] をクリックします。
[実行環境] までスクロールします。
- [デフォルト]（推奨）を選択します。
[リビジョンスケーリング] セクションで、次の操作を行います。
- [インスタンスの最小数] に「0」と入力します。
- インスタンスの最大数: 100 と入力します（または、予想される負荷に基づいて調整します）。
[作成] をクリックします。
サービスが作成されるまで待ちます（1 ～ 2 分）。
サービスを作成すると、インラインコードエディタが自動的に開きます。

関数コードを追加する

[関数のエントリポイント] に「main」と入力します。

インラインコードエディタで、次の 2 つのファイルを作成します。

最初のファイル: main.py:

#!/usr/bin/env python3
# Cloud Run Function: Pull Duo Admin API v2 Authentication Logs to GCS (raw JSON pages)
# Notes:
# - Duo v2 requires mintime/maxtime in *milliseconds* (13-digit epoch).
# - Pagination via metadata.next_offset ("<millis>,<txid>").
# - We save state (mintime_ms) in ms to resume next run without gaps.

import functions_framework
from google.cloud import storage
import os
import json
import time
import hmac
import hashlib
import base64
import email.utils
import urllib.parse
from urllib.request import Request, urlopen
from urllib.error import HTTPError, URLError

DUO_IKEY = os.environ["DUO_IKEY"]
DUO_SKEY = os.environ["DUO_SKEY"]
DUO_API_HOSTNAME = os.environ["DUO_API_HOSTNAME"].strip()
GCS_BUCKET = os.environ["GCS_BUCKET"]
GCS_PREFIX = os.environ.get("GCS_PREFIX", "duo/auth/").strip("/")
STATE_KEY = os.environ.get("STATE_KEY", "duo/auth/state.json")
LIMIT = min(int(os.environ.get("LIMIT", "500")), 1000)  # default 500, max 1000

storage_client = storage.Client()

def _canon_params(params: dict) -> str:
    parts = []
    for k in sorted(params.keys()):
        v = params[k]
        if v is None:
            continue
        parts.append(f"{urllib.parse.quote(str(k), '~')}={urllib.parse.quote(str(v), '~')}")
    return "&".join(parts)

def _sign(method: str, host: str, path: str, params: dict) -> dict:
    now = email.utils.formatdate()
    canon = "\n".join([
        now,
        method.upper(),
        host.lower(),
        path,
        _canon_params(params)
    ])
    sig = hmac.new(
        DUO_SKEY.encode("utf-8"),
        canon.encode("utf-8"),
        hashlib.sha1
    ).hexdigest()
    auth = base64.b64encode(f"{DUO_IKEY}:{sig}".encode()).decode()
    return {
        "Date": now,
        "Authorization": f"Basic {auth}"
    }

def _http(method: str, path: str, params: dict, timeout: int = 60, max_retries: int = 5) -> dict:
    host = DUO_API_HOSTNAME
    assert host.startswith("api-") and host.endswith(".duosecurity.com"), \
        "DUO_API_HOSTNAME must be like api-XXXXXXXX.duosecurity.com"

    qs = _canon_params(params)
    url = f"https://{host}{path}" + (f"?{qs}" if qs else "")

    attempt, backoff = 0, 1.0
    while True:
        req = Request(url, method=method.upper())
        req.add_header("Accept", "application/json")
        for k, v in _sign(method, host, path, params).items():
            req.add_header(k, v)

        try:
            with urlopen(req, timeout=timeout) as r:
                return json.loads(r.read().decode("utf-8"))
        except HTTPError as e:
            if (e.code == 429 or 500 <= e.code <= 599) and attempt < max_retries:
                time.sleep(backoff)
                attempt += 1
                backoff *= 2
                continue
            raise
        except URLError:
            if attempt < max_retries:
                time.sleep(backoff)
                attempt += 1
                backoff *= 2
                continue
            raise

def _read_state_ms() -> int | None:
    try:
        bucket = storage_client.bucket(GCS_BUCKET)
        blob = bucket.blob(STATE_KEY)
        if blob.exists():
            state_data = blob.download_as_text()
            val = json.loads(state_data).get("mintime")
            if val is None:
                return None
            # Backward safety: if seconds were stored, convert to ms
            return int(val) * 1000 if len(str(int(val))) <= 10 else int(val)
    except Exception:
        return None

def _write_state_ms(mintime_ms: int):
    bucket = storage_client.bucket(GCS_BUCKET)
    blob = bucket.blob(STATE_KEY)
    body = json.dumps({"mintime": int(mintime_ms)}).encode("utf-8")
    blob.upload_from_string(body, content_type="application/json")

def _write_page(payload: dict, when_epoch_s: int, page: int) -> str:
    bucket = storage_client.bucket(GCS_BUCKET)
    key = f"{GCS_PREFIX}/{time.strftime('%Y/%m/%d', time.gmtime(when_epoch_s))}/duo-auth-{page:05d}.json"
    blob = bucket.blob(key)
    blob.upload_from_string(
        json.dumps(payload, separators=(",", ":")).encode("utf-8"),
        content_type="application/json"
    )
    return key

def fetch_and_store():
    now_s = int(time.time())
    # Duo recommends a ~2-minute delay buffer; use maxtime = now - 120 seconds (in ms)
    maxtime_ms = (now_s - 120) * 1000
    mintime_ms = _read_state_ms() or (maxtime_ms - 3600 * 1000)  # 1 hour on first run

    page = 0
    total = 0
    next_offset = None

    while True:
        params = {
            "mintime": mintime_ms,
            "maxtime": maxtime_ms,
            "limit": LIMIT
        }
        if next_offset:
            params["next_offset"] = next_offset

        data = _http("GET", "/admin/v2/logs/authentication", params)
        _write_page(data, maxtime_ms // 1000, page)
        page += 1

        resp = data.get("response")
        items = resp if isinstance(resp, list) else []
        total += len(items)

        meta = data.get("metadata") or {}
        next_offset = meta.get("next_offset")
        if not next_offset:
            break

    # Advance window to maxtime_ms for next run
    _write_state_ms(maxtime_ms)

    return {
        "ok": True,
        "pages": page,
        "events": total,
        "next_mintime_ms": maxtime_ms
    }

@functions_framework.cloud_event
def main(cloud_event):
    """
    Cloud Run function triggered by Pub/Sub to fetch Duo authentication logs and write to GCS.

    Args:
        cloud_event: CloudEvent object containing Pub/Sub message
    """
    try:
        result = fetch_and_store()
        print(f"Successfully processed {result['events']} events in {result['pages']} pages")
        print(f"Next mintime_ms: {result['next_mintime_ms']}")
    except Exception as e:
        print(f"Error processing logs: {str(e)}")
        raise

2 つ目のファイル: requirements.txt:

functions-framework==3.*
google-cloud-storage==2.*

[デプロイ] をクリックして、関数を保存してデプロイします。
デプロイが完了するまで待ちます（2 ～ 3 分）。

注: Pub/Sub トリガー構成では、必要なサブスクリプションと権限が自動的に作成されます。

Cloud Scheduler ジョブの作成

Cloud Scheduler は、定期的に Pub/Sub トピックにメッセージをパブリッシュし、Cloud Run functions の関数をトリガーします。

GCP Console で、[Cloud Scheduler] に移動します。
[ジョブを作成] をクリックします。

次の構成情報を提供してください。

設定	値
名前	`duo-auth-collector-hourly`
リージョン	Cloud Run functions と同じリージョンを選択する
周波数	`0 * * * *`（1 時間ごとに正時）
タイムゾーン	タイムゾーンを選択します（UTC を推奨）。
ターゲットタイプ	Pub/Sub
トピック	トピック `duo-auth-trigger` を選択します。
メッセージ本文	`{}`（空の JSON オブジェクト）

[作成] をクリックします。

スケジュールの頻度のオプション

ログの量とレイテンシの要件に基づいて頻度を選択します。

頻度	CRON 式	ユースケース
5 分毎	`/5 * * *`	大容量、低レイテンシ
15 分ごと	`/15 * * *`	検索量が普通
1 時間ごと	`0 * * * *`	標準（推奨）
6 時間ごと	`0 /6 * *`	少量、バッチ処理
毎日	`0 0 * * *`	履歴データの収集

スケジューラジョブをテストする

Cloud Scheduler コンソールで、ジョブを見つけます。
[強制実行] をクリックして手動でトリガーします。
数秒待ってから、[Cloud Run> サービス> duo-auth-collector > ログ] に移動します。
関数が正常に実行されたことを確認します。
GCS バケットをチェックして、ログが書き込まれたことを確認します。

Google SecOps サービスアカウントを取得する

Google SecOps は、一意のサービスアカウントを使用して GCS バケットからデータを読み取ります。このサービスアカウントにバケットへのアクセス権を付与する必要があります。

サービスアカウントのメールアドレスを取得する

[SIEM 設定] > [フィード] に移動します。
[Add New Feed] をクリックします。
[単一フィードを設定] をクリックします。
[フィード名] フィールドに、フィードの名前を入力します（例: Duo Authentication Logs）。
[ソースタイプ] として [Google Cloud Storage V2] を選択します。
[ログタイプ] として [Duo Auth] を選択します。
[サービスアカウントを取得する] をクリックします。一意のサービスアカウントメールアドレスが表示されます（例:）。
```
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
```
このメールアドレスをコピーして、次のステップで使用します。

注: 各 Google SecOps インスタンスには一意のサービスアカウントがあります。他のドキュメントや例のサービスアカウントは使用しないでください。

Google SecOps サービスアカウントに IAM 権限を付与する

Google SecOps サービスアカウントには、GCS バケットに対する Storage オブジェクト閲覧者ロールが必要です。

[Cloud Storage] > [バケット] に移動します。
バケット名をクリックします。
[権限] タブに移動します。
[アクセス権を付与] をクリックします。
次の構成の詳細を指定します。
- プリンシパルを追加: Google SecOps サービスアカウントのメールアドレスを貼り付けます。
- ロールを割り当てる: [ストレージオブジェクト閲覧者] を選択します。
[保存] をクリックします。

注: 削除オプションの [転送されたファイルを削除] または [転送されたファイルと空のディレクトリを削除] を使用する場合は、Storage オブジェクト閲覧者ではなく、Storage オブジェクト管理者ロールを付与します。

Duo 認証ログを取り込むように Google SecOps でフィードを構成する

[SIEM 設定] > [フィード] に移動します。
[Add New Feed] をクリックします。
[単一フィードを設定] をクリックします。
[フィード名] フィールドに、フィードの名前を入力します（例: Duo Authentication Logs）。
[ソースタイプ] として [Google Cloud Storage V2] を選択します。
[ログタイプ] として [Duo Auth] を選択します。
[次へ] をクリックします。
次の入力パラメータの値を指定します。
- ストレージバケットの URL: 接頭辞パスを含む GCS バケット URI を入力します。
```
gs://duo-auth-logs/duo/auth/
```
  - 次のように置き換えます。
    - duo-auth-logs: GCS バケット名。
    - duo/auth/: ログが保存されるオプションの接頭辞/フォルダパス（ルートの場合は空のままにします）。
  - 例:
    - ルートバケット: gs://company-logs/
    - 接頭辞あり: gs://company-logs/duo-logs/
    - サブフォルダあり: gs://company-logs/duo/auth/
  注: URI の末尾には常にスラッシュ（/）を付けます。
- Source deletion option: 必要に応じて削除オプションを選択します。
  - なし: 転送後にファイルを削除しません（テストにおすすめ）。
  - 転送されたファイルを削除する: 転送が完了した後にファイルを削除します。
  - 転送されたファイルと空のディレクトリを削除する: 転送が完了した後にファイルと空のディレクトリを削除します。
    
    注: 削除オプションを選択する場合は、サービスアカウントに Storage オブジェクト閲覧者ではなく、Storage オブジェクト管理者のロールが必要です。必要に応じて IAM 権限を更新します。
- ファイルの最大経過日数: 指定した日数以内に変更されたファイルを含めます。デフォルトは 180 日です。
- アセットの名前空間: アセットの名前空間。
- Ingestion labels: このフィードのイベントに適用されるラベル。
[次へ] をクリックします。
[Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。

UDM マッピングテーブル

ログフィールド	UDM マッピング	論理
access_device.browser	target.resource.attribute.labels.value	access_device.browser が存在する場合、その値は UDM にマッピングされます。
access_device.hostname	principal.hostname	access_device.hostname が存在し、空でない場合、その値は UDM にマッピングされます。空で、event_type が USER_CREATION の場合、event_type は USER_UNCATEGORIZED に変更されます。access_device.hostname が空で、hostname フィールドが存在する場合、hostname の値が使用されます。
access_device.ip	principal.ip	access_device.ip が存在し、有効な IPv4 アドレスである場合、その値は UDM にマッピングされます。有効な IPv4 アドレスでない場合は、キー access_device.ip を持つ additional.fields に文字列値として追加されます。
access_device.location.city	principal.location.city	存在する場合、値は UDM にマッピングされます。
access_device.location.country	principal.location.country_or_region	存在する場合、値は UDM にマッピングされます。
access_device.location.state	principal.location.state	存在する場合、値は UDM にマッピングされます。
access_device.os	principal.platform	存在する場合、値は対応する UDM 値（MAC、WINDOWS、LINUX）に変換されます。
access_device.os_version	principal.platform_version	存在する場合、値は UDM にマッピングされます。
application.key	target.resource.id	存在する場合、値は UDM にマッピングされます。
application.name	target.application	存在する場合、値は UDM にマッピングされます。
auth_device.ip	target.ip	存在し、「None」でない場合、値は UDM にマッピングされます。
auth_device.location.city	target.location.city	存在する場合、値は UDM にマッピングされます。
auth_device.location.country	target.location.country_or_region	存在する場合、値は UDM にマッピングされます。
auth_device.location.state	target.location.state	存在する場合、値は UDM にマッピングされます。
auth_device.name	target.hostname または target.user.phone_numbers	auth_device.name が存在し、正規化後の電話番号である場合、target.user.phone_numbers に追加されます。それ以外の場合は、target.hostname にマッピングされます。
client_ip	target.ip	存在し、「None」でない場合、値は UDM にマッピングされます。
client_section	target.resource.attribute.labels.value	client_section が存在する場合、その値はキー client_section を使用して UDM にマッピングされます。
dn	target.user.userid	dn が存在し、user.name と username が存在しない場合、userid は grok を使用して dn フィールドから抽出され、UDM にマッピングされます。event_type は USER_LOGIN に設定されます。
event_type	metadata.product_event_type AND metadata.event_type	この値は metadata.product_event_type にマッピングされます。また、metadata.event_type の決定にも使用されます。「authentication」は USER_LOGIN に、「enrollment」は USER_CREATION になります。空の場合や、どちらでもない場合は、GENERIC_EVENT になります。
要素	extensions.auth.mechanism と extensions.auth.auth_details	値は、対応する UDM auth.mechanism 値（HARDWARE_KEY、REMOTE_INTERACTIVE、LOCAL、OTP）に変換されます。元の値は extensions.auth.auth_details にもマッピングされます。
hostname	principal.hostname	存在し、access_device.hostname が空の場合、値は UDM にマッピングされます。
log_format	target.resource.attribute.labels.value	log_format が存在する場合、その値はキー log_format を使用して UDM にマッピングされます。
loglevel._classuuid_	target.resource.attribute.labels.value	loglevel._classuuid_ が存在する場合、その値はキー class_uuid を使用して UDM にマッピングされます。
log_level.name	target.resource.attribute.labels.value AND security_result.severity	log_level.name が存在する場合、その値はキー名を使用して UDM にマッピングされます。値が「info」の場合、security_result.severity は INFORMATIONAL に設定されます。
log_logger.unpersistable	target.resource.attribute.labels.value	log_logger.unpersistable が存在する場合、その値はキー unpersistable を使用して UDM にマッピングされます。
log_namespace	target.resource.attribute.labels.value	log_namespace が存在する場合、その値はキー log_namespace を使用して UDM にマッピングされます。
log_source	target.resource.attribute.labels.value	log_source が存在する場合、その値はキー log_source を使用して UDM にマッピングされます。
msg	security_result.summary	存在し、理由が空の場合、値は UDM にマッピングされます。
reason	security_result.summary	存在する場合、値は UDM にマッピングされます。
件の結果	security_result.action_details と security_result.action	存在する場合、値は security_result.action_details にマッピングされます。「success」または「SUCCESS」は security_result.action ALLOW に変換され、それ以外の場合は BLOCK に変換されます。
server_section	target.resource.attribute.labels.value	server_section が存在する場合、その値はキー server_section を使用して UDM にマッピングされます。
server_section_ikey	target.resource.attribute.labels.value	server_section_ikey が存在する場合、その値はキー server_section_ikey を持つ UDM にマッピングされます。
ステータス	security_result.action_details と security_result.action	存在する場合、値は security_result.action_details にマッピングされます。「許可」は security_result.action ALLOW に、「拒否」は BLOCK に変換されます。
timestamp	metadata.event_timestamp AND event.timestamp	値はタイムスタンプに変換され、metadata.event_timestamp と event.timestamp の両方にマッピングされます。
txid	metadata.product_log_id AND network.session_id	この値は、metadata.product_log_id と network.session_id の両方にマッピングされます。
user.groups	target.user.group_identifiers	配列内のすべての値が target.user.group_identifiers に追加されます。
user.key	target.user.product_object_id	存在する場合、値は UDM にマッピングされます。
user.name	target.user.userid	存在する場合、値は UDM にマッピングされます。
ユーザー名	target.user.userid	存在し、user.name が存在しない場合、値は UDM にマッピングされます。event_type は USER_LOGIN に設定されます。
（パーサーロジック）	metadata.vendor_name	常に「DUO_SECURITY」に設定されます。
（パーサーロジック）	metadata.product_name	常に「MULTI-FACTOR_AUTHENTICATION」に設定されます。
（パーサーロジック）	metadata.log_type	未加工ログの最上位の log_type フィールドから取得されます。
（パーサーロジック）	extensions.auth.type	常に「SSO」に設定されます。

ご不明な点がございましたら、コミュニティメンバーや Google SecOps のプロフェッショナルから回答を得ることができます。

Duo 認証ログを収集する

始める前に

オプション 1: サードパーティ API を使用して Duo 認証ログを取り込む

Duo の前提条件（API 認証情報）を収集する

Duo 認証ログを取り込むように Google SecOps でフィードを構成する

オプション 2: Google Cloud Storage を使用して Duo 認証ログを取り込む

Duo Admin API の認証情報を収集する

Google Cloud Storage バケットを作成する

Cloud Run functions のサービス アカウントを作成する

サービス アカウントの作成

GCS バケットに対する IAM 権限を付与する

Pub/Sub トピックの作成

ログを収集する Cloud Run 関数を作成する

関数コードを追加する

Cloud Scheduler ジョブの作成

スケジュールの頻度のオプション

スケジューラ ジョブをテストする

Google SecOps サービス アカウントを取得する

サービス アカウントのメールアドレスを取得する

Google SecOps サービス アカウントに IAM 権限を付与する

Duo 認証ログを取り込むように Google SecOps でフィードを構成する

UDM マッピング テーブル

Cloud Run functions のサービスアカウントを作成する

サービスアカウントの作成

スケジューラジョブをテストする

Google SecOps サービスアカウントを取得する

サービスアカウントのメールアドレスを取得する

Google SecOps サービスアカウントに IAM 権限を付与する

UDM マッピングテーブル