DomainTools Iris Investigate の結果を収集する

このドキュメントでは、Amazon S3 を使用して DomainTools Iris Investigate の結果を Google Security Operations に取り込む方法について説明します。パーサーは、DomainTools の Iris API からの未加工の JSON データを、Google SecOps の Unified Data Model（UDM）に準拠した構造化形式に変換します。ドメインの詳細、連絡先情報、セキュリティ リスク、SSL 証明書、その他の関連属性に関する情報を抽出し、対応する UDM フィールドにマッピングして、一貫した分析と脅威インテリジェンスを実現します。

始める前に

• Google SecOps インスタンス
• DomainTools エンタープライズ アカウントへの特権アクセス（Iris Investigate への API アクセス）
• AWS（S3、IAM、Lambda、EventBridge）への特権アクセス

DomainTools API キーとエンドポイントを取得する

1. DomainTools API ダッシュボードにログインします（API キーをリセットできるのは API 所有者アカウントのみです）。
2. [My Account] セクションで、[Account Summary] タブにある [View API Dashboard] リンクを選択します。
3. [API Username] セクションに移動して、ユーザー名を取得します。
4. 同じタブで、[API キー] を見つけます。
5. キーをコピーして安全な場所に保存します。

6. 新しいキーが必要な場合は、[API キーをリセット] を選択します。

7. Iris Investigate エンドポイント（https://api.domaintools.com/v1/iris-investigate/）をメモします。

Google SecOps 用に AWS S3 バケットと IAM を構成する

1. バケットの作成のユーザーガイドに沿って、Amazon S3 バケットを作成します。
2. 後で参照できるように、バケットの名前とリージョンを保存します（例: domaintools-iris）。
3. IAM ユーザーの作成のユーザーガイドに沿って、ユーザーを作成します。
4. 作成したユーザーを選択します。
5. [セキュリティ認証情報] タブを選択します。
6. [アクセスキー] セクションで [アクセスキーを作成] をクリックします。
7. [ユースケース] として [サードパーティ サービス] を選択します。
8. [次へ] をクリックします。
9. 省略可: 説明タグを追加します。
10. [アクセスキーを作成] をクリックします。
11. [CSV ファイルをダウンロード] をクリックして、[アクセスキー] と [シークレット アクセスキー] を保存し、後で使用できるようにします。
12. [完了] をクリックします。
13. [権限] タブを選択します。
14. [権限ポリシー] セクションで、[権限を追加] をクリックします。
15. [権限を追加] を選択します。
16. [ポリシーを直接アタッチする] を選択します。
17. AmazonS3FullAccess ポリシーを検索して選択します。
18. [次へ] をクリックします。
19. [権限を追加] をクリックします。

S3 アップロードの IAM ポリシーとロールを構成する

1. AWS コンソールで、[IAM] > [ポリシー] > [ポリシーの作成] > [JSON] タブに移動します。

2. 次のポリシーを入力します。

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Sid": "AllowPutDomainToolsIrisObjects",
         "Effect": "Allow",
         "Action": "s3:PutObject",
         "Resource": "arn:aws:s3:::domaintools-iris/*"
       }
     ]
   }
   

   • 別のバケット名を入力した場合は、domaintools-iris を置き換えます。

3. [次へ] > [ポリシーを作成] をクリックします。

4. [IAM] > [ロール] > [ロールの作成] > [AWS サービス] > [Lambda] に移動します。

5. 新しく作成したポリシーを関連付けます。

6. ロールに「WriteDomainToolsIrisToS3Role」という名前を付けて、[ロールを作成] をクリックします。

Lambda 関数を作成する

1. AWS コンソールで、[Lambda] > [Functions] > [Create function] に移動します。
2. [Author from scratch] をクリックします。

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

   設定	値
   名前	domaintools_iris_to_s3
   ランタイム	Python 3.13
   アーキテクチャ	x86_64
   実行ロール	WriteDomainToolsIrisToS3Role

