Atlassian Organization の監査ログを収集する

パーサーのバージョン: 1.0

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

このドキュメントでは、Google Cloud Storage V2 を使用して Atlassian Organization Audit Log の監査ログを Google Security Operations に取り込む方法について説明します。

Atlassian 組織監査ログは、admin.atlassian.com で Atlassian 組織全体の管理アクティビティを一元的に記録します。ユーザー プロビジョニング、グループ メンバーシップの変更、認証ポリシーの更新、API キー管理などのイベントをキャプチャし、セキュリティ チームが組織管理者による変更をモニタリングして調査できるようにします。

始める前に

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

  • Google SecOps インスタンス
  • Cloud Storage API が有効になっている GCP プロジェクト
  • GCS バケットを作成、管理する権限
  • GCS バケットの IAM ポリシーを管理する権限
  • Cloud Run サービス、Pub/Sub トピック、Cloud Scheduler ジョブを作成する権限
  • admin.atlassian.com での Atlassian 組織管理への特権アクセス
  • Atlassian Guard サブスクリプション(Standard または Premium)または 1 つ以上の Enterprise プラン(組織の監査ログに必要)を持つ Atlassian 組織

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

  1. Google Cloud コンソールに移動します。
  2. プロジェクトを選択するか、新しいプロジェクトを作成します。
  3. ナビゲーション メニューで、[Cloud Storage] > [バケット] に移動します。
  4. [バケットを作成] をクリックします。
  5. 次の構成情報を提供してください。

    設定 値
    バケットに名前を付ける グローバルに一意の名前(atlassian-audit-logs など)を入力します。
    ロケーション タイプ ニーズに応じて選択します(リージョン、デュアルリージョン、マルチリージョン)。
    ロケーション ロケーションを選択します(例: us-central1)。
    ストレージ クラス Standard(頻繁にアクセスされるログに推奨)
    アクセス制御 均一(推奨)
    保護ツール 省略可: オブジェクトのバージョニングまたは保持ポリシーを有効にする
  6. [作成] をクリックします。

Atlassian API 認証情報を収集する

