SailPoint IAM ログを収集する

このドキュメントでは、Amazon S3 を使用して SailPoint Identity and Access Management（IAM）ログを Google Security Operations に取り込む方法について説明します。パーサーは JSON 形式と XML 形式のログを処理し、Unified Data Model（UDM）に変換します。単一の UDM イベント（ProvisioningPlan、AccountRequest、SOAP-ENV）、複数の UDM イベント（ProvisioningProject）、UDM エンティティ（Identity）を区別し、それぞれに特定の解析ロジックとフィールド マッピングを適用します。これには、非 XML データの汎用イベント処理も含まれます。

始める前に

次の前提条件を満たしていることを確認してください。

• Google SecOps インスタンス。
• SailPoint Identity Security Cloud への特権アクセス。
• AWS（S3、IAM、Lambda、EventBridge）への特権アクセス。

SailPoint IAM の前提条件（ID、API キー、組織 ID、トークン）を収集する

1. 管理者として SailPoint Identity Security Cloud 管理コンソールにログインします。
2. [グローバル> セキュリティ設定> API 管理] に移動します。
3. [Create API Client] をクリックします。
4. 権限付与タイプとして [クライアント認証情報] を選択します。
5. 次の構成の詳細を入力します。
   • 名前: わかりやすい名前を入力します（例: Google SecOps Export API）。
   • 説明: API クライアントの説明を入力します。
   • スコープ: sp:scopes:all を選択します。
6. [作成] をクリックし、生成された API 認証情報を安全な場所に保存します。
7. SailPoint テナントのベース URL（https://tenant.api.identitynow.com など）を記録します。
8. 次の詳細をコピーして安全な場所に保存します。
   • IDN_CLIENT_ID。
   • IDN_CLIENT_SECRET。
   • IDN_BASE。

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

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

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

1. AWS コンソールで、[IAM] > [ポリシー] に移動します。
2. [ポリシーを作成> [JSON] タブ] をクリックします。
3. 次のポリシーをコピーして貼り付けます。

4. ポリシー JSON（別のバケット名を入力した場合は sailpoint-iam-logs を置き換えます）:

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Sid": "AllowPutObjects",
         "Effect": "Allow",
         "Action": "s3:PutObject",
         "Resource": "arn:aws:s3:::sailpoint-iam-logs/*"
       },
       {
         "Sid": "AllowGetStateObject",
         "Effect": "Allow",
         "Action": "s3:GetObject",
         "Resource": "arn:aws:s3:::sailpoint-iam-logs/sailpoint/iam/state.json"
       }
     ]
   }
   

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

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

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

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

