Google は AI 技術を使用して、コンテンツをご希望の言語に翻訳しています。AI 翻訳には誤りが含まれる場合があります。

Atlassian Confluence のログを収集する

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

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

Google SecOps SIEM

このドキュメントでは、Atlassian Confluence ログを Google Security Operations に取り込む方法について説明します。パーサーは、まず Atlassian Confluence ログ用に設計された正規表現（grok パターン）を使用して、未加工のログメッセージからフィールドを抽出します。grok 解析が失敗した場合や、ログが JSON 形式の場合、コードはメッセージを JSON として解析しようとします。最後に、抽出されたフィールドは Google SecOps UDM スキーマにマッピングされ、追加のコンテキストで拡充されます。

始める前に

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

Google SecOps インスタンス
監査ログにアクセスできる Atlassian Confluence Cloud アカウント、または管理者権限を持つ Confluence Data Center/Server
GCP ベースの方法の場合: GCP（GCS、IAM、Cloud Run、Pub/Sub、Cloud Scheduler）への特権アクセス
Bindplane メソッドの場合: Windows Server 2016 以降、または systemd を使用する Linux ホスト

統合オプションの概要

このガイドでは、2 つの統合パスについて説明します。

オプション 1: Bindplane + Syslog 経由の Confluence Data Center/Server
オプション 2: GCP Cloud Run 関数 + GCS を介した Confluence Cloud Audit Logs（JSON 形式）

Confluence のデプロイタイプとインフラストラクチャに最適なオプションを選択します。

オプション 1: Bindplane + Syslog 経由の Confluence Data Center/Server

このオプションは、syslog 経由でログを Bindplane エージェントに送信し、そのログを Google SecOps に転送するように Confluence Data Center または Server を構成します。

Google SecOps の取り込み認証ファイルを取得する

Google SecOps コンソールにログインします。
[SIEM 設定] > [収集エージェント] に移動します。
[ダウンロード] をクリックして、取り込み認証ファイル をダウンロードします。
Bindplane エージェントがインストールされるシステムに、ファイルを安全に保存します。

注: この JSON ファイルには、Bindplane エージェントを Google SecOps に対して認証するための認証情報が含まれています。

Google SecOps のお客様 ID を取得する

Google SecOps コンソールにログインします。
[SIEM 設定] > [プロファイル] に移動します。
[組織の詳細情報] セクションから [お客様 ID] をコピーして保存します。

注: Bindplane エージェントエクスポータを構成するには、顧客 ID が必要です。

Bindplane エージェントをインストールする

次の手順に沿って、Windows または Linux オペレーティングシステムに Bindplane エージェントをインストールします。

Windows のインストール

管理者としてコマンドプロンプトまたは PowerShell を開きます。

次のコマンドを実行します。

msiexec /i "https://github.com/observIQ/bindplane-otel-collector/releases/latest/download/observiq-otel-collector.msi" /quiet

インストールが完了するまで待ちます。
次のコマンドを実行して、インストールの内容を確認します。
```
sc query observiq-otel-collector
```

サービスは RUNNING と表示されます。

Linux のインストール

root 権限または sudo 権限でターミナルを開きます。

次のコマンドを実行します。

sudo sh -c "$(curl -fsSlL https://github.com/observIQ/bindplane-otel-collector/releases/latest/download/install_unix.sh)" install_unix.sh

インストールが完了するまで待ちます。
次のコマンドを実行して、インストールの内容を確認します。
```
sudo systemctl status observiq-otel-collector
```

サービスが [アクティブ（実行中）] と表示されます。

その他のインストールリソース

その他のインストールオプションとトラブルシューティングについては、Bindplane エージェントのインストールガイドをご覧ください。

syslog を取り込んで Google SecOps に送信するように Bindplane エージェントを構成する

構成ファイルを探す

Linux:

sudo systemctl status observiq-otel-collector

Windows:

notepad "C:\Program Files\observIQ OpenTelemetry Collector\config.yaml"

構成ファイルを編集します。

config.yaml の内容全体を次の構成に置き換えます。

receivers:
  udplog:
    listen_address: "0.0.0.0:514"

exporters:
  chronicle/confluence_logs:
    compression: gzip
    creds_file_path: '/etc/bindplane-agent/ingestion-auth.json'
    customer_id: 'YOUR_CUSTOMER_ID'
    endpoint: malachiteingestion-pa.googleapis.com
    log_type: ATLASSIAN_CONFLUENCE
    raw_log_field: body
    ingestion_labels:
      service: confluence

