1Password Audit Events ログを収集する

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

Google SecOps SIEM

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

1Password は、チームが認証情報、シークレット、機密情報を安全に保存、共有、管理できるようにするパスワード管理プラットフォームです。1Password Events API を使用すると、1Password Business アカウントで行われた管理アクションとポリシーアクションをキャプチャする監査イベントデータにアクセスできます。

始める前に

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

Google SecOps インスタンス
Cloud Storage API が有効になっている GCP プロジェクト
GCS バケットを作成および管理する権限
GCS バケットの IAM ポリシーを管理する権限
Cloud Run サービス、Pub/Sub トピック、Cloud Scheduler ジョブを作成する権限
1Password Business アカウント
1Password アカウントのオーナーまたは管理者ロール

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

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

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

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

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

1Password の認証情報を収集する

イベントレポートの統合を設定する

1Password.com でアカウントにログインします。
サイドバーで [統合] を選択します。
他の統合をすでに設定している場合は、[統合] ページで [ディレクトリ] を選択します。
[イベントレポート] セクションで、[その他] を選択します。
[名前] フィールドに、統合の名前を入力します（例: Google SecOps Audit Events）。
[統合を追加] をクリックします。

署名なしトークンを作成する

統合の詳細ページで、[ベアラートークンを追加] をクリックします。
次の構成の詳細を指定します。
- トークン名: わかりやすい名前を入力します（例: SecOps GCS Collector - Audit Events）。
- Expires After: 必要に応じて有効期限を選択します。有効期限のないトークンには [なし] を選択し、有効期限のあるトークンには [30 日]、[90 日]、[180 日] のいずれかを選択します。
- [レポートするイベント]: 次のチェックボックスをオンにします。
  - 監査イベント
[Issue Token] をクリックします。
[Save your token] ページで、[Save in 1Password] をクリックするか、トークンをコピーして安全な場所に保存します。

注: ベアラートークンは、1 回だけ表示される JWT 認証情報です。トークンを紛失した場合は、トークンを取り消して新しいトークンを発行する必要があります。
[統合の詳細を表示] をクリックして、統合が有効になっていることを確認します。

Events API のベース URL を確認する

ベース URL は、1Password アカウントをホストするサーバーによって異なります。

アカウントが	Events API のベース URL は次のとおりです。
`1password.com`	`https://events.1password.com`
`ent.1password.com`	`https://events.ent.1password.com`
`1password.ca`	`https://events.1password.ca`
`1password.eu`	`https://events.1password.eu`

テスト API へのアクセス

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

# Replace with your actual bearer token and base URL
BEARER_TOKEN="<your-bearer-token>"
API_BASE="https://events.1password.com"

# Test API access using the introspect endpoint
curl -v \
    -H "Authorization: Bearer $BEARER_TOKEN" \
    "$API_BASE/api/v2/auth/introspect"

成功したレスポンスは、トークンがアクセスできるイベントタイプ（["auditevents"] など）を一覧表示する features フィールドを含む JSON オブジェクトを返します。

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

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

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

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

これらのロールは、次の目的で必要です。

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

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

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

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

Pub/Sub トピックの作成

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

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

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

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

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

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

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

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

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

[変数とシークレット] をクリックします。

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

変数名	値の例	説明
`GCS_BUCKET`	`onepassword-audit-secops-logs`	GCS バケット名
`GCS_PREFIX`	`onepassword-audit`	ログファイルの接頭辞
`STATE_KEY`	`onepassword-audit/state.json`	状態ファイルのパス
`OP_API_BASE`	`https://events.1password.com`	1Password Events API のベース URL
`OP_BEARER_TOKEN`	`<your-bearer-token>`	1Password Events API ベアラートークン
`MAX_RECORDS`	`10000`	実行あたりの最大レコード数
`PAGE_SIZE`	`1000`	ページあたりのレコード数（最大 1,000）
`LOOKBACK_HOURS`	`24`	最初のルックバック期間（時間単位）

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

# 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', 'onepassword-audit')
STATE_KEY = os.environ.get('STATE_KEY', 'onepassword-audit/state.json')
API_BASE = os.environ.get('OP_API_BASE')
BEARER_TOKEN = os.environ.get('OP_BEARER_TOKEN')
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000'))
PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '1000'))
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))

# 1Password Events API v2 audit events endpoint
AUDIT_EVENTS_PATH = '/api/v2/auditevents'