Lambda 関数を作成する

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

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

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

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

   #!/usr/bin/env python3
   # Lambda: Pull SailPoint Identity Security Cloud audit events and store raw JSON payloads to S3
   # - Uses /v3/search API with pagination for audit events.
   # - Preserves vendor-native JSON format for identity events.
   # - Retries with exponential backoff; unique S3 keys to avoid overwrites.
   
   import os, json, time, uuid, urllib.parse
   from urllib.request import Request, urlopen
   from urllib.error import URLError, HTTPError
   
   import boto3
   
   S3_BUCKET   = os.environ["S3_BUCKET"]
   S3_PREFIX   = os.environ.get("S3_PREFIX", "sailpoint/iam/")
   STATE_KEY   = os.environ.get("STATE_KEY", "sailpoint/iam/state.json")
   WINDOW_SEC  = int(os.environ.get("WINDOW_SECONDS", "3600"))  # default 1h
   HTTP_TIMEOUT= int(os.environ.get("HTTP_TIMEOUT", "60"))
   IDN_BASE    = os.environ["IDN_BASE"]  # e.g. https://tenant.api.identitynow.com
   CLIENT_ID   = os.environ["IDN_CLIENT_ID"]
   CLIENT_SECRET = os.environ["IDN_CLIENT_SECRET"]
   SCOPE       = os.environ.get("IDN_SCOPE", "sp:scopes:all")
   PAGE_SIZE   = int(os.environ.get("PAGE_SIZE", "250"))
   MAX_PAGES   = int(os.environ.get("MAX_PAGES", "20"))
   MAX_RETRIES = int(os.environ.get("MAX_RETRIES", "3"))
   USER_AGENT  = os.environ.get("USER_AGENT", "sailpoint-iam-to-s3/1.0")
   
   s3 = boto3.client("s3")
   
   def _load_state():
       try:
           obj = s3.get_object(Bucket=S3_BUCKET, Key=STATE_KEY)
           return json.loads(obj["Body"].read())
       except Exception:
           return {}
   
   def _save_state(st):
       s3.put_object(
           Bucket=S3_BUCKET,
           Key=STATE_KEY,
           Body=json.dumps(st, separators=(",", ":")).encode("utf-8"),
           ContentType="application/json",
       )
   
   def _iso(ts: float) -> str:
       return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(ts))
   
   def _get_oauth_token() -> str:
       """Get OAuth2 access token using Client Credentials flow"""
       token_url = f"{IDN_BASE.rstrip('/')}/oauth/token"
   
       data = urllib.parse.urlencode({
           'grant_type': 'client_credentials',
           'client_id': CLIENT_ID,
           'client_secret': CLIENT_SECRET,
           'scope': SCOPE
       }).encode('utf-8')
   
       req = Request(token_url, data=data, method="POST")
       req.add_header("Content-Type", "application/x-www-form-urlencoded")
       req.add_header("User-Agent", USER_AGENT)
   
       with urlopen(req, timeout=HTTP_TIMEOUT) as r:
           response = json.loads(r.read())
           return response["access_token"]
   
   def _search_events(access_token: str, created_from: str, search_after: list = None) -> list:
       """Search for audit events using SailPoint's /v3/search API"""
       search_url = f"{IDN_BASE.rstrip('/')}/v3/search"
   
       # Build search query for events created after specified time
       query_str = f'created:">={created_from}"'
   
       payload = {
           "indices": ["events"],
           "query": {"query": query_str},
           "sort": ["created", "+id"],
           "limit": PAGE_SIZE
       }
   
       if search_after:
           payload["searchAfter"] = search_after
   
       attempt = 0
       while True:
           req = Request(search_url, data=json.dumps(payload).encode('utf-8'), method="POST")
           req.add_header("Content-Type", "application/json")
           req.add_header("Accept", "application/json")
           req.add_header("Authorization", f"Bearer {access_token}")
           req.add_header("User-Agent", USER_AGENT)
   
           try:
               with urlopen(req, timeout=HTTP_TIMEOUT) as r:
                   response = json.loads(r.read())
                   # Handle different response formats
                   if isinstance(response, list):
                       return response
                   return response.get("results", response.get("data", []))
           except (HTTPError, URLError) as e:
               attempt += 1
               print(f"HTTP error on attempt {attempt}: {e}")
               if attempt > MAX_RETRIES:
                   raise
               # exponential backoff with jitter
               time.sleep(min(60, 2 ** attempt) + (time.time() % 1))
   
   def _put_events_data(events: list, from_ts: float, to_ts: float, page_num: int) -> str:
       # Create unique S3 key for events data
       ts_path = time.strftime("%Y/%m/%d", time.gmtime(to_ts))
       uniq = f"{int(time.time()*1e6)}_{uuid.uuid4().hex[:8]}"
       key = f"{S3_PREFIX}{ts_path}/sailpoint_iam_{int(from_ts)}_{int(to_ts)}_p{page_num:03d}_{uniq}.json"
   
       s3.put_object(
           Bucket=S3_BUCKET, 
           Key=key, 
           Body=json.dumps(events, separators=(",", ":")).encode("utf-8"), 
           ContentType="application/json",
           Metadata={
               'source': 'sailpoint-iam',
               'from_timestamp': str(int(from_ts)),
               'to_timestamp': str(int(to_ts)),
               'page_number': str(page_num),
               'events_count': str(len(events))
           }
       )
       return key
   
   def _get_item_id(item: dict) -> str:
       """Extract ID from event item, trying multiple possible fields"""
       for field in ("id", "uuid", "eventId", "_id"):
           if field in item and item[field]:
               return str(item[field])
       return ""
   
   def lambda_handler(event=None, context=None):
       st = _load_state()
       now = time.time()
       from_ts = float(st.get("last_to_ts") or (now - WINDOW_SEC))
       to_ts = now
   
       # Get OAuth token
       access_token = _get_oauth_token()
   
       created_from = _iso(from_ts)
       print(f"Fetching SailPoint IAM events from: {created_from}")
   
       # Handle pagination state
       last_created = st.get("last_created")
       last_id = st.get("last_id")
       search_after = [last_created, last_id] if (last_created and last_id) else None
   
       pages = 0
       total_events = 0
       written_keys = []
       newest_created = last_created or created_from
       newest_id = last_id or ""
   
       while pages < MAX_PAGES:
           events = _search_events(access_token, created_from, search_after)
   
           if not events:
               break
   
           # Write page to S3
           key = _put_events_data(events, from_ts, to_ts, pages + 1)
           written_keys.append(key)
           total_events += len(events)
   
           # Update pagination state from last item
           last_event = events[-1]
           last_event_created = last_event.get("created") or last_event.get("metadata", {}).get("created")
           last_event_id = _get_item_id(last_event)
   
           if last_event_created:
               newest_created = last_event_created
           if last_event_id:
               newest_id = last_event_id
   
           search_after = [newest_created, newest_id]
           pages += 1
   
           # If we got less than page size, we're done
           if len(events) < PAGE_SIZE:
               break
   
       print(f"Successfully retrieved {total_events} events across {pages} pages")
   
       # Save state for next run
       st["last_to_ts"] = to_ts
       st["last_created"] = newest_created
       st["last_id"] = newest_id
       st["last_successful_run"] = now
       _save_state(st)
   
       return {
           "statusCode": 200,
           "body": {
               "success": True,
               "pages": pages,
               "total_events": total_events,
               "s3_keys": written_keys,
               "from_timestamp": from_ts,
               "to_timestamp": to_ts,
               "last_created": newest_created,
               "last_id": newest_id
           }
       }
   
   if __name__ == "__main__":
       print(lambda_handler())
   

