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 バケットを作成する
- Google Cloud コンソールに移動します。
- プロジェクトを選択するか、新しいプロジェクトを作成します。
- ナビゲーション メニューで、[Cloud Storage] > [バケット] に移動します。
- [バケットを作成] をクリックします。
次の構成情報を提供してください。
設定 値 バケットに名前を付ける グローバルに一意の名前( atlassian-audit-logsなど)を入力します。ロケーション タイプ ニーズに応じて選択します(リージョン、デュアルリージョン、マルチリージョン)。 ロケーション ロケーションを選択します(例: us-central1)。ストレージ クラス Standard(頻繁にアクセスされるログに推奨) アクセス制御 均一(推奨) 保護ツール 省略可: オブジェクトのバージョニングまたは保持ポリシーを有効にする [作成] をクリックします。
Atlassian API 認証情報を収集する
API キーを作成
- admin.atlassian.com にログインします。
- 複数の組織がある場合は、組織を選択します。
- [設定] > [API キー] に移動します。
- [API キーを作成] をクリックします。
- 鍵のわかりやすい名前を入力します(例:
SecOps-Audit-Collector)。 [有効期限] の日付を設定します。
[作成] をクリックします。
API キーの値をコピーして安全に保存します。このダイアログを閉じると、API キーの値を復元できなくなります。
ブラウザのアドレスバーの URL(
https://admin.atlassian.com/o/{orgId}/...の UUID)から組織 ID をメモします。[完了] をクリックします。
権限を確認する
API を介して組織の監査ログにアクセスするには、組織管理者である必要があります。
- admin.atlassian.com にログインします。
- [設定] > [API キー] に移動します。
- [API キー] ページが表示され、キーを作成できる場合は、必要な組織管理者権限があります。
- このオプションが表示されない場合は、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 による呼び出し権限を持つサービス アカウントが必要です。
サービス アカウントの作成
- GCP コンソールで、[IAM と管理] > [サービス アカウント] に移動します。
- [サービス アカウントを作成] をクリックします。
- 次の構成情報を指定します。
- サービス アカウント名: 「
atlassian-audit-collector-sa」と入力します。 - サービス アカウントの説明: 「
Service account for Cloud Run function to collect Atlassian Organization Audit logs」と入力します。
- サービス アカウント名: 「
- [作成して続行] をクリックします。
- [このサービス アカウントにプロジェクトへのアクセスを許可する] セクションで、次のロールを追加します。
- [ロールを選択] をクリックします。
- [Storage オブジェクト管理者] を検索して選択します。
- [+ 別のロールを追加] をクリックします。
- [Cloud Run 起動元] を検索して選択します。
- [+ 別のロールを追加] をクリックします。
- [Cloud Functions 起動元] を検索して選択します。
- [続行] をクリックします。
- [完了] をクリックします。
これらのロールは、次の操作に必要です。
- Storage オブジェクト管理者: GCS バケットにログを書き込み、状態ファイルを管理します。
- Cloud Run 起動元: Pub/Sub が関数を呼び出せるようにします。
- Cloud Functions 起動元: 関数の呼び出しを許可します。
GCS バケットに対する IAM 権限を付与する
GCS バケットに対する書き込み権限をサービス アカウントに付与します。
- [Cloud Storage] > [バケット] に移動します。
- バケット名(
atlassian-audit-logsなど)をクリックします。 - [権限] タブに移動します。
- [アクセス権を付与] をクリックします。
- 次の構成情報を指定します。
- プリンシパルを追加: サービス アカウントのメールアドレス(例:
atlassian-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com)を入力します。 - ロールを割り当てる: [Storage オブジェクト管理者] を選択します。
- プリンシパルを追加: サービス アカウントのメールアドレス(例:
- [保存] をクリックします。
Pub/Sub トピックの作成
Cloud Scheduler がパブリッシュし、Cloud Run functions の関数がサブスクライブする Pub/Sub トピックを作成します。
- GCP コンソールで、[Pub/Sub] > [トピック] に移動します。
- [トピックを作成] をクリックします。
- 次の構成情報を指定します。
- トピック ID: 「
atlassian-audit-logs-trigger」と入力します。 - その他の設定はデフォルトのままにします。
- トピック ID: 「
- [作成] をクリックします。
ログを収集する Cloud Run 関数を作成する
Cloud Run functions の関数は、Cloud Scheduler からの Pub/Sub メッセージによってトリガーされ、Atlassian Organizations REST API から監査ログイベントをポーリングして GCS に書き込みます。
- GCP Console で Cloud Run に移動します。
- [サービスを作成] をクリックします。
- [関数](インライン編集を使用して関数を作成する)を選択します。
[構成] セクションで、次の構成の詳細を指定します。
設定 値 サービス名 atlassian-audit-collectorリージョン GCS バケットに一致するリージョンを選択します(例: us-central1)。ランタイム [Python 3.12] 以降を選択します。 [トリガー(省略可)] セクションで、次の操作を行います。
- [+ トリガーを追加] をクリックします。
- [Cloud Pub/Sub] を選択します。
- [Cloud Pub/Sub トピックを選択してください] で、Pub/Sub トピック(
atlassian-audit-logs-trigger)を選択します。 - [保存] をクリックします。
[認証] セクションで、次の操作を行います。
- [認証が必要] を選択します。
- Identity and Access Management(IAM)を確認します。
下にスクロールして、[コンテナ、ネットワーキング、セキュリティ] を開きます。
[セキュリティ] タブに移動します。
- サービス アカウント: サービス アカウント(
atlassian-audit-collector-sa)を選択します。
- サービス アカウント: サービス アカウント(
[コンテナ] タブに移動します。
- [変数とシークレット] をクリックします。
- 環境変数ごとに [+ 変数を追加] をクリックします。
変数名 値の例 説明 GCS_BUCKETatlassian-audit-logsGCS バケット名 GCS_PREFIXatlassian-auditログファイルの接頭辞 STATE_KEYatlassian-audit/state.json状態ファイルのパス ATL_TOKENyour-atlassian-api-keyAtlassian API キー ATL_ORG_IDyour-organization-idAtlassian 組織 ID MAX_RECORDS10001 回の実行あたりの最大レコード数 PAGE_SIZE2001 ページあたりのレコード数(最大 500) LOOKBACK_HOURS24最初のルックバック期間 [変数とシークレット] セクションで、[リクエスト] まで下にスクロールします。
- リクエストタイムアウト:
600秒(10 分)を入力してください
- リクエストタイムアウト:
[設定] タブに移動します。
- [リソース] セクションで、次の操作を行います。
- メモリ: 512 MiB 以上を選択します。
- CPU: 1 を選択します。
- [リソース] セクションで、次の操作を行います。
[リビジョン スケーリング] セクションで、次の操作を行います。
- [インスタンスの最小数]: 「
0」と入力します。 - [インスタンスの最大数]: 「
100」と入力します(または、予想される負荷に基づいて調整します)。
- [インスタンスの最小数]: 「
[作成] をクリックします。
サービスが作成されるまで待ちます(1 ~ 2 分)。
サービスが作成されると、インライン コードエディタが自動的に開きます。
関数コードを追加する
- [エントリ ポイント] フィールドに「main」と入力します。
インライン コードエディタで、次の 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[デプロイ] をクリックして関数を保存し、デプロイします。
デプロイが完了するまで待ちます(2 ~ 3 分)。
Cloud Scheduler ジョブの作成
Cloud Scheduler は、定期的に Pub/Sub トピックにメッセージをパブリッシュし、Cloud Run 関数をトリガーします。
- GCP Console で、[Cloud Scheduler] に移動します。
- [ジョブを作成] をクリックします。
次の構成情報を提供してください。
設定 値 名前 atlassian-audit-collector-hourlyリージョン Cloud Run functions の関数と同じリージョンを選択します。 周波数 0 * * * *(1 時間ごと、毎正時)タイムゾーン タイムゾーンを選択(UTC を推奨) ターゲット タイプ Pub/Sub トピック Pub/Sub トピック( atlassian-audit-logs-trigger)を選択します。メッセージ本文 {}(空の JSON オブジェクト)[作成] をクリックします。
スケジュール頻度のオプション
ログの量とレイテンシの要件に基づいて頻度を選択します。
| 頻度 | CRON 式 | ユースケース |
|---|---|---|
| 5 分おき | */5 * * * * |
大容量、低レイテンシ |
| 15 分ごと | */15 * * * * |
検索量が普通 |
| 1 時間ごと | 0 * * * * |
標準(推奨) |
| 6 時間ごと | 0 */6 * * * |
少量のバッチ処理 |
| 毎日 | 0 0 * * * |
過去のデータの収集 |
統合をテストする
- Cloud Scheduler コンソールで、ジョブを見つけます。
- [強制実行] をクリックして、ジョブを手動でトリガーします。
- 数秒待ちます。
- [Cloud Run] > [サービス] に移動します。
- 関数名(
atlassian-audit-collector)をクリックします。 - [ログ] タブをクリックします。
関数が正常に実行されたことを確認します。以下のものを探します。
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[Cloud Storage] > [バケット] に移動します。
バケット名(
atlassian-audit-logs)をクリックします。接頭辞フォルダ(
atlassian-audit/)に移動します。現在のタイムスタンプで新しい
.ndjsonファイルが作成されたことを確認します。
ログにエラーが表示される場合:
- HTTP 401: 環境変数の API キーを確認します。
- HTTP 403: 組織管理者権限と Atlassian Guard サブスクリプションを確認します。
- HTTP 429: レート制限 - 関数はバックオフを使用して自動的に再試行します
- 環境変数が設定されていない: 必要な変数がすべて設定されていることを確認します。
Google SecOps サービス アカウントを取得する
Google SecOps は、一意のサービス アカウントを使用して GCS バケットからデータを読み取ります。このサービス アカウントにバケットへのアクセス権を付与する必要があります。
サービス アカウントのメールアドレスを取得する
- [設定] > [データ管理] > [フィード] に移動します。SIEM のみのプラットフォームを使用している場合は、[設定] > [フィード] に移動します。
- [Add New Feed] をクリックします。
- [単一フィードを設定] をクリックします。
- [フィード名] フィールドに、フィードの名前を入力します(例:
Atlassian Audit Logs)。 - [Source type] として [Google Cloud Storage V2] を選択します。
- [ログタイプ] として [ATLASSIAN_AUDIT] を選択します。
- [サービス アカウントを取得する] をクリックします。
一意のサービス アカウントのメールアドレスが表示されます。例:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.comこのメールアドレスをコピーして、次のステップで使用します。
[次へ] をクリックします。
次の入力パラメータの値を指定します。
ストレージ バケットの URL: 接頭辞パスを含む GCS バケット URI を入力します。
gs://atlassian-audit-logs/atlassian-audit/
Source deletion option: 必要に応じて削除オプションを選択します。
- なし: 転送後にファイルを削除しません(テストに推奨)。
- 転送されたファイルを削除する: 転送が完了した後にファイルを削除します。
転送されたファイルと空のディレクトリを削除する: 転送が成功した後にファイルと空のディレクトリを削除します。
ファイルの最長存続期間: 過去数日間に変更されたファイルを含めます(デフォルトは 180 日)。
Asset namespace: アセットの名前空間。
Ingestion labels: このフィードのイベントに適用されるラベル。
[次へ] をクリックします。
[Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。
Google SecOps サービス アカウントに IAM 権限を付与する
Google SecOps サービス アカウントには、GCS バケットに対する Storage オブジェクト閲覧者のロールが必要です。
- [Cloud Storage] > [バケット] に移動します。
- バケット名(
atlassian-audit-logs)をクリックします。 - [権限] タブに移動します。
- [アクセス権を付与] をクリックします。
- 次の構成情報を指定します。
- プリンシパルを追加: Google SecOps サービス アカウントのメールアドレスを貼り付けます。
- ロールを割り当てる: [Storage オブジェクト閲覧者] を選択します。
[保存] をクリックします。
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 のプロフェッショナルから回答を得ることができます。