def parse_datetime(value: str) -> datetime:
    """Parse RFC 3339 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 1Password
    audit event logs and write them to GCS as NDJSON.
    """
    if not all([GCS_BUCKET, API_BASE, BEARER_TOKEN]):
        print('Error: Missing required environment variables '
            '(GCS_BUCKET, OP_API_BASE, OP_BEARER_TOKEN)')
        return

    try:
        bucket = storage_client.bucket(GCS_BUCKET)

        # Load state (stores cursor)
        state = load_state(bucket, STATE_KEY)

        now = datetime.now(timezone.utc)

        print('--- Processing endpoint: auditevents ---')

        saved_cursor = state.get('cursor_auditevents')

        records, last_cursor = fetch_endpoint(
            api_base=API_BASE,
            path=AUDIT_EVENTS_PATH,
            bearer_token=BEARER_TOKEN,
            saved_cursor=saved_cursor,
            lookback_hours=LOOKBACK_HOURS,
            page_size=PAGE_SIZE,
            max_records=MAX_RECORDS,
        )

        if records:
            timestamp = now.strftime('%Y%m%d_%H%M%S')
            object_key = (f'{GCS_PREFIX}/'
                        f'auditevents_{timestamp}.ndjson')
            blob = bucket.blob(object_key)

            ndjson = '\n'.join(
                json.dumps(r, ensure_ascii=False) for r in records
            ) + '\n'
            blob.upload_from_string(
                ndjson, content_type='application/x-ndjson'
            )

            print(f'Wrote {len(records)} auditevents records '
                f'to gs://{GCS_BUCKET}/{object_key}')

        # Save cursor even if no records (for next poll)
        if last_cursor:
            state['cursor_auditevents'] = last_cursor

        # Save state
        state['last_run'] = now.isoformat()
        save_state(bucket, STATE_KEY, state)

        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, state: dict):
    """Save state to GCS."""
    try:
        blob = bucket.blob(key)
        blob.upload_from_string(
            json.dumps(state, indent=2),
            content_type='application/json',
        )
        print(f'Saved state: {json.dumps(state)}')
    except Exception as e:
        print(f'Warning: Could not save state: {e}')

def fetch_endpoint(api_base, path, bearer_token, saved_cursor,
                lookback_hours, page_size, max_records):
    """
    Fetch events from the 1Password Events API v2 audit events
    endpoint.

    The 1Password Events API uses cursor-based pagination with POST
    requests. The first request sends a ResetCursor object with
    optional start_time and limit. Subsequent requests send the
    cursor returned from the previous response.

    Args:
        api_base: Events API base URL
        path: Endpoint path
        bearer_token: JWT bearer token
        saved_cursor: Cursor from previous run (or None)
        lookback_hours: Hours to look back on first run
        page_size: Max events per page (1-1000)
        max_records: Max total events per run

    Returns:
        Tuple of (records list, last_cursor string)
    """
    url = f'{api_base.rstrip("/")}{path}'

    headers = {
        'Authorization': f'Bearer {bearer_token}',
        'Content-Type': 'application/json',
        'Accept': 'application/json',
        'User-Agent': 'GoogleSecOps-1PasswordAuditCollector/1.0',
    }

    records = []
    cursor = saved_cursor
    page_num = 0
    backoff = 1.0

    while True:
        page_num += 1

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

        # Build request body
        if cursor:
            # Continuing cursor: resume from last position
            body = json.dumps({'cursor': cursor})
        else:
            # ResetCursor: first request or no saved state
            start_time = (
                datetime.now(timezone.utc)
                - timedelta(hours=lookback_hours)
            )
            body = json.dumps({
                'limit': page_size,
                'start_time': start_time.strftime(
                    '%Y-%m-%dT%H:%M:%SZ'
                ),
            })

        try:
            response = http.request(
                'POST', url, body=body, headers=headers,
            )

            # Handle rate limiting (600 req/min, 30000 req/hour)
            if response.status == 429:
                retry_after = int(
                    response.headers.get(
                        'Retry-After', str(int(backoff))
                    )
                )
                print(f'Rate limited (429). Retrying after '
                    f'{retry_after}s...')
                time.sleep(retry_after)
                backoff = min(backoff * 2, 60.0)
                continue

            backoff = 1.0

            if response.status != 200:
                response_text = response.data.decode('utf-8')
                print(f'HTTP Error {response.status}: '
                    f'{response_text}')
                break

            data = json.loads(response.data.decode('utf-8'))

            page_items = data.get('items', [])
            cursor = data.get('cursor')
            has_more = data.get('has_more', False)

            if page_items:
                print(f'Page {page_num}: Retrieved '
                    f'{len(page_items)} events')
                records.extend(page_items)

            if not has_more:
                print(f'No more pages (has_more=false)')
                break

            if not cursor:
                print(f'No cursor returned, stopping')
                break

        except urllib3.exceptions.HTTPError as e:
            print(f'HTTP error: {str(e)}')
            break
        except Exception as e:
            print(f'Error fetching events: {str(e)}')
            break

    print(f'Retrieved {len(records)} total records '
        f'from {page_num} pages')
    return records, cursor

2 つ目のファイル: 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] に移動します。
[ジョブを作成] をクリックします。

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

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

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

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

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

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

統合をテストする

Cloud Scheduler コンソールで、onepassword-audit-collector-hourly を見つけます。
[強制実行] をクリックして、ジョブを手動でトリガーします。
数秒待ちます。
Cloud Run > サービスに移動します。
[onepassword-audit-collector] をクリックします。
[Logs] タブをクリックします。

関数が正常に実行されたことを確認します。以下のものを探します。

--- Processing endpoint: auditevents ---
Page 1: Retrieved X events
Wrote X auditevents records to gs://onepassword-audit-secops-logs/onepassword-audit/auditevents_YYYYMMDD_HHMMSS.ndjson
Successfully processed X records

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

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

HTTP 401: OP_BEARER_TOKEN 環境変数のベアラートークンを確認します。トークンの有効期限が切れている可能性があります。
HTTP 429: レート制限。関数はバックオフで自動的に再試行します。1Password Events API では、1 分あたり 600 件のリクエストと 1 時間あたり 30,000 件のリクエストが許可されます。
環境変数が不足している: 必要な変数がすべて Cloud Run functions の構成で設定されていることを確認します。
レコードが返されない: イントロスペクションエンドポイントを使用して、ベアラートークンが監査イベントにアクセスできることを確認します。

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

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

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

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

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

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

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

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

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

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