API キーを作成

  1. admin.atlassian.com にログインします。
  2. 複数の組織がある場合は、組織を選択します。
  3. [設定] > [API キー] に移動します。
  4. [API キーを作成] をクリックします。
  5. 鍵のわかりやすい名前を入力します(例: SecOps-Audit-Collector)。
  6. [有効期限] の日付を設定します。

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

  8. API キーの値をコピーして安全に保存します。このダイアログを閉じると、API キーの値を復元できなくなります。

  9. ブラウザのアドレスバーの URL(https://admin.atlassian.com/o/{orgId}/... の UUID)から組織 ID をメモします。

  10. [完了] をクリックします。

権限を確認する

API を介して組織の監査ログにアクセスするには、組織管理者である必要があります。

  1. admin.atlassian.com にログインします。
  2. [設定] > [API キー] に移動します。
  3. [API キー] ページが表示され、キーを作成できる場合は、必要な組織管理者権限があります。
  4. このオプションが表示されない場合は、Atlassian 組織管理者にお問い合わせのうえ、組織管理者権限を付与してもらってください。

テスト API へのアクセス

  • 統合に進む前に、認証情報をテストします。

    # Replace with your actual credentials
    API_KEY="<your-atlassian-api-key>"
    ORG_ID="<your-organization-id>"
    
    # Test API access using the events-stream endpoint
    curl -v \
        --header "Authorization: Bearer ${API_KEY}" \
        --header "Accept: application/json" \
        "https://api.atlassian.com/admin/v1/orgs/${ORG_ID}/events-stream?limit=1"
    

成功したレスポンスは、イベント オブジェクトの data 配列を含む JSON 本文とともに HTTP 200 を返します。

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

Cloud Run functions の関数には、GCS バケットへの書き込み権限と Pub/Sub による呼び出し権限を持つサービス アカウントが必要です。

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

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

これらのロールは、次の操作に必要です。

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

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

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

  1. [Cloud Storage] > [バケット] に移動します。
  2. バケット名(atlassian-audit-logs など)をクリックします。
  3. [権限] タブに移動します。
  4. [アクセス権を付与] をクリックします。
  5. 次の構成情報を指定します。
    • プリンシパルを追加: サービス アカウントのメールアドレス(例: atlassian-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com)を入力します。
    • ロールを割り当てる: [Storage オブジェクト管理者] を選択します。
  6. [保存] をクリックします。

Pub/Sub トピックの作成

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

  1. GCP コンソールで、[Pub/Sub] > [トピック] に移動します。
  2. [トピックを作成] をクリックします。
  3. 次の構成情報を指定します。
    • トピック ID: 「atlassian-audit-logs-trigger」と入力します。
    • その他の設定はデフォルトのままにします。
  4. [作成] をクリックします。

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

Cloud Run functions の関数は、Cloud Scheduler からの Pub/Sub メッセージによってトリガーされ、Atlassian Organizations REST API から監査ログイベントをポーリングして GCS に書き込みます。

  1. GCP Console で Cloud Run に移動します。
  2. [サービスを作成] をクリックします。
  3. [関数](インライン編集を使用して関数を作成する)を選択します。
  4. [構成] セクションで、次の構成の詳細を指定します。

    設定 値
    サービス名 atlassian-audit-collector
    リージョン GCS バケットに一致するリージョンを選択します(例: us-central1)。
    ランタイム [Python 3.12] 以降を選択します。
  5. [トリガー(省略可)] セクションで、次の操作を行います。

    1. [+ トリガーを追加] をクリックします。
    2. [Cloud Pub/Sub] を選択します。
    3. [Cloud Pub/Sub トピックを選択してください] で、Pub/Sub トピック(atlassian-audit-logs-trigger)を選択します。
    4. [保存] をクリックします。
  6. [認証] セクションで、次の操作を行います。

    1. [認証が必要] を選択します。
    2. Identity and Access Management(IAM)を確認します。
  7. 下にスクロールして、[コンテナ、ネットワーキング、セキュリティ] を開きます。

  8. [セキュリティ] タブに移動します。

    • サービス アカウント: サービス アカウント(atlassian-audit-collector-sa)を選択します。
  9. [コンテナ] タブに移動します。

    1. [変数とシークレット] をクリックします。
    2. 環境変数ごとに [+ 変数を追加] をクリックします。
    変数名 値の例 説明
    GCS_BUCKET atlassian-audit-logs GCS バケット名
    GCS_PREFIX atlassian-audit ログファイルの接頭辞
    STATE_KEY atlassian-audit/state.json 状態ファイルのパス
    ATL_TOKEN your-atlassian-api-key Atlassian API キー
    ATL_ORG_ID your-organization-id Atlassian 組織 ID
    MAX_RECORDS 1000 1 回の実行あたりの最大レコード数
    PAGE_SIZE 200 1 ページあたりのレコード数(最大 500)
    LOOKBACK_HOURS 24 最初のルックバック期間
  10. [変数とシークレット] セクションで、[リクエスト] まで下にスクロールします。

    • リクエストタイムアウト: 600秒(10 分)を入力してください
  11. [設定] タブに移動します。

    • [リソース] セクションで、次の操作を行います。
      • メモリ: 512 MiB 以上を選択します。
      • CPU: 1 を選択します。
  12. [リビジョン スケーリング] セクションで、次の操作を行います。

    • [インスタンスの最小数]: 「0」と入力します。
    • [インスタンスの最大数]: 「100」と入力します(または、予想される負荷に基づいて調整します)。
  13. [作成] をクリックします。

  14. サービスが作成されるまで待ちます(1 ~ 2 分)。

  15. サービスが作成されると、インライン コードエディタが自動的に開きます。

関数コードを追加する

  1. [エントリ ポイント] フィールドに「main」と入力します。
  2. インライン コードエディタで、次の 2 つのファイルを作成します。

    • main.py:
    import functions_framework
    from google.cloud import storage
    import json
    import os
    import urllib3
    from datetime import datetime, timezone, timedelta
    from urllib.parse import urlparse, parse_qs
    import time
    
    # Initialize HTTP client with timeouts
    http = urllib3.PoolManager(
      timeout=urllib3.Timeout(connect=5.0, read=30.0),
      retries=False,
    )
    
    # Initialize Storage client
    storage_client = storage.Client()
    
    # Environment variables
    GCS_BUCKET = os.environ.get('GCS_BUCKET')
    GCS_PREFIX = os.environ.get('GCS_PREFIX', 'atlassian-audit')
    STATE_KEY = os.environ.get('STATE_KEY', 'atlassian-audit/state.json')
    ATL_TOKEN = os.environ.get('ATL_TOKEN')
    ATL_ORG_ID = os.environ.get('ATL_ORG_ID')
    MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '1000'))
    PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '200'))
    LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
    
    def parse_datetime(value):
      """Parse ISO datetime string to datetime object."""
      if value.endswith("Z"):
        value = value[:-1] + "+00:00"
      return datetime.fromisoformat(value)
    
    @functions_framework.cloud_event
    def main(cloud_event):
      """
      Cloud Run function triggered by Pub/Sub to fetch Atlassian Organization
      Audit logs via the events-stream endpoint and write to GCS.
    
      Args:
        cloud_event: CloudEvent object containing Pub/Sub message
      """
    
      if not all([GCS_BUCKET, ATL_TOKEN, ATL_ORG_ID]):
        print('Error: Missing required environment variables')
        return
    
      try:
        # Get GCS bucket
        bucket = storage_client.bucket(GCS_BUCKET)
    
        # Load state
        state = load_state(bucket, STATE_KEY)
    
        # Determine time window
        now = datetime.now(timezone.utc)
        last_time = None
    
        if isinstance(state, dict) and state.get("last_event_time"):
          try:
            last_time = parse_datetime(state["last_event_time"])
            # Overlap by 2 minutes to catch any delayed events
            last_time = last_time - timedelta(minutes=2)
          except Exception as e:
            print(f"Warning: Could not parse last_event_time: {e}")
    
        if last_time is None:
          last_time = now - timedelta(hours=LOOKBACK_HOURS)
    
        print(f"Fetching logs from {last_time.isoformat()} to {now.isoformat()}")
    
        # Fetch logs using the events-stream polling endpoint
        records, newest_event_time = fetch_logs(
          start_time=last_time,
          end_time=now,
          page_size=PAGE_SIZE,
          max_records=MAX_RECORDS,
        )
    
        if not records:
          print("No new log records found.")
          save_state(bucket, STATE_KEY, now.isoformat())
          return
    
        # Write to GCS as NDJSON
        timestamp_str = now.strftime('%Y%m%d_%H%M%S')
        object_key = f"{GCS_PREFIX}/logs_{timestamp_str}.ndjson"
        blob = bucket.blob(object_key)
    
        ndjson = '\n'.join([json.dumps(record, ensure_ascii=False) for record in records]) + '\n'
        blob.upload_from_string(ndjson, content_type='application/x-ndjson')
    
        print(f"Wrote {len(records)} records to gs://{GCS_BUCKET}/{object_key}")
    
        # Update state with newest event time
        if newest_event_time:
          save_state(bucket, STATE_KEY, newest_event_time)
        else:
          save_state(bucket, STATE_KEY, now.isoformat())
    
        print(f"Successfully processed {len(records)} records")
    
      except Exception as e:
        print(f'Error processing logs: {str(e)}')
        raise
    
    def load_state(bucket, key):
      """Load state from GCS."""
      try:
        blob = bucket.blob(key)
        if blob.exists():
          state_data = blob.download_as_text()
          return json.loads(state_data)
      except Exception as e:
        print(f"Warning: Could not load state: {e}")
    
      return {}
    
    def save_state(bucket, key, last_event_time_iso):
      """Save the last event timestamp to GCS state file."""
      try:
        state = {'last_event_time': last_event_time_iso}
        blob = bucket.blob(key)
        blob.upload_from_string(
          json.dumps(state, indent=2),
          content_type='application/json'
        )
        print(f"Saved state: last_event_time={last_event_time_iso}")
      except Exception as e:
        print(f"Warning: Could not save state: {e}")
    
    def fetch_logs(start_time, end_time, page_size, max_records):
      """
      Fetch audit logs from Atlassian Organization events-stream API
      with cursor-based pagination.
    
      The events-stream endpoint is designed for polling use cases and
      supports higher throughput than the /events query endpoint.
    
      API reference:
      https://developer.atlassian.com/cloud/admin/organization/rest/api-group-events/
    
      Args:
        start_time: Start time for log query
        end_time: End time for log query
        page_size: Number of records per page (max 500)
        max_records: Maximum total records to fetch
    
      Returns:
        Tuple of (records list, newest_event_time ISO string)
      """
      endpoint = f"https://api.atlassian.com/admin/v1/orgs/{ATL_ORG_ID}/events-stream"
    
      headers = {
        'Authorization': f'Bearer {ATL_TOKEN}',
        'Accept': 'application/json',
        'User-Agent': 'GoogleSecOps-AtlassianAuditCollector/1.0'
      }
    
      records = []
      newest_time = None
      page_num = 0
      backoff = 1.0
      cursor = None
    
      while True:
        page_num += 1
    
        if len(records) >= max_records:
          print(f"Reached max_records limit ({max_records})")
          break
    
        # Build request URL with query parameters
        params = [
          f"from={start_time.strftime('%Y-%m-%dT%H:%M:%S.000Z')}",
          f"to={end_time.strftime('%Y-%m-%dT%H:%M:%S.000Z')}",
          f"limit={min(page_size, max_records - len(records))}",
        ]
        if cursor:
          params.append(f"cursor={cursor}")
    
        url = f"{endpoint}?{'&'.join(params)}"
    
        try:
          response = http.request('GET', url, headers=headers)
    
          # Handle rate limiting with exponential backoff
          if response.status == 429:
            retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
            print(f"Rate limited (429). Retrying after {retry_after}s...")
            time.sleep(retry_after)
            backoff = min(backoff * 2, 30.0)
            continue
    
          backoff = 1.0
    
          if response.status != 200:
            print(f"HTTP Error: {response.status}")
            response_text = response.data.decode('utf-8')
            print(f"Response body: {response_text}")
            return [], None
    
          data = json.loads(response.data.decode('utf-8'))
    
          page_results = data.get('data', [])
    
          if not page_results:
            print(f"No more results (empty page)")
            break
    
          print(f"Page {page_num}: Retrieved {len(page_results)} events")
          records.extend(page_results)
    
          # Track newest event time from attributes
          for event in page_results:
            try:
              attrs = event.get('attributes', {})
              event_time = attrs.get('time')
              if event_time:
                if newest_time is None or parse_datetime(event_time) > parse_datetime(newest_time):
                  newest_time = event_time
            except Exception as e:
              print(f"Warning: Could not parse event time: {e}")
    
          # Check for next page using links.next or meta.next cursor
          next_link = data.get('links', {}).get('next')
          if not next_link:
            next_cursor = data.get('meta', {}).get('next')
            if not next_cursor:
              print("No more pages (no next cursor)")
              break
            cursor = next_cursor
          else:
            # Extract cursor from next link URL
            parsed = urlparse(next_link)
            query_params = parse_qs(parsed.query)
            cursor = query_params.get('cursor', [None])[0]
            if not cursor:
              break
    
        except Exception as e:
          print(f"Error fetching logs: {e}")
          return [], None
    
      print(f"Retrieved {len(records)} total records from {page_num} pages")
      return records, newest_time
    
    • requirements.txt:
    functions-framework==3.*
    google-cloud-storage==2.*
    urllib3>=2.0.0
    
  3. [デプロイ] をクリックして関数を保存し、デプロイします。

  4. デプロイが完了するまで待ちます(2 ~ 3 分)。