5. [構成] > [環境変数] に移動します。

6. [編集>新しい環境変数を追加] をクリックします。

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

   環境変数

   キー	値の例
   S3_BUCKET	sailpoint-iam-logs
   S3_PREFIX	sailpoint/iam/
   STATE_KEY	sailpoint/iam/state.json
   WINDOW_SECONDS	3600
   HTTP_TIMEOUT	60
   MAX_RETRIES	3
   USER_AGENT	sailpoint-iam-to-s3/1.0
   IDN_BASE	https://tenant.api.identitynow.com
   IDN_CLIENT_ID	your-client-id（ステップ 2 から）
   IDN_CLIENT_SECRET	your-client-secret（ステップ 2 から）
   IDN_SCOPE	sp:scopes:all
   PAGE_SIZE	250
   MAX_PAGES	20

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

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

10. [全般設定] パネルで、[編集] をクリックします。

11. [Timeout] を [5 minutes (300 seconds)] に変更し、[Save] をクリックします。

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

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

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

1. AWS コンソール > IAM > ユーザーに移動します。
2. [ユーザーを追加] をクリックします。
3. 次の構成の詳細を入力します。
   • ユーザー: 「secops-reader」と入力します。
   • アクセスの種類: [アクセスキー - プログラムによるアクセス] を選択します。
4. [ユーザーを作成] をクリックします。
5. 最小限の読み取りポリシー（カスタム）を関連付ける: [ユーザー] > [secops-reader] > [権限] > [権限を追加] > [ポリシーを直接関連付ける] > [ポリシーを作成]。

6. JSON:

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": ["s3:GetObject"],
         "Resource": "arn:aws:s3:::sailpoint-iam-logs/*"
       },
       {
         "Effect": "Allow",
         "Action": ["s3:ListBucket"],
         "Resource": "arn:aws:s3:::sailpoint-iam-logs"
       }
     ]
   }
   

7. 名前 = secops-reader-policy。

8. [ポリシーを作成> 検索/選択> 次へ> 権限を追加] をクリックします。

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

10. [アクセスキーを作成] をクリックします。

11. .CSV をダウンロードします。（これらの値はフィードに貼り付けます）。

SailPoint IAM のログを取り込むように Google SecOps でフィードを構成する

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

UDM マッピング テーブル

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