NetApp Console(旧称 BlueXP)の監査ログを収集する

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

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

NetApp Console は、オンプレミス環境とクラウド環境にまたがるハイブリッド マルチクラウド ストレージとデータ サービスを管理するための統合コントロール プレーンです。監査サービスは、Console サービスによって実行されたオペレーションを記録します。これには、送信元 IP アドレス、ワークスペース、使用された Console エージェント、フォレンジック分析とコンプライアンス要件に役立つその他のテレメトリー データが含まれます。

始める前に

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

  • Google SecOps インスタンス
  • Cloud Storage、Cloud Run、Pub/Sub、Cloud Scheduler API が有効になっている GCP プロジェクト
  • GCS バケットを作成、管理する権限
  • GCS バケットの IAM ポリシーを管理する権限
  • Cloud Run サービス、Pub/Sub トピック、Cloud Scheduler ジョブを作成する権限
  • サービス アカウントを作成する権限を持つ NetApp コンソールへの管理者アクセス権
  • NetApp Console アカウント ID

NetApp Console API アクセスを構成する

Google SecOps が監査ログを取得できるようにするには、適切な権限を持つサービス アカウントを作成し、API 認証情報を生成する必要があります。

アカウント ID を取得する

  1. ブラウザを使用して NetApp Console に移動します。
  2. NetApp Console の認証情報または NetApp Support Site の認証情報を使用してログインします。
  3. ページ上部の [アカウント] プルダウンをクリックします。
  4. 選択したアカウントの [アカウントを管理] をクリックします。
  5. [概要] セクションで、[アカウント ID] の値をコピーします。

サービス アカウントを作成する

  1. NetApp コンソールで、[Administration] > [Identity and access] に移動します。
  2. [メンバー] を選択します。
  3. [メンバーを追加] を選択します。
  4. [メンバータイプ] で [サービス アカウント] を選択します。
  5. サービス アカウントの名前を入力します(例: Google SecOps Integration)。
  6. クライアント シークレット認証を使用するには、[秘密鍵 JWT 認証を使用する] のチェックを外します。

  7. [組織、フォルダ、プロジェクトを選択] セクションで、組織を選択します。

  8. [カテゴリ] で、[組織] を選択します。

  9. [ロール] で、[組織閲覧者] を選択します。

  10. [追加] をクリックします。

API 認証情報を記録する

サービス アカウントを作成すると、認証情報を示すダイアログが表示されます。

  • クライアント ID: 固有のクライアント ID(例: TvPPs4SeM5smEElsGmdDUznljhN3YY8s)
  • Client Secret: API シークレット キー
  1. クライアント ID とクライアント シークレットの両方をダウンロードするか、安全な場所にコピーします。
  2. [閉じる] をクリックします。

権限を確認する

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

  1. NetApp コンソールにログインします。
  2. [管理] > [ID とアクセス] に移動します。
  3. [メンバー] を選択します。
  4. 作成したサービス アカウントを見つけて、組織閲覧者のロールが付与されていることを確認します。
  5. ロールが割り当てられていない場合は、サービス アカウント名をクリックして、ロールを組織閲覧者に更新します。

テスト API へのアクセス

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

    # Replace with your actual credentials
    CLIENT_ID="your-client-id"
    CLIENT_SECRET="your-client-secret"
    ACCOUNT_ID="your-account-id"
    
    # Obtain access token
    TOKEN=$(curl -s -X POST "https://netapp-cloud-account.auth0.com/oauth/token" \
        -H "Content-Type: application/json" \
        -d '{
            "grant_type": "client_credentials",
            "client_id": "'"${CLIENT_ID}"'",
            "client_secret": "'"${CLIENT_SECRET}"'",
            "audience": "https://api.cloud.netapp.com"
        }' | python3 -c "import sys,json; print(json.load(sys.stdin)['access_token'])")
    
    # Test audit API access
    curl -s -H "Authorization: Bearer ${TOKEN}" \
        "https://cloudmanager.cloud.netapp.com/audit/${ACCOUNT_ID}?offset=0" \
        | python3 -m json.tool
    

成功すると、指定したアカウントの監査レコードを含む JSON オブジェクトが返されます。

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

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

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

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

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

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

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