Cloud Scheduler ジョブの作成

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

  1. GCP Console で、[Cloud Scheduler] に移動します。
  2. [ジョブを作成] をクリックします。
  3. 次の構成情報を提供してください。

    設定 値
    名前 atlassian-audit-collector-hourly
    リージョン Cloud Run functions の関数と同じリージョンを選択します。
    周波数 0 * * * *(1 時間ごと、毎正時)
    タイムゾーン タイムゾーンを選択(UTC を推奨)
    ターゲット タイプ Pub/Sub
    トピック Pub/Sub トピック(atlassian-audit-logs-trigger)を選択します。
    メッセージ本文 {}(空の JSON オブジェクト)
  4. [作成] をクリックします。

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

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

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

統合をテストする

  1. Cloud Scheduler コンソールで、ジョブを見つけます。
  2. [強制実行] をクリックして、ジョブを手動でトリガーします。
  3. 数秒待ちます。
  4. [Cloud Run] > [サービス] に移動します。
  5. 関数名(atlassian-audit-collector)をクリックします。
  6. [ログ] タブをクリックします。
  7. 関数が正常に実行されたことを確認します。以下のものを探します。

    Fetching logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
    Page 1: Retrieved X events
    Wrote X records to gs://atlassian-audit-logs/atlassian-audit/logs_YYYYMMDD_HHMMSS.ndjson
    Successfully processed X records
    
  8. [Cloud Storage] > [バケット] に移動します。

  9. バケット名(atlassian-audit-logs)をクリックします。

  10. 接頭辞フォルダ(atlassian-audit/)に移動します。

  11. 現在のタイムスタンプで新しい .ndjson ファイルが作成されたことを確認します。