service:
  pipelines:
    logs/confluence:
      receivers:
        - udplog
      exporters:
        - chronicle/confluence_logs

構成パラメータ

各プレースホルダを次のように置き換えます。

listen_address: 自社のインフラストラクチャでの必要性に応じて、ポートと IP アドレスを置き換えます。0.0.0.0:514 を使用して、ポート 514 のすべてのインターフェースをリッスンします。
creds_file_path: 認証ファイルが保存されたパスに更新します。
- Linux: /etc/bindplane-agent/ingestion-auth.json
- Windows: C:\Program Files\observIQ OpenTelemetry Collector\ingestion-auth.json
customer_id: YOUR_CUSTOMER_ID は、前の手順の実際のお客様 ID に置き換えます。
endpoint: リージョナルエンドポイント URL:
- 米国: malachiteingestion-pa.googleapis.com
- ヨーロッパ: europe-malachiteingestion-pa.googleapis.com
- アジア: asia-southeast1-malachiteingestion-pa.googleapis.com

構成ファイルを保存する

編集が完了したら、ファイルを保存します。

Linux: Ctrl+O、Enter、Ctrl+X の順に押します。
Windows: [ファイル> 保存] をクリックします。

Bindplane エージェントを再起動して変更を適用する

Linux で Bindplane エージェントを再起動するには:

次のコマンドを実行します。

sudo systemctl restart observiq-otel-collector

サービスが実行されていることを確認します。
```
sudo systemctl status observiq-otel-collector
```

ログでエラーを確認します。

sudo journalctl -u observiq-otel-collector -f

Windows で Bindplane エージェントを再起動するには:

次のいずれかのオプションを選択します。
- 管理者としてコマンドプロンプトまたは PowerShell を開きます。
```
net stop observiq-otel-collector && net start observiq-otel-collector
```
- サービスコンソール:
  1. Win+R キーを押して「services.msc」と入力し、Enter キーを押します。
  2. observIQ OpenTelemetry Collector を見つけます。
  3. 右クリックして [再起動] を選択します。
サービスが実行されていることを確認します。
```
sc query observiq-otel-collector
```

ログでエラーを確認します。

type "C:\Program Files\observIQ OpenTelemetry Collector\log\collector.log"

Confluence Data Center/Server で Syslog 転送を構成する

オプション A: ローカルログファイルを転送するように rsyslog を構成する（推奨）

ログをファイルに書き込むように Confluence を構成します（デフォルトの動作）。

rsyslog が存在しない場合はインストールします。

sudo apt-get install rsyslog  # Debian/Ubuntu
sudo yum install rsyslog      # RHEL/CentOS

rsyslog 構成ファイル /etc/rsyslog.d/confluence.conf を作成します。

# Forward Confluence logs to Bindplane
$ModLoad imfile

# Application logs
$InputFileName /opt/atlassian/confluence/logs/atlassian-confluence.log
$InputFileTag confluence-app:
$InputFileStateFile stat-confluence-app
$InputFileSeverity info
$InputFileFacility local0
$InputRunFileMonitor

# Audit logs (JSON format in DC/Server)
$InputFileName <confluence-home-directory>/log/audit/audit.log
$InputFileTag confluence-audit:
$InputFileStateFile stat-confluence-audit
$InputFileSeverity info
$InputFileFacility local1
$InputRunFileMonitor

# Forward to Bindplane agent
*.* @@BINDPLANE_AGENT_IP:514

BINDPLANE_AGENT_IP は、Bindplane エージェントの IP アドレス（192.168.1.100 など）に置き換えます。
Confluence のインストールに基づいてログファイルパスを調整します。
- 通常、アプリケーションログは <confluence-install>/logs/ または <local-home>/logs/ です。
- 監査ログ: <confluence-home-directory>/log/audit/（JSON 形式）
- Confluence のホームディレクトリを確認するには、[設定] > [一般設定] > [システム情報] に移動し、[Confluence Home] または [Local Home] を探します。
rsyslog を再起動します。
```
sudo systemctl restart rsyslog
```

オプション B: Log4j2 Syslog 転送を構成する

SSH または RDP を使用して Confluence サーバーにログインします。
Log4j2 構成ファイルは次の場所にあります。
```
<confluence-install>/confluence/WEB-INF/classes/log4j2.xml
```
注: Confluence のインストールに基づいてパスを調整します。