Pub/Sub トピックの作成

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

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

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

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

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

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

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

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

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

    1. [変数とシークレット] をクリックします。
    2. 環境変数ごとに [+ 変数を追加] をクリックします。
    変数名 値の例 説明
    GCS_BUCKET netapp-bluexp-audit-logs GCS バケット名
    GCS_PREFIX netapp-bluexp-audit ログファイルの接頭辞
    STATE_KEY netapp-bluexp-audit/state.json 状態ファイルのパス
    NETAPP_CLIENT_ID your-client-id NetApp Console サービス アカウントのクライアント ID
    NETAPP_CLIENT_SECRET your-client-secret NetApp Console サービス アカウントのクライアント シークレット
    NETAPP_ACCOUNT_ID account-AbCdEfGh NetApp Console アカウント ID
    MAX_RECORDS 5000 実行あたりの最大レコード数
    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
    import time
    
    http = urllib3.PoolManager(
      timeout=urllib3.Timeout(connect=5.0, read=30.0),
      retries=False,
    )
    
    storage_client = storage.Client()
    
    GCS_BUCKET = os.environ.get('GCS_BUCKET')
    GCS_PREFIX = os.environ.get('GCS_PREFIX', 'netapp-bluexp-audit')
    STATE_KEY = os.environ.get('STATE_KEY', 'netapp-bluexp-audit/state.json')
    NETAPP_CLIENT_ID = os.environ.get('NETAPP_CLIENT_ID')
    NETAPP_CLIENT_SECRET = os.environ.get('NETAPP_CLIENT_SECRET')
    NETAPP_ACCOUNT_ID = os.environ.get('NETAPP_ACCOUNT_ID')
    MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '5000'))
    LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
    
    TOKEN_URL = 'https://netapp-cloud-account.auth0.com/oauth/token'
    API_BASE = 'https://cloudmanager.cloud.netapp.com'
    AUDIENCE = 'https://api.cloud.netapp.com'
    
    def to_unix_millis(dt):
      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):
      if value.endswith('Z'):
        value = value[:-1] + '+00:00'
      return datetime.fromisoformat(value)
    
    def get_access_token():
      body = json.dumps({
        'grant_type': 'client_credentials',
        'client_id': NETAPP_CLIENT_ID,
        'client_secret': NETAPP_CLIENT_SECRET,
        'audience': AUDIENCE,
      }).encode('utf-8')
    
      response = http.request(
        'POST', TOKEN_URL,
        body=body,
        headers={'Content-Type': 'application/json'}
      )
    
      if response.status != 200:
        raise Exception(
          f"Token request failed: {response.status} - "
          f"{response.data.decode('utf-8')}"
        )
    
      data = json.loads(response.data.decode('utf-8'))
      token = data.get('access_token')
      if not token:
        raise Exception('No access_token in token response')
    
      print('Successfully obtained NetApp Console access token')
      return token
    
    @functions_framework.cloud_event
    def main(cloud_event):
      if not all([GCS_BUCKET, NETAPP_CLIENT_ID, NETAPP_CLIENT_SECRET, NETAPP_ACCOUNT_ID]):
        print('Error: Missing required environment variables')
        return
    
      try:
        bucket = storage_client.bucket(GCS_BUCKET)
        state = load_state(bucket)
        now = datetime.now(timezone.utc)
    
        if isinstance(state, dict) and state.get('last_event_time'):
          try:
            last_val = state['last_event_time']
            if last_val.endswith('Z'):
              last_val = last_val[:-1] + '+00:00'
            last_time = datetime.fromisoformat(last_val)
            last_time = last_time - timedelta(minutes=2)
          except Exception as e:
            print(f'Warning: Could not parse last_event_time: {e}')
            last_time = now - timedelta(hours=LOOKBACK_HOURS)
        else:
          last_time = now - timedelta(hours=LOOKBACK_HOURS)
    
        print(f'Fetching audit logs from {last_time.isoformat()} to {now.isoformat()}')
    
        token = get_access_token()
    
        records, newest_time = fetch_audit_logs(token, last_time, now)
    
        if not records:
          print('No new audit records found.')
          save_state(bucket, now.isoformat())
          return
    
        timestamp = now.strftime('%Y%m%d_%H%M%S')
        object_key = f'{GCS_PREFIX}/netapp_bluexp_audit_{timestamp}.ndjson'
        blob = bucket.blob(object_key)
    
        ndjson = '\n'.join(
          [json.dumps(r, ensure_ascii=False, default=str) for r 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}')
    
        save_state(bucket, newest_time if newest_time else now.isoformat())
    
        print(f'Successfully processed {len(records)} audit records')
    
      except Exception as e:
        print(f'Error processing audit logs: {str(e)}')
        raise
    
    def fetch_audit_logs(token, start_time, end_time):
      endpoint = f'{API_BASE}/audit/{NETAPP_ACCOUNT_ID}'
    
      headers = {
        'Authorization': f'Bearer {token}',
        'Accept': 'application/json',
        'User-Agent': 'GoogleSecOps-NetAppBlueXPCollector/1.0',
      }
    
      start_millis = to_unix_millis(start_time)
      end_millis = to_unix_millis(end_time)
    
      records = []
      newest_time = None
      offset = 0
      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
    
        url = (
          f'{endpoint}'
          f'?fromLastModified={start_millis}'
          f'&toLastModified={end_millis}'
          f'&offset={offset}'
        )
    
        try:
          response = http.request('GET', url, headers=headers)
    
          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('auditEntries', [])
    
          if not page_results:
            print('No more results (empty page)')
            break
    
          print(f'Page {page_num}: Retrieved {len(page_results)} audit records')
          records.extend(page_results)
    
          for entry in page_results:
            try:
              last_modified = entry.get('lastModified')
              if last_modified:
                entry_dt = datetime.fromtimestamp(
                  last_modified / 1000, tz=timezone.utc
                )
                entry_time = entry_dt.isoformat()
                if newest_time is None or parse_datetime(entry_time) > parse_datetime(newest_time):
                  newest_time = entry_time
            except Exception as e:
              print(f'Warning: Could not parse entry time: {e}')
    
          count = data.get('count', len(page_results))
          if count < 100:
            print(f'Reached last page (count={count})')
            break
    
          offset += count
    
        except Exception as e:
          print(f'Error fetching audit logs: {e}')
          return [], None
    
      print(f'Retrieved {len(records)} total audit records from {page_num} pages')
      return records, newest_time
    
    def load_state(bucket):
      try:
        blob = bucket.blob(STATE_KEY)
        if blob.exists():
          return json.loads(blob.download_as_text())
      except Exception as e:
        print(f'Warning: Could not load state: {e}')
      return {}
    
    def save_state(bucket, last_event_time_iso):
      try:
        state = {
          'last_event_time': last_event_time_iso,
          'last_run': datetime.now(timezone.utc).isoformat(),
        }
        blob = bucket.blob(STATE_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}')
    
    • 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. 次の構成情報を提供してください。

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

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

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

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

統合をテストする

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

    Fetching audit logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
    Successfully obtained NetApp Console access token
    Page 1: Retrieved X audit records
    Wrote X records to gs://netapp-bluexp-audit-logs/netapp-bluexp-audit/netapp_bluexp_audit_YYYYMMDD_HHMMSS.ndjson
    Successfully processed X audit records
    
  8. [Cloud Storage] > [バケット] に移動します。

  9. netapp-bluexp-audit-logs をクリックします。

  10. netapp-bluexp-audit/ フォルダに移動します。

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

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

  • HTTP 401: NETAPP_CLIENT_ID 環境変数と NETAPP_CLIENT_SECRET 環境変数が正しいことを確認する
  • HTTP 403: NetApp Console でサービス アカウントに組織閲覧者ロールがあることを確認します。
  • HTTP 429: レート制限 - 関数は指数バックオフで自動的に再試行されます
  • 環境変数がありません: 必要な変数がすべて Cloud Run functions の構成で設定されていることを確認します。

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

UDM マッピング テーブル

ログフィールド UDM マッピング ロジック
endTime additional.fields 空でない場合は、各ラベルと統合されます
responseData additional.fields
weHash additional.fields
wePublicId additional.fields
hasFailedRecord additional.fields
アクション additional.fields
サービス additional.fields
hasRecords additional.fields
workspaceId additional.fields
lastModified additional.fields
startTime additional.fields
agentName metadata.event_type agentName が空でない場合は STATUS_UPDATE、それ以外の場合は GENERIC_EVENT に設定
agentId metadata.product_deployment_id 空でない場合、値が直接コピーされます
requestId metadata.product_log_id 空でない場合、値が直接コピーされます。
network network 空でない場合、直接名前が変更されます
agentName principal.asset.hostname 空でない場合、値が直接コピーされます
fileName principal.file.names 空でない場合はマージされます
agentName principal.hostname 空でない場合、値が直接コピーされます。
accountId principal.resource.id 空でない場合、値が直接コピーされます。
resourceName principal.resource.name 空でない場合、値が直接コピーされます
accountId principal.resource.product_object_id accountId が空でない場合はその値、それ以外の場合は resourceId が空でない場合はその値
resourceId principal.resource.product_object_id
principalId principal.user.userid 空でない場合、値が直接コピーされます
ステータス security_result.action ステータスが (?i)SUCCESS の場合は ALLOW、FAILURE または UNSUCCESSFUL_ATTEMPT の場合は BLOCK に設定します。
ステータス security_result.action_details 空でない場合、値が直接コピーされます。
ターゲット ターゲット 空でない場合、直接名前が変更されます

変更ログ

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

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