ログにエラーが表示される場合:

  • HTTP 401: 環境変数の API キーを確認します。
  • HTTP 403: 組織管理者権限と Atlassian Guard サブスクリプションを確認します。
  • HTTP 429: レート制限 - 関数はバックオフを使用して自動的に再試行します
  • 環境変数が設定されていない: 必要な変数がすべて設定されていることを確認します。

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

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

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

  1. [設定] > [データ管理] > [フィード] に移動します。SIEM のみのプラットフォームを使用している場合は、[設定] > [フィード] に移動します。
  2. [Add New Feed] をクリックします。
  3. [単一フィードを設定] をクリックします。
  4. [フィード名] フィールドに、フィードの名前を入力します(例: Atlassian Audit Logs)。
  5. [Source type] として [Google Cloud Storage V2] を選択します。
  6. [ログタイプ] として [ATLASSIAN_AUDIT] を選択します。
  7. [サービス アカウントを取得する] をクリックします。
  8. 一意のサービス アカウントのメールアドレスが表示されます。例:

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
    
  9. このメールアドレスをコピーして、次のステップで使用します。

  10. [次へ] をクリックします。

  11. 次の入力パラメータの値を指定します。

    • ストレージ バケットの URL: 接頭辞パスを含む GCS バケット URI を入力します。

      gs://atlassian-audit-logs/atlassian-audit/
      
    • Source deletion option: 必要に応じて削除オプションを選択します。

      • なし: 転送後にファイルを削除しません(テストに推奨)。
      • 転送されたファイルを削除する: 転送が完了した後にファイルを削除します。
      • 転送されたファイルと空のディレクトリを削除する: 転送が成功した後にファイルと空のディレクトリを削除します。

    • ファイルの最長存続期間: 過去数日間に変更されたファイルを含めます(デフォルトは 180 日)。

    • Asset namespace: アセットの名前空間。

    • Ingestion labels: このフィードのイベントに適用されるラベル。

  12. [次へ] をクリックします。

  13. [Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。

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

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

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

UDM マッピング テーブル

ログフィールド UDM マッピング ロジック
attributes.time metadata.event_timestamp 未加工ログの attributes.time フィールドからパースされます。
attributes.action metadata.product_event_type 値は attributes.action フィールドから取得されます。
attributes.actor.id principal.user.product_object_id 値は attributes.actor.id フィールドから取得されます。
attributes.actor.name principal.user.user_display_name 値は attributes.actor.name フィールドから取得されます。
attributes.actor.email principal.user.email_addresses 値は attributes.actor.email フィールドから取得されます。
attributes.location.ip principal.ip 値は attributes.location.ip フィールドから取得されます。
attributes.location.geo principal.location.country_or_region 値は attributes.location.geo フィールドから取得されます。
attributes.location.countryName principal.location.name 値は attributes.location.countryName フィールドから取得されます。
attributes.context[].id target.resource.product_object_id 値はコンテキスト id フィールドから取得されます。
attributes.context[].type target.resource.resource_subtype 値はコンテキスト type フィールドから取得されます。
attributes.container[].id target.resource.attribute.labels コンテナ id がラベルとして追加されます。
attributes.container[].type target.resource.attribute.labels コンテナ type がラベルとして追加されます。
message.content security_result.summary 値は message.content フィールドから取得されます。
metadata.event_type 使用可能なフィールドに基づいてマッピングされます。リソースの変更の場合は USER_RESOURCE_UPDATE_CONTENT、それ以外の場合は GENERIC_EVENT。
metadata.product_name Atlassian Admin に設定します。
metadata.vendor_name Atlassian に設定します。

変更ログ

このパーサーの変更ログを表示する

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