構成ファイルを編集して Syslog アペンダーを追加します。

<Configuration>
  <Appenders>
    <!-- Existing appenders -->
    <Syslog name="SyslogAppender" 
            host="BINDPLANE_AGENT_IP" 
            port="514" 
            protocol="UDP"
            format="RFC 5424"
            facility="LOCAL0">
      <PatternLayout pattern="%d{ISO8601} %p [%t] [%c{1}] %m%n"/>
    </Syslog>
  </Appenders>

  <Loggers>
    <Root level="info">
      <AppenderRef ref="SyslogAppender"/>
      <!-- Other appender refs -->
    </Root>

    <!-- Audit logger -->
    <Logger name="com.atlassian.confluence.event.events.security.AuditEvent" 
            level="info" 
            additivity="false">
      <AppenderRef ref="SyslogAppender"/>
    </Logger>
  </Loggers>
</Configuration>

BINDPLANE_AGENT_IP は、Bindplane エージェントの IP アドレス（192.168.1.100 など）に置き換えます。
Confluence を再起動して変更を適用します。
```
sudo systemctl restart confluence
```

オプション 2: GCP Cloud Run 関数と GCS を介した Confluence Cloud 監査ログ

Confluence Cloud API 認証情報を収集する

Atlassian アカウントにログインします。
https://id.atlassian.com/manage-profile/security/api-tokens にアクセスします。
[Create API Token] をクリックします。
トークンのラベルを入力します（例: Google Security Operations Integration）。
[作成] をクリックします。
API トークンをコピーして安全に保存します。
Confluence Cloud サイトの URL（例: https://yoursite.atlassian.net）をメモします。
Atlassian アカウントのメールアドレス（認証に使用）をメモします。

注: API トークンには、監査ログにアクセスするための「Confluence 管理者」グローバル権限が必要です。

権限を確認する

アカウントに必要な権限があることを確認するには:

Confluence Cloud にログインします。
右上にある設定アイコンをクリックします。
左側のナビゲーションに [モニタリング] > [監査ログ] が表示されている場合は、必要な権限があります。
このオプションが表示されない場合は、管理者にお問い合わせのうえ、Confluence 管理者権限を付与してもらってください。

テスト API へのアクセス

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

# Replace with your actual credentials
CONFLUENCE_EMAIL="your-email@example.com"
CONFLUENCE_API_TOKEN="your-api-token"
CONFLUENCE_URL="https://yoursite.atlassian.net"

# Test API access
curl -u "${CONFLUENCE_EMAIL}:${CONFLUENCE_API_TOKEN}" \
  -H "Accept: application/json" \
  "${CONFLUENCE_URL}/wiki/rest/api/audit"

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Pub/Sub トピックの作成

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

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

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

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

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

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

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

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

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

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

変数名	値の例	説明
`GCS_BUCKET`	`confluence-audit-logs`	GCS バケット名
`GCS_PREFIX`	`confluence-audit`	ログファイルの接頭辞
`STATE_KEY`	`confluence-audit/state.json`	状態ファイルのパス
`CONFLUENCE_URL`	`https://yoursite.atlassian.net`	Confluence サイトの URL
`CONFLUENCE_EMAIL`	`your-email@example.com`	Atlassian アカウントのメールアドレス
`CONFLUENCE_API_TOKEN`	`your-api-token-here`	API トークン
`MAX_RECORDS`	`1000`	実行あたりの最大レコード数

[変数とシークレット] セクションで、[リクエスト] まで下にスクロールします。
- リクエストのタイムアウト: 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
import time
import base64

# 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', 'confluence-audit/')
STATE_KEY = os.environ.get('STATE_KEY', 'confluence-audit/state.json')
CONFLUENCE_URL = os.environ.get('CONFLUENCE_URL')
CONFLUENCE_EMAIL = os.environ.get('CONFLUENCE_EMAIL')
CONFLUENCE_API_TOKEN = os.environ.get('CONFLUENCE_API_TOKEN')
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '1000'))

def to_unix_millis(dt: datetime) -> int:
    """Convert datetime to Unix epoch milliseconds."""
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    dt = dt.astimezone(timezone.utc)
    return int(dt.timestamp() * 1000)

