Tines 監査ログを収集する

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

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

Tines は、セキュリティ チームがワークフローを構築してセキュリティ オペレーションを自動化できるノーコード自動化プラットフォームです。Tines 監査ログを使用すると、Tines プラットフォーム内のユーザー アクション、構成の変更、システム イベントを把握できます。

始める前に

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

  • Google SecOps インスタンス
  • Cloud Storage API が有効になっている GCP プロジェクト
  • GCS バケットを作成および管理する権限
  • GCS バケットの IAM ポリシーを管理する権限
  • Cloud Run サービス、Pub/Sub トピック、Cloud Scheduler ジョブを作成する権限
  • Tines への特権アクセス

Tines の URL を取得する

  1. ブラウザでテナントの Tines UI を開きます。
  2. アドレスバーからドメインをコピーします。これは TINES_BASE_URL として使用します。
  3. 形式: https://<tenant-domain>(例: https://<tenant-domain>.tines.com)。

後の手順で使用する値:

  • TINES_BASE_URL - 例: https://<domain>.tines.com
  • TINES_API_KEY - 次の手順で作成するトークン
  1. ナビゲーション メニュー > API キーに移動します。
  2. [+ 新しいキー] をクリックします。
  3. [Service API key] を選択します。
  4. わかりやすい名前(例: SecOps Audit Logs)を入力します。
  5. [作成] をクリックします。
  6. 生成されたトークンをすぐにコピーして安全に保存します。このトークンは TINES_API_KEY として使用します。

オプション 2 - 個人用 API キー(サービスキーが利用できない場合)

  1. ナビゲーション メニュー > API キーに移動します。
  2. [+ 新しいキー] をクリックします。
  3. [Personal API key] を選択します。
  4. わかりやすい名前を入力します。
  5. [作成] をクリックします。
  6. 生成されたトークンをコピーして安全に保存します。

監査ログの読み取り権限を付与する

  1. テナント オーナーとしてログインします(または、テナント オーナーに依頼してログインしてもらいます)。
  2. [設定> 管理 > ユーザー管理] に移動します(または、左上のメニューでチーム名をクリックして [ユーザー] を選択します)。
  3. Service API キーに関連付けられているサービス アカウント ユーザーを見つけます(API キーと同じ名前になります)。個人用 API キーを使用している場合は、ご自身のユーザー アカウントを探します。
  4. ユーザーをクリックしてプロフィールを開きます。
  5. [テナント権限] セクションで、AUDIT_LOG_READ を有効にします。
  6. [保存] をクリックします。

権限を確認する

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

  1. Tines にログインします。
  2. [設定> モニタリング> 監査ログ] に移動します。
  3. 監査ログエントリが表示される場合は、必要な権限があります。
  4. このオプションが表示されない場合は、テナント オーナーに連絡して AUDIT_LOG_READ 権限を付与してもらってください。

テスト API アクセス

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

    # Replace with your actual credentials
    TINES_BASE_URL="https://<tenant-domain>.tines.com"
    TINES_API_KEY="<your-api-key>"
    
    # Test API access
    curl -X GET "${TINES_BASE_URL}/api/v1/audit_logs?per_page=1" \
      -H "Authorization: Bearer ${TINES_API_KEY}" \
      -H "Content-Type: application/json"
    

監査ログエントリを含む JSON レスポンスが返されます。

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

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

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

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

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

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

  1. GCP Console で、[IAM と管理>サービス アカウント] に移動します。
  2. [サービス アカウントを作成] をクリックします。
  3. 次の構成の詳細を指定します。
    • サービス アカウント名: 「tines-audit-to-gcs-sa」と入力します。
    • サービス アカウントの説明: 「Service account for Cloud Run function to collect Tines Audit Logs」と入力します。
  4. [作成して続行] をクリックします。
  5. [このサービス アカウントにプロジェクトへのアクセスを許可する] セクションで、次のロールを追加します。
    1. [ロールを選択] をクリックします。
    2. [ストレージ オブジェクト管理者] を検索して選択します。
    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. バケット名をクリックします。
  3. [権限] タブに移動します。
  4. [アクセス権を付与] をクリックします。
  5. 次の構成の詳細を指定します。
    • プリンシパルを追加: サービス アカウントのメールアドレス(例: tines-audit-to-gcs-sa@PROJECT_ID.iam.gserviceaccount.com)を入力します。
    • ロールを割り当てる: [Storage オブジェクト管理者] を選択します。
  6. [保存] をクリックします。