4. 関数を作成したら、[コード] タブを開き、スタブを削除して次のコード（domaintools_iris_to_s3.py）を入力します。

   #!/usr/bin/env python3
   # Lambda: Pull DomainTools Iris Investigate results to S3 (no transform)
   
   import os, json, time, urllib.parse
   from urllib.request import Request, urlopen
   from urllib.error import HTTPError
   import boto3
   
   # --- Environment ---
   S3_BUCKET    = os.environ["S3_BUCKET"].strip()
   S3_PREFIX    = os.environ.get("S3_PREFIX", "domaintools/iris/").strip()
   STATE_KEY    = os.environ.get("STATE_KEY", "domaintools/iris/state.json").strip()
   
   DT_API_KEY   = os.environ["DT_API_KEY"].strip()
   DT_API_SECRET= os.environ.get("DT_API_SECRET", "").strip()  # optional if your account uses key-only auth
   
   USE_MODE     = os.environ.get("USE_MODE", "HASH").strip().upper()  # HASH | DOMAINS | QUERY
   SEARCH_HASHES= [h.strip() for h in os.environ.get("SEARCH_HASHES", "").split(";") if h.strip()]
   DOMAINS      = [d.strip() for d in os.environ.get("DOMAINS", "").split(";") if d.strip()]
   QUERY_LIST   = [q.strip() for q in os.environ.get("QUERY_LIST", "").split(";") if q.strip()]
   
   PAGE_SIZE    = int(os.environ.get("PAGE_SIZE", "500"))
   MAX_PAGES    = int(os.environ.get("MAX_PAGES", "20"))
   USE_NEXT     = os.environ.get("USE_NEXT", "true").lower() == "true"
   HTTP_TIMEOUT = int(os.environ.get("HTTP_TIMEOUT", "60"))
   RETRIES      = int(os.environ.get("HTTP_RETRIES", "2"))
   
   BASE_URL = "https://api.domaintools.com/v1/iris-investigate/"
   HDRS = {
       "X-Api-Key": DT_API_KEY,
       "Accept": "application/json",
   }
   if DT_API_SECRET:
       HDRS["X-Api-Secret"] = DT_API_SECRET
   
   s3 = boto3.client("s3")
   
   # --- HTTP helpers ---
   
   def _http_get(url: str) -> dict:
       req = Request(url, method="GET")
       for k, v in HDRS.items():
           req.add_header(k, v)
       attempt = 0
       while True:
           try:
               with urlopen(req, timeout=HTTP_TIMEOUT) as r:
                   return json.loads(r.read().decode("utf-8"))
           except HTTPError as e:
               if e.code in (429, 500, 502, 503, 504) and attempt < RETRIES:
                   delay = int(e.headers.get("Retry-After", "2"))
                   time.sleep(max(1, delay))
                   attempt += 1
                   continue
               raise
   
   def _build_url(params: dict) -> str:
       return BASE_URL + ("?" + urllib.parse.urlencode(params, doseq=True) if params else "")
   
   # --- S3 helpers ---
   
   def _write_page(obj: dict, label: str, page: int) -> str:
       ts = time.strftime("%Y/%m/%d/%H%M%S", time.gmtime())
       key = f"{S3_PREFIX.rstrip('/')}/{ts}-{label}-p{page:05d}.json"
       s3.put_object(
           Bucket=S3_BUCKET, Key=key,
           Body=json.dumps(obj, separators=(",", ":")).encode("utf-8"),
           ContentType="application/json",
       )
       return key
   
   # --- Iris paging ---
   
   def _first_page_params() -> dict:
       params: dict[str, object] = {"page_size": str(PAGE_SIZE)}
       if USE_NEXT:
           params["next"] = "true"
       return params
   
   def _paginate(label: str, params: dict) -> tuple[int, int]:
       pages = 0
       total = 0
       url = _build_url(params)
       while pages < MAX_PAGES:
           data = _http_get(url)
           _write_page(data, label, pages)
   
           resp = data.get("response") or {}
           results = resp.get("results") or []
           total += len(results)
           pages += 1
   
           # Prefer `next` absolute URL if present
           next_url = resp.get("next") if isinstance(resp, dict) else None
           if next_url:
               url = next_url
               continue
   
           # Fallback: position pager when `next=true` not used/supported
           if resp.get("has_more_results") and resp.get("position"):
               base = _first_page_params()
               base.pop("next", None)
               base["position"] = resp["position"]
               url = _build_url(base)
               continue
   
           break
       return pages, total
   
   # --- Mode runners ---
   
   def run_hashes(hashes: list[str]) -> dict:
       agg_pages = agg_results = 0
       for h in hashes:
           params = _first_page_params()
           params["search_hash"] = h
           p, r = _paginate(f"hash-{h}", params)
           agg_pages += p
           agg_results += r
       return {"pages": agg_pages, "results": agg_results}
   
   def run_domains(domains: list[str]) -> dict:
       agg_pages = agg_results = 0
       for d in domains:
           params = _first_page_params()
           # DomainTools accepts `domain` as a filter in Investigate search
           params["domain"] = d
           p, r = _paginate(f"domain-{d}", params)
           agg_pages += p
           agg_results += r
       return {"pages": agg_pages, "results": agg_results}
   
   def run_queries(queries: list[str]) -> dict:
       agg_pages = agg_results = 0
       for q in queries:
           # Merge arbitrary k=v pairs from the query string
           base = _first_page_params()
           for k, v in urllib.parse.parse_qsl(q, keep_blank_values=True):
               base.setdefault(k, v)
           p, r = _paginate(f"query-{q.replace('=','-')}", base)
           agg_pages += p
           agg_results += r
       return {"pages": agg_pages, "results": agg_results}
   
   # --- Entry point ---
   
   def lambda_handler(event=None, context=None):
       if USE_MODE == "HASH" and SEARCH_HASHES:
           res = run_hashes(SEARCH_HASHES)
       elif USE_MODE == "DOMAINS" and DOMAINS:
           res = run_domains(DOMAINS)
       elif USE_MODE == "QUERY" and QUERY_LIST:
           res = run_queries(QUERY_LIST)
       else:
           raise ValueError("Invalid USE_MODE or missing parameters. Set USE_MODE to HASH | DOMAINS | QUERY and provide SEARCH_HASHES | DOMAINS | QUERY_LIST accordingly.")
   
       return {"ok": True, "mode": USE_MODE, **res}
   
   if __name__ == "__main__":
       print(json.dumps(lambda_handler(), indent=2))
   

5. [構成> 環境変数 > 編集 > 新しい環境変数を追加] に移動します。

6. 次の環境変数を入力し、実際の値に置き換えます。

   キー	値の例	説明
   S3_BUCKET	domaintools-iris	データが保存される S3 バケットの名前。
   S3_PREFIX	domaintools/iris/	オブジェクトの省略可能な S3 接頭辞（サブフォルダ）。
   STATE_KEY	domaintools/iris/state.json	状態/チェックポイント ファイルのキー（省略可）。
   DT_API_KEY	DT-XXXXXXXXXXXXXXXXXXXX	DomainTools API キー。
   DT_API_SECRET	YYYYYYYYYYYYYYYYYYYYYYYY	DomainTools API シークレット（該当する場合）。
   USE_MODE	HASH | DOMAINS | QUERY	使用するモードを選択します（一度に有効にできるモードは 1 つのみです）。
   SEARCH_HASHES	hash1;hash2;hash3	USE_MODE=HASH の場合は必須。Iris UI からの保存済み検索ハッシュのセミコロン区切りリスト。
   DOMAINS	example.com;domaintools.com	USE_MODE=DOMAINS の場合は必須。セミコロンで区切られたドメインのリスト。
   QUERY_LIST	ip=1.1.1.1;ip=8.8.8.8;domain=example.org	USE_MODE=QUERY の場合は必須。セミコロンで区切られたクエリ文字列のリスト（k=v&k2=v2）。
   PAGE_SIZE	500	1 ページあたりの行数（デフォルトは 500）。
   MAX_PAGES	20	リクエストあたりの最大ページ数

7. 関数が作成されたら、そのページにとどまるか、[Lambda] > [関数] > [your-function] を開きます。

8. [CONFIGURATION] タブを選択します。

9. [全般的な構成] パネルで、[編集] をクリックします。

10. [Timeout] を [15 minutes (900 seconds)] に変更し、[Save] をクリックします。

EventBridge スケジュールを作成する

1. Amazon EventBridge > Scheduler > スケジュールの作成に移動します。
2. 次の構成の詳細を入力します。
   • 定期的なスケジュール: レート（1 hour）。
   • ターゲット: Lambda 関数。
   • 名前: domaintools-iris-1h
3. [スケジュールを作成] をクリックします。

省略可: Google SecOps 用の読み取り専用の IAM ユーザーと鍵を作成する

1. AWS コンソールで、[IAM] > [Users] に移動し、[Add users] をクリックします。
2. 次の構成の詳細を入力します。
   • ユーザー: 一意の名前を入力します（例: secops-reader）。
   • アクセスタイプ: [Access key - Programmatic access] を選択します。
   • [ユーザーを作成] をクリックします。
3. 最小限の読み取りポリシー（カスタム）を適用する: [ユーザー] > secops-reader を選択 > [権限] > [権限を追加] > [ポリシーを直接適用] > [ポリシーを作成]

4. JSON エディタで次のポリシーを入力します。

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": ["s3:GetObject"],
         "Resource": "arn:aws:s3:::<your-bucket>/*"
       },
       {
         "Effect": "Allow",
         "Action": ["s3:ListBucket"],
         "Resource": "arn:aws:s3:::<your-bucket>"
       }
     ]
   }
   

5. 名前を secops-reader-policy に設定します。

6. [ポリシーの作成> 検索/選択> 次へ> 権限を追加] に移動します。

7. [セキュリティ認証情報] > [アクセスキー] > [アクセスキーを作成] に移動します。

8. CSV をダウンロードします（これらの値はフィードに入力されます）。

DomainTools Iris Investigate の結果を取り込むように Google SecOps でフィードを構成する

1. [SIEM 設定] > [フィード] に移動します。
2. [Add New Feed] をクリックします。
3. [フィード名] フィールドに、フィードの名前を入力します（例: DomainTools Iris Investigate）。
4. [ソースタイプ] として [Amazon S3 V2] を選択します。
5. [ログタイプ] として [DomainTools Threat Intelligence] を選択します。
6. [次へ] をクリックします。
7. 次の入力パラメータの値を指定します。
   • S3 URI: s3://domaintools-iris/domaintools/iris/
   • Source deletion options: 必要に応じて削除オプションを選択します。
   • 最大ファイル経過時間: デフォルトは 180 日です。
   • アクセスキー ID: S3 バケットにアクセスできるユーザー アクセスキー。
   • シークレット アクセスキー: S3 バケットにアクセスできるユーザーのシークレット キー。
   • アセットの Namespace: domaintools.threat_intel
   • Ingestion labels: このフィードのイベントに適用されるラベル。
8. [次へ] をクリックします。
9. [Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。

UDM マッピング テーブル

さらにサポートが必要な場合 コミュニティ メンバーや Google SecOps のプロフェッショナルから回答を得ることができます。