def parse_datetime(value: str) -> datetime:
    """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 Confluence Cloud audit logs and write to GCS.

    Args:
        cloud_event: CloudEvent object containing Pub/Sub message
    """

    if not all([GCS_BUCKET, CONFLUENCE_URL, CONFLUENCE_EMAIL, CONFLUENCE_API_TOKEN]):
        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=24)

        print(f"Fetching logs from {last_time.isoformat()} to {now.isoformat()}")

        # Convert to Unix milliseconds
        start_millis = to_unix_millis(last_time)
        end_millis = to_unix_millis(now)

        # Fetch logs
        records, newest_event_time = fetch_logs(
            api_base=CONFLUENCE_URL,
            email=CONFLUENCE_EMAIL,
            api_token=CONFLUENCE_API_TOKEN,
            start_time_ms=start_millis,
            end_time_ms=end_millis,
            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 = now.strftime('%Y%m%d_%H%M%S')
        object_key = f"{GCS_PREFIX}/logs_{timestamp}.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: str):
    """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(api_base: str, email: str, api_token: str, start_time_ms: int, end_time_ms: int, max_records: int):
    """
    Fetch logs from Confluence Cloud Audit API with pagination and rate limiting.

    Args:
        api_base: Confluence site URL
        email: Atlassian account email
        api_token: API token
        start_time_ms: Start time in Unix milliseconds
        end_time_ms: End time in Unix milliseconds
        max_records: Maximum total records to fetch

    Returns:
        Tuple of (records list, newest_event_time ISO string)
    """
    # Clean up URL
    base_url = api_base.rstrip('/')

    # Build authentication header
    auth_string = f"{email}:{api_token}"
    auth_bytes = auth_string.encode('utf-8')
    auth_b64 = base64.b64encode(auth_bytes).decode('utf-8')
    headers = {
        'Authorization': f'Basic {auth_b64}',
        'Accept': 'application/json',
        'User-Agent': 'GoogleSecOps-ConfluenceCollector/1.0'
    }

    records = []
    newest_time = None
    page_num = 0
    backoff = 1.0
    start_index = 0

    while True:
        page_num += 1

        if len(records) >= max_records:
            print(f"Reached max_records limit ({max_records})")
            break

        # Build request URL
        url = f"{base_url}/wiki/rest/api/audit?startDate={start_time_ms}&endDate={end_time_ms}&start={start_index}&limit=100"

        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('results', [])

            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
            for event in page_results:
                try:
                    # creationDate is in Unix milliseconds
                    event_time_ms = event.get('creationDate')
                    if event_time_ms:
                        event_dt = datetime.fromtimestamp(event_time_ms / 1000, tz=timezone.utc)
                        event_time = event_dt.isoformat()
                        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 more results
            current_size = data.get('size', 0)
            if current_size < 100:
                print(f"Reached last page (size={current_size} < limit=100)")
                break

            start_index += current_size

        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[:max_records], newest_time

requirements.txt:

functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0

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

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

Cloud Scheduler ジョブの作成

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

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

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

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

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

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

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

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

統合をテストする

Cloud Scheduler コンソールで、ジョブを見つけます。
[強制実行] をクリックして、ジョブを手動でトリガーします。
数秒待ちます。
Cloud Run > サービスに移動します。
confluence-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://bucket-name/prefix/logs_YYYYMMDD_HHMMSS.ndjson
Successfully processed X records

[Cloud Storage] > [バケット] に移動します。
バケット名をクリックします。
confluence-audit/ フォルダに移動します。
現在のタイムスタンプで新しい .ndjson ファイルが作成されたことを確認します。

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

HTTP 401: 環境変数で API 認証情報を確認する
HTTP 403: アカウントに Confluence 管理者権限があることを確認する
HTTP 429: レート制限 - 関数はバックオフで自動的に再試行されます
環境変数が不足している: 必要な変数がすべて設定されていることを確認します

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

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

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

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

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

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

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

UDM マッピングテーブル

ログフィールド	UDM マッピング	ロジック
エージェント	read_only_udm.network.http.user_agent	「agent」フィールドから取得された値。
app_protocol	read_only_udm.network.application_protocol	「app_protocol」フィールドから派生します。「app_protocol」に「HTTPS」、「HTTP」、「SSH」、「RDP」が含まれている場合は、対応するプロトコルが使用されます。それ以外の場合は、デフォルトで「UNKNOWN_APPLICATION_PROTOCOL」になります。
app_protocol	read_only_udm.network.application_protocol_version	「app_protocol」フィールドから取得された値。
auditType.action	read_only_udm.security_result.action	「auditType.action」フィールドから派生します。「auditType.action」に「successful」が含まれている場合、値は「ALLOW」に設定されます。「restricted」が含まれている場合、値は「BLOCK」に設定されます。
auditType.action	read_only_udm.security_result.summary	「auditType」が空でなく、「auditType_area」が「SECURITY」の場合、「auditType.action」フィールドから取得された値。
auditType.actionI18nKey	read_only_udm.metadata.product_event_type	「auditType」が空でない場合、「auditType.actionI18nKey」フィールドから取得された値。
auditType.area	read_only_udm.security_result.detection_fields.value	「auditType.area」フィールドから取得された値が、検出フィールドの「value」フィールドに割り当てられます。この検出フィールドの「key」フィールドは「auditType area」に設定されています。このマッピングは、「auditType」が空でない場合に実行されます。
auditType.category	read_only_udm.security_result.category_details	「auditType」が空でない場合、「auditType.category」フィールドから取得された値。
auditType.categoryI18nKey	read_only_udm.security_result.detection_fields.value	「auditType.categoryI18nKey」フィールドから取得された値が、検出フィールドの「value」フィールドに割り当てられます。この検出フィールドの「key」フィールドは「auditType categoryI18nKey」に設定されています。このマッピングは、「auditType」が空でない場合に実行されます。
auditType.level	read_only_udm.security_result.detection_fields.value	「auditType.level」フィールドから取得された値が、検出フィールドの「value」フィールドに割り当てられます。この検出フィールドの「key」フィールドは「auditType level」に設定されています。このマッピングは、「auditType」が空でない場合に実行されます。
author.displayName	read_only_udm.principal.user.user_display_name	「author.displayName」フィールドから取得された値。
author.externalCollaborator	read_only_udm.security_result.about.resource.attribute.labels.value	「author.externalCollaborator」フィールドから取得された値が、「key」フィールドが「externalCollaborator」に設定されたラベルの「value」フィールドに割り当てられます。
author.id	read_only_udm.principal.user.userid	「author.type」が「user」で、「principal_user_present」が「false」の場合、「author.id」フィールドから取得された値。
author.isExternalCollaborator	read_only_udm.security_result.about.resource.attribute.labels.value	「author.isExternalCollaborator」フィールドから取得された値が、「key」フィールドが「isExternalCollaborator」に設定されたラベルの「value」フィールドに割り当てられます。
author.name	read_only_udm.principal.user.user_display_name	「author.type」が「user」で「principal_user_present」が「false」の場合、「author.name」フィールドから取得された値。
bytes_in	read_only_udm.network.received_bytes	「bytes_in」フィールドに数字が含まれている場合は、その値が使用されます。それ以外の場合は、デフォルトの 0 になります。
category	read_only_udm.security_result.category_details	「category」フィールドから取得された値。
changedValues	read_only_udm.principal.resource.attribute.labels	「changedValues」の各要素を反復処理し、「changedValue [index] [key]」のようなキーと、「changedValues」配列の対応する値から値を持つラベルを作成します。
作成日	read_only_udm.metadata.event_timestamp	「creationDate」フィールドから取得された値。UNIX または UNIX_MS タイムスタンプとして解析されます。
extraAttributes	read_only_udm.principal.resource.attribute.labels	「extraAttributes」の各要素を反復処理し、「name」フィールドと「nameI18nKey」フィールドに基づいてキーを持つラベルと、対応する「value」フィールドの値を作成します。
http_verb	read_only_udm.network.http.method	「http_verb」フィールドから取得された値。
ip	read_only_udm.target.ip	「ip」フィールドから取得された値。
principal_host	read_only_udm.principal.hostname	「principal_host」フィールドから取得された値。
referral_url	read_only_udm.network.http.referral_url	「referral_url」フィールドから取得された値。
remoteAddress	read_only_udm.principal.ip	「remoteAddress」フィールドから取得された値。IP アドレスとして解析されます。
response_code	read_only_udm.network.http.response_code	「response_code」フィールドから取得された値。
session_duration	read_only_udm.additional.fields.value.string_value	「session_duration」フィールドから取得した値が、ラベルの「string_value」フィールドに割り当てられます。このラベルの「key」フィールドは「Session Duration」に設定されています。
ソース	read_only_udm.principal.ip	「source」フィールドから取得された値。IP アドレスとして解析されます。
src_ip	read_only_udm.principal.ip	「remoteAddress」が空の場合、「src_ip」フィールドから取得された値。
概要	read_only_udm.security_result.summary	「summary」フィールドから取得された値。
sysAdmin	read_only_udm.security_result.about.resource.attribute.labels.value	「sysAdmin」フィールドから取得された値が、ラベルの「value」フィールドに割り当てられます。このラベルの「key」フィールドは「sysAdmin」に設定されています。
superAdmin	read_only_udm.security_result.about.resource.attribute.labels.value	「superAdmin」フィールドから取得された値が、ラベルの「value」フィールドに割り当てられます。このラベルの「key」フィールドは「superAdmin」に設定されています。
target_url	read_only_udm.target.url	「target_url」フィールドから取得された値。
timestamp	read_only_udm.metadata.event_timestamp	「timestamp」フィールドから取得された値。日付と時刻の文字列として解析されます。
user_id	read_only_udm.principal.user.userid	「user_id」フィールドから取得された値。
	read_only_udm.metadata.event_type	このフィールドの値は一連のチェックによって決定され、デフォルトは「GENERIC_EVENT」です。「principal_host」、「user_id」、「has_principal」、「author.type」などの他のフィールドの有無と内容に基づいて、「NETWORK_HTTP」、「USER_UNCATEGORIZED」、「STATUS_UPDATE」などの特定の値に設定されます。
	read_only_udm.metadata.vendor_name	「ATLASSIAN」に設定します。
	read_only_udm.metadata.product_name	「CONFLUENCE」に設定します。
	read_only_udm.metadata.log_type	「ATLASSIAN_CONFLUENCE」に設定します。
	read_only_udm.principal.user.user_display_name	このフィールドの値は、コンテキストに応じて「author.displayName」または「affectedObject.name」から取得できます。
	read_only_udm.target.process.pid	このフィールドの値は、コンテキストに応じて「principal_host」または「pid」のいずれかから取得できます。
	read_only_udm.principal.resource.attribute.labels	このフィールドには、「affectedObjects」、「changedValues」、「extraAttributes」などのフィールドから派生したさまざまなラベルが入力されます。これらのラベルのキーと値は、これらのフィールドの特定のコンテンツに基づいて動的に生成されます。
`affectedObjects.id", "affectedObjects.name", "affectedObjects.type`	`principal.resource.attribute.labels`	変更履歴からマッピング
`affectedObjects.id`	`principal.user.userid`	変更履歴からマッピング
`affectedObjects.name`	`principal.user.user_display_name`	変更履歴からマッピング
`changedValues.from", "changedValues.i18nKey", "changedValues.key" and "changedValues.to`	`principal.resource.attribute.labels`	変更履歴からマッピング
`extraAttributes.name", "extraAttributes.nameI18nKey" and "extraAttributes.value`	`principal.resource.attribute.labels`	変更履歴からマッピング
`system`	`principal.hostname`	変更履歴からマッピング
`session_duration`	`additional fields`	変更履歴からマッピング

変更履歴

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

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

Atlassian Confluence のログを収集する

始める前に

統合オプションの概要

オプション 1: Bindplane + Syslog 経由の Confluence Data Center/Server

Google SecOps の取り込み認証ファイルを取得する

Google SecOps のお客様 ID を取得する

Bindplane エージェントをインストールする

Windows のインストール

Linux のインストール

その他のインストール リソース

syslog を取り込んで Google SecOps に送信するように Bindplane エージェントを構成する

構成ファイルを探す

構成ファイルを編集します。

構成パラメータ

構成ファイルを保存する

Bindplane エージェントを再起動して変更を適用する

Confluence Data Center/Server で Syslog 転送を構成する

オプション A: ローカル ログファイルを転送するように rsyslog を構成する（推奨）

オプション B: Log4j2 Syslog 転送を構成する

オプション 2: GCP Cloud Run 関数と GCS を介した Confluence Cloud 監査ログ

Confluence Cloud API 認証情報を収集する

権限を確認する

テスト API へのアクセス

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

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

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

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

Pub/Sub トピックの作成

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

関数コードを追加する

Cloud Scheduler ジョブの作成

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

統合をテストする

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

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

UDM マッピング テーブル

変更履歴

その他のインストールリソース

オプション A: ローカルログファイルを転送するように rsyslog を構成する（推奨）

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

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

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

UDM マッピングテーブル