Pub/Sub トピックの作成

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

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

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

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

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

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

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

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

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

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

    1. [変数とシークレット] をクリックします。
    2. 環境変数ごとに [+ 変数を追加] をクリックします。
    変数名 値の例 説明
    GCS_BUCKET tines-audit-logs GCS バケット名
    GCS_PREFIX tines/audit/ ログファイルの接頭辞
    STATE_KEY tines/audit/state.json 状態ファイルのパス
    TINES_BASE_URL https://your-tenant.tines.com API ベース URL
    TINES_API_KEY your-tines-api-key API キー
    LOOKBACK_SECONDS 3600 最初のルックバック期間
    PAGE_SIZE 500 1 ページあたりのレコード数
    MAX_PAGES 20 実行あたりの最大ページ数
    HTTP_TIMEOUT 60 HTTP リクエスト タイムアウト
    HTTP_RETRIES 3 HTTP 再試行回数
  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
    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', 'tines/audit/')
    STATE_KEY = os.environ.get('STATE_KEY', 'tines/audit/state.json')
    TINES_BASE_URL = os.environ.get('TINES_BASE_URL')
    TINES_API_KEY = os.environ.get('TINES_API_KEY')
    LOOKBACK_SECONDS = int(os.environ.get('LOOKBACK_SECONDS', '3600'))
    PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '500'))
    MAX_PAGES = int(os.environ.get('MAX_PAGES', '20'))
    HTTP_TIMEOUT = int(os.environ.get('HTTP_TIMEOUT', '60'))
    HTTP_RETRIES = int(os.environ.get('HTTP_RETRIES', '3'))
    
    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 Tines Audit Logs and write to GCS.
    
        Args:
            cloud_event: CloudEvent object containing Pub/Sub message
        """
    
        if not all([GCS_BUCKET, TINES_BASE_URL, TINES_API_KEY]):
            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(seconds=LOOKBACK_SECONDS)
    
            print(f"Fetching logs from {last_time.isoformat()} to {now.isoformat()}")
    
            # Fetch logs
            records, newest_event_time = fetch_logs(
                api_base=TINES_BASE_URL,
                api_key=TINES_API_KEY,
                start_time=last_time,
                page_size=PAGE_SIZE,
                max_pages=MAX_PAGES,
            )
    
            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, api_key: str, start_time: datetime, page_size: int, max_pages: int):
        """
        Fetch logs from Tines API with pagination and rate limiting.
    
        Args:
            api_base: API base URL
            api_key: Tines API key
            start_time: Start time for log query
            page_size: Number of records per page
            max_pages: Maximum pages to fetch
    
        Returns:
            Tuple of (records list, newest_event_time ISO string)
        """
        base_url = api_base.rstrip('/')
        endpoint = f"{base_url}/api/v1/audit_logs"
    
        headers = {
            'Authorization': f'Bearer {api_key}',
            'Accept': 'application/json',
            'Content-Type': 'application/json',
            'User-Agent': 'GoogleSecOps-TinesCollector/1.0'
        }
    
        records = []
        newest_time = None
        page_num = 1
        backoff = 1.0
    
        while page_num <= max_pages:
            # Build URL with query parameters
            params = {
                'page': page_num,
                'per_page': page_size
            }
            param_str = '&'.join([f"{k}={v}" for k, v in params.items()])
            url = f"{endpoint}?{param_str}"
    
            try:
                response = http.request('GET', url, headers=headers, timeout=HTTP_TIMEOUT)
    
                # 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'))
    
                # Extract results
                page_results = []
                if isinstance(data, dict):
                    page_results = data.get('audit_logs', [])
                elif isinstance(data, list):
                    page_results = data
    
                if not page_results:
                    print(f"No more results (empty page)")
                    break
    
                # Filter by start_time
                filtered_results = []
                for event in page_results:
                    try:
                        event_time = event.get('created_at')
                        if event_time:
                            event_dt = parse_datetime(event_time)
                            if event_dt >= start_time:
                                filtered_results.append(event)
                                # Track newest event time
                                if newest_time is None or event_dt > parse_datetime(newest_time):
                                    newest_time = event_time
                    except Exception as e:
                        print(f"Warning: Could not parse event time: {e}")
                        filtered_results.append(event)
    
                print(f"Page {page_num}: Retrieved {len(page_results)} events, {len(filtered_results)} after filtering")
                records.extend(filtered_results)
    
                # Check for more results
                if isinstance(data, dict):
                    meta = data.get('meta', {})
                    next_page = meta.get('next_page_number')
                    if not next_page:
                        print("No more pages (no next_page_number)")
                        break
                    page_num = next_page
                else:
                    # If response is a list, check if we got fewer results than requested
                    if len(page_results) < page_size:
                        print(f"Reached last page (size={len(page_results)} < limit={page_size})")
                        break
                    page_num += 1
    
            except Exception as e:
                print(f"Error fetching logs: {e}")
                if page_num <= HTTP_RETRIES:
                    print(f"Retrying... (attempt {page_num}/{HTTP_RETRIES})")
                    time.sleep(backoff)
                    backoff = min(backoff * 2, 30.0)
                    continue
                return [], None
    
        print(f"Retrieved {len(records)} total records from {page_num} pages")
        return records, newest_time
    
    • 2 つ目のファイル: 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 functions の関数をトリガーします。

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

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

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

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

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

統合をテストする

  1. Cloud Scheduler コンソールで、ジョブ(tines-audit-hourly)を見つけます。
  2. [強制実行] をクリックして、ジョブを手動でトリガーします。
  3. 数秒待ちます。
  4. Cloud Run > サービスに移動します。
  5. 関数名(tines-audit-to-gcs)をクリックします。
  6. [Logs] タブをクリックします。
  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://bucket-name/tines/audit/logs_YYYYMMDD_HHMMSS.ndjson
    Successfully processed X records
    
  8. [Cloud Storage] > [バケット] に移動します。

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

  10. tines/audit/ フォルダに移動します。

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

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

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

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

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

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

  1. [SIEM 設定] > [フィード] に移動します。
  2. [Add New Feed] をクリックします。
  3. [単一フィードを設定] をクリックします。
  4. [フィード名] フィールドに、フィードの名前を入力します(例: Tines Audit Logs)。
  5. [ソースタイプ] として [Google Cloud Storage V2] を選択します。
  6. [ログタイプ] として [Tines] を選択します。
  7. [サービス アカウントを取得する] をクリックします。一意のサービス アカウント メールアドレスが表示されます(例:)。

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

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

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

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

Tines Audit Logs を取り込むように Google SecOps でフィードを構成する

  1. [SIEM 設定] > [フィード] に移動します。
  2. [Add New Feed] をクリックします。
  3. [単一フィードを設定] をクリックします。
  4. [フィード名] フィールドに、フィードの名前を入力します(例: Tines Audit Logs)。
  5. [ソースタイプ] として [Google Cloud Storage V2] を選択します。
  6. [ログタイプ] として [Tines] を選択します。
  7. [次へ] をクリックします。
  8. 次の入力パラメータの値を指定します。

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

      gs://tines-audit-logs/tines/audit/
      
      • 次のように置き換えます。

        • tines-audit-logs: GCS バケット名。
        • tines/audit/: ログが保存される接頭辞/フォルダパス。
    • Source deletion option: 必要に応じて削除オプションを選択します。

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

    • ファイルの最大経過日数: 指定した日数以内に変更されたファイルを含めます。デフォルトは 180 日です。

    • アセットの名前空間: アセットの名前空間

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

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

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

ご不明な点がございましたら、コミュニティ メンバーや Google SecOps のプロフェッショナルから回答を得ることができます。