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

Cisco Application Centric Infrastructure（ACI）ログを収集する

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

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

Google SecOps SIEM

このドキュメントでは、Cisco Application Centric Infrastructure（ACI）ログを Google Security Operations に取り込む方法について説明します。パーサーは、まず Grok パターンを使用して、受信した Cisco ACI ログを syslog メッセージとして処理しようとします。syslog の解析に失敗すると、メッセージが JSON 形式であると想定し、それに応じて解析します。最後に、抽出されたフィールドを統合データモデル（UDM）にマッピングします。

この統合では、次の 2 つの方法がサポートされています。

オプション 1: Bindplane エージェントを使用した Syslog 形式
オプション 2: APIC REST API を使用した JSON 形式の Google Cloud Storage

各オプションは自己完結型であり、インフラストラクチャの要件とログ形式の設定に基づいて個別に実装できます。

オプション 1: Bindplane エージェントを使用した Syslog

このオプションでは、Cisco ACI ファブリックを構成して、syslog メッセージを Bindplane エージェントに送信します。Bindplane エージェントは、分析のためにメッセージを Google Security Operations に転送します。

始める前に

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

Google SecOps インスタンス
Windows 2016 以降、または systemd を使用する Linux ホスト
プロキシの背後で実行している場合は、Bindplane エージェントの要件に従ってファイアウォールポートが開いていることを確認します。
Cisco APIC コンソールへの特権アクセス

Bindplane エージェントの Google SecOps 認証情報を取得する

Bindplane エージェントを Google SecOps に対して認証するには、次のいずれかの方法を選択します。

新規デプロイ: Chronicle API を使用します（オプション 1）。
以前のデプロイ: 2027 年 7 月 20 日に Ingestion API が廃止されるまで、取り込み認証ファイル（オプション 2）を引き続き使用します。

オプション 1: Chronicle API のサービスアカウントを作成する（推奨）

Google Cloud コンソールにログインし、Google SecOps インスタンスにリンクされている Google Cloud プロジェクトを選択します。
[IAM と管理] > [サービスアカウント] に移動します。
[サービスアカウントを作成] をクリックします。
次の構成情報を指定します。
- サービスアカウント名: アカウントの名前を入力します（例: bindplane-ingest）。
- サービスアカウントの説明: 説明を入力します（例: Bindplane agent log ingestion）。
[作成して続行] をクリックします。
[このサービスアカウントにプロジェクトへのアクセスを許可する] セクションで、[ロールを選択] をクリックし、[Chronicle 編集者] を選択します。
[続行] をクリックして [完了] をクリックします。
作成したサービスアカウントをクリックし、[キー] タブに移動します。
[鍵を追加] > [新しい鍵を作成] の順にクリックします。
[JSON] を選択し、[作成] をクリックします。鍵ファイルがパソコンにダウンロードされます。
鍵ファイルを ingestion-auth.json として Bindplane コレクタディレクトリ（Linux の場合は /opt/observiq-otel-collector/ingestion-auth.json、Windows の場合は C:\Program Files\observIQ OpenTelemetry Collector\ingestion-auth.json）に保存するか、config.yaml の creds_file_path を更新して、保存先のパスと一致させます。
[IAM と管理] > [設定] に移動し、プロジェクト番号をコピーします。これはエクスポータの構成に必要です。

オプション 2: 取り込み認証ファイルをダウンロードする（レガシー）

Google SecOps コンソールにログインします。
[設定] > [データ管理] > [収集エージェント] に移動します。SIEM のみのプラットフォームを使用している場合は、[設定] > [収集エージェント] に移動します。
取り込み認証ファイルをダウンロードします。
Bindplane エージェントがインストールされるシステムにファイルを安全に保存します。

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

[Settings] > [Administration] > [Profile] に移動します。SIEM のみのプラットフォームを使用している場合は、[設定] > [プロファイル] に移動します。
[組織の詳細情報] セクションから [お客様 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
```
サービスが active (running) と表示されます。

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

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

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

構成ファイルにアクセスする

config.yaml ファイルを見つけます。通常、Linux の場合は /opt/observiq-otel-collector/ ディレクトリ、Windows の場合はインストールディレクトリにあります。
テキストエディタ（nano、vi、メモ帳など）を使用してファイルを開きます。

config.yaml ファイルを編集する

config.yaml ファイル

receivers:
  udplog:
    # Replace the port and IP address as required
    listen_address: "0.0.0.0:514"

exporters:
  chronicle/chronicle_w_labels:
    protocol: https
    # Region of your instance, for example us, europe, asia-southeast1
    location: <LOCATION>
    # Project number of the Google Cloud project linked to the instance
    project: '<PROJECT_NUMBER>'
    compression: gzip
    # Adjust the path to the credentials file you downloaded
    creds_file_path: '/path/to/ingestion-authentication-file.json'
    # Replace with your actual customer ID
    customer_id: <CUSTOMER_ID>
    endpoint: chronicle.googleapis.com
    # Add optional ingestion labels for better organization
    log_type: 'CISCO_ACI'
    raw_log_field: body
    ingestion_labels:
      service:

pipelines:
  logs/source0__chronicle_w_labels-0:
    receivers:
      - udplog
    exporters:
      - chronicle/chronicle_w_labels

構成パラメータ

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

listen_address: 自社のインフラストラクチャでの必要性に応じて、ポートと IP アドレスを置き換えます。0.0.0.0:514 を使用して、ポート 514 のすべてのインターフェースをリッスンします。
creds_file_path: サービスアカウントキーファイル（オプション 1）が保存されたパスに更新します。
- Linux: /opt/observiq-otel-collector/ingestion-auth.json
- Windows: C:\Program Files\observIQ OpenTelemetry Collector\ingestion-auth.json
customer_id: YOUR_CUSTOMER_ID は、前の手順で取得した実際のお客様 ID に置き換えます。
location: <LOCATION> は、Google SecOps インスタンスのリージョン（us、europe、asia-southeast1 など）に置き換えます。エクスポーターは、chronicle.googleapis.com の前にこの値を付加します。完全なリストについては、リージョンエンドポイントをご覧ください。
project: <PROJECT_NUMBER> を、[IAM と管理] > [設定] のプロジェクト番号に置き換えます。
endpoint: chronicle.googleapis.com に設定したままにします。

以前の Ingestion API エクスポータ

オプション 2 の取り込み認証ファイルを使用する場合は、exporters ブロックを次のように置き換えます。creds_file_path を取り込み認証ファイルのパスに設定します。インスタンスが米国以外にある場合は、malachiteingestion-pa.googleapis.com をリージョンエンドポイント（例: europe-malachiteingestion-pa.googleapis.com）に置き換えます。

exporters:
  chronicle/chronicle_w_labels:
    compression: gzip
    # Adjust the path to the credentials file you downloaded
    creds_file_path: '/path/to/ingestion-authentication-file.json'
    # Replace with your actual customer ID
    customer_id: <CUSTOMER_ID>
    endpoint: malachiteingestion-pa.googleapis.com
    # Add optional ingestion labels for better organization
    log_type: 'CISCO_ACI'
    raw_log_field: body
    ingestion_labels:
      service:

構成ファイルを保存する

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

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

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

Linux

Linux システムで Bindplane エージェントを再起動するには、次の操作を行います。

エージェントを再起動します。

sudo systemctl restart observiq-otel-collector

サービスのステータスを確認します。
```
sudo systemctl status observiq-otel-collector
```
エージェントログでエラーを追跡します。
```
sudo journalctl -u observiq-otel-collector -f
```

Windows

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

Cisco ACI で Syslog 転送を構成する

帯域外管理契約を構成する

Cisco APIC コンソールにログインします。
[Tenants] > [mgmt] > [Contracts] > [Filters] の順に移動します。
[フィルタを作成] をクリックします。
次の構成情報を指定します。
- 名前: 「syslog-udp-514」と入力します。
- エントリ名: 「syslog」と入力します。
- EtherType: [IP] を選択します。
- IP Protocol: [UDP] を選択します。
- 宛先ポート範囲の開始: 「514」と入力します。
- 宛先ポート範囲（終了）: 「514」と入力します。
[送信] をクリックします。

管理契約を作成

[Tenants] > [mgmt] > [Contracts] > [Standard] に移動します。
[Create Contract] をクリックします。
次の構成情報を指定します。
- 名前: 「mgmt-syslog-contract」と入力します。
- [スコープ]: [コンテキスト] を選択します。
[送信] をクリックします。
契約を展開し、[Subjects] をクリックします。
[Create Contract Subject] をクリックします。
次の構成情報を指定します。
- 名前: 「syslog-subject」と入力します。
- Apply Both Directions: このオプションをオンにします。
[送信] をクリックします。
件名を展開し、[フィルタ] をクリックします。
[Create Filter Binding] をクリックします。
syslog-udp-514 フィルタを選択します。
[送信] をクリックします。

Syslog 宛先グループを構成する

[Admin] > [External Data Collectors] > [Monitoring Destinations] > [Syslog] に移動します。
[Syslog] を右クリックし、[Create Syslog Monitoring Destination Group] を選択します。
次の構成情報を指定します。
- 名前: 「Chronicle-Syslog-Group」と入力します。
- 管理状態: [有効] を選択します。
- 形式: [aci] を選択します。
[次へ] をクリックします。
[Create Syslog Monitoring Destination] ダイアログで、次の操作を行います。
- 名前: 「Chronicle-BindPlane」と入力します。
- ホスト: Bindplane エージェントサーバーの IP アドレスを入力します。
- ポート: 「514」と入力します。
- 管理状態: [有効] を選択します。
- 重大度: [情報] を選択します（詳細なログをキャプチャするため）。
[送信] をクリックします。

モニタリングポリシーを構成する - ファブリックモニタリング

[Fabric] > [Fabric Policies] > [Policies] > [Monitoring] > [Common Policy] に移動します。
[Callhome/Smart Callhome/SNMP/Syslog/TACACS] を開きます。
[Syslog] を右クリックし、[Create Syslog Source] を選択します。
次の構成情報を指定します。
- 名前: 「Chronicle-Fabric-Syslog」と入力します。
- 監査ログ: 監査イベントを含める場合にオンにします。
- イベント: システムイベントを含める場合はオンにします。
- 障害: 障害イベントを含める場合はオンにします。
- セッションログ: セッションログを含める場合は、オンにします。
- 宛先グループ: Chronicle-Syslog-Group を選択します。
[送信] をクリックします。

モニタリングポリシーを構成する - アクセスモニタリング

[Fabric] > [Access Policies] > [Policies] > [Monitoring] > [Default Policy] の順に移動します。
[Callhome/Smart Callhome/SNMP/Syslog] を展開します。
[Syslog] を右クリックし、[Create Syslog Source] を選択します。
次の構成情報を指定します。
- 名前: 「Chronicle-Access-Syslog」と入力します。
- 監査ログ: 監査イベントを含める場合にオンにします。
- イベント: システムイベントを含める場合はオンにします。
- 障害: 障害イベントを含める場合はオンにします。
- セッションログ: セッションログを含める場合は、オンにします。
- 宛先グループ: Chronicle-Syslog-Group を選択します。
[送信] をクリックします。

システム Syslog メッセージポリシーを構成する

[Fabric] > [Fabric Policies] > [Policies] > [Monitoring] > [Common Policy] に移動します。
[Syslog Messages Policies] を開きます。
[default] をクリックします。
[Facility Filter] セクションで、次の操作を行います。
- ファシリティ: [デフォルト] を選択します。
- 最小重大度: [情報] に変更します。
[送信] をクリックします。

オプション 2: Google Cloud Storage を使用した JSON

このオプションでは、APIC REST API を使用して、Cisco ACI ファブリックから JSON 形式のイベント、障害、監査ログを収集し、Google SecOps の取り込み用に Google Cloud Storage に保存します。

始める前に

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

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

Cisco ACI APIC の前提条件を収集する

APIC 認証情報を取得する

HTTPS を使用して Cisco APIC コンソールにログインします。
[Admin] > [AAA]（APIC 6.0 以降）または [Admin] > [Authentication] > [AAA]（以前のリリース）に移動します。

注: AAA メニューパスは APIC 6.0(1) 以降で変更されました。
適切な権限を持つ既存のローカルユーザーを作成するか、使用します。
次の詳細情報をコピーして安全な場所に保存します。
- APIC ユーザー名: モニタリングデータへの読み取りアクセス権を持つローカルユーザー
- APIC パスワード: ユーザーのパスワード
- APIC URL: APIC の HTTPS URL（例: https://apic.example.com）
注: ユーザーアカウントには、Cisco のロールベースアクセス制御に従って、クエリ対象の特定のクラス（faultInst、eventRecord、aaaModLR）に対する読み取り権限が必要です。

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

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

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

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

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

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

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

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

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

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

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

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

サービスアカウント（cisco-aci-collector-sa）に GCS バケットへの書き込み権限を付与します。

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

Pub/Sub トピックの作成

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

GCP コンソールで、[Pub/Sub] > [トピック] に移動します。
[トピックを作成] をクリックします。
次の構成情報を指定します。
- トピック ID: 「cisco-aci-trigger」と入力します。
- その他の設定はデフォルトのままにします。
[作成] をクリックします。

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

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

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

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

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

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

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

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

変数名	値の例	説明
`GCS_BUCKET`	`cisco-aci-logs`	GCS バケット名
`GCS_PREFIX`	`cisco-aci-events`	ログファイルの接頭辞
`STATE_KEY`	`cisco-aci-events/state.json`	状態ファイルのパス
`APIC_URL`	`https://apic.example.com`	APIC HTTPS URL
`APIC_USERNAME`	`your-apic-username`	APIC ユーザー名
`APIC_PASSWORD`	`your-apic-password`	APIC パスワード
`PAGE_SIZE`	`100`	1 ページあたりのレコード数
`MAX_PAGES`	`10`	実行あたりの最大ページ数

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

# Configure logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Initialize HTTP client with timeouts
http = urllib3.PoolManager(
    timeout=urllib3.Timeout(connect=5.0, read=60.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', 'cisco-aci-events')
STATE_KEY = os.environ.get('STATE_KEY', 'cisco-aci-events/state.json')
APIC_URL = os.environ.get('APIC_URL')
APIC_USERNAME = os.environ.get('APIC_USERNAME')
APIC_PASSWORD = os.environ.get('APIC_PASSWORD')
PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '100'))
MAX_PAGES = int(os.environ.get('MAX_PAGES', '10'))

@functions_framework.cloud_event
def main(cloud_event):
    """
    Cloud Run function triggered by Pub/Sub to fetch Cisco ACI logs and write to GCS.

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

    if not all([GCS_BUCKET, APIC_URL, APIC_USERNAME, APIC_PASSWORD]):
        logger.error('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
        last_timestamp = state.get('last_timestamp')
        if not last_timestamp:
            last_timestamp = (datetime.utcnow() - timedelta(hours=1)).isoformat() + 'Z'

        logger.info(f"Starting Cisco ACI data collection for bucket: {GCS_BUCKET}")

        # Authenticate to APIC
        session_token = authenticate_apic(APIC_URL, APIC_USERNAME, APIC_PASSWORD)
        headers = {
            'Cookie': f'APIC-cookie={session_token}',
            'Accept': 'application/json',
            'Content-Type': 'application/json'
        }

        # Data types to collect
        data_types = ['faultInst', 'eventRecord', 'aaaModLR']
        all_collected_data = []

        for data_type in data_types:
            logger.info(f"Collecting {data_type} data")
            collected_data = collect_aci_data(
                APIC_URL,
                headers,
                data_type,
                last_timestamp,
                PAGE_SIZE,
                MAX_PAGES
            )

            # Tag each record with its type
            for record in collected_data:
                record['_data_type'] = data_type

            all_collected_data.extend(collected_data)
            logger.info(f"Collected {len(collected_data)} {data_type} records")

        logger.info(f"Total records collected: {len(all_collected_data)}")

        # Store data in GCS if any were collected
        if all_collected_data:
            timestamp_str = datetime.utcnow().strftime('%Y%m%d_%H%M%S')
            s3_key = f"{GCS_PREFIX}/cisco_aci_events_{timestamp_str}.ndjson"

            # Convert to NDJSON format (one JSON object per line)
            ndjson_content = '\n'.join(json.dumps(record) for record in all_collected_data)

            # Upload to GCS
            blob = bucket.blob(s3_key)
            blob.upload_from_string(
                ndjson_content,
                content_type='application/x-ndjson'
            )

            logger.info(f"Uploaded {len(all_collected_data)} records to gs://{GCS_BUCKET}/{s3_key}")

            # Update state file with latest timestamp from collected data
            latest_timestamp = get_latest_timestamp_from_records(all_collected_data)
            if not latest_timestamp:
                latest_timestamp = datetime.utcnow().isoformat() + 'Z'

            update_state(bucket, STATE_KEY, latest_timestamp)
        else:
            logger.info("No new log records found.")

        logger.info(f"Successfully processed {len(all_collected_data)} records")

    except Exception as e:
        logger.error(f'Error processing logs: {str(e)}')
        raise

def authenticate_apic(apic_url, username, password):
    """Authenticate to APIC and return session token"""
    login_url = f"{apic_url}/api/aaaLogin.json"
    login_data = {
        "aaaUser": {
            "attributes": {
                "name": username,
                "pwd": password
            }
        }
    }

    response = http.request(
        'POST',
        login_url,
        body=json.dumps(login_data).encode('utf-8'),
        headers={'Content-Type': 'application/json'},
        timeout=30
    )

    if response.status != 200:
        raise RuntimeError(f"APIC authentication failed: {response.status} {response.data[:256]!r}")

    response_data = json.loads(response.data.decode('utf-8'))
    token = response_data['imdata'][0]['aaaLogin']['attributes']['token']
    logger.info("Successfully authenticated to APIC")
    return token

def collect_aci_data(apic_url, headers, data_type, last_timestamp, page_size, max_pages):
    """Collect data from APIC REST API with pagination"""
    all_data = []
    page = 0

    while page < max_pages:
        # Build API URL with pagination and time filters
        api_url = f"{apic_url}/api/class/{data_type}.json"
        params = [
            f'page-size={page_size}',
            f'page={page}',
            f'order-by={data_type}.created|asc'
        ]

        # Add time filter to prevent duplicates
        if last_timestamp:
            params.append(f'query-target-filter=gt({data_type}.created,"{last_timestamp}")')

        full_url = f"{api_url}?{'&'.join(params)}"

        logger.info(f"Fetching {data_type} page {page} from APIC")

        # Make API request
        response = http.request('GET', full_url, headers=headers, timeout=60)

        if response.status != 200:
            logger.error(f"API request failed: {response.status} {response.data[:256]!r}")
            break

        data = json.loads(response.data.decode('utf-8'))
        records = data.get('imdata', [])

        if not records:
            logger.info(f"No more {data_type} records found")
            break

        # Extract the actual data from APIC format
        extracted_records = []
        for record in records:
            if data_type in record:
                extracted_records.append(record[data_type])

        all_data.extend(extracted_records)
        page += 1

        # If we got less than page_size records, we've reached the end
        if len(records) < page_size:
            break

    return all_data

def get_last_timestamp(bucket, state_key):
    """Get the last run timestamp from GCS state file"""
    try:
        blob = bucket.blob(state_key)
        if blob.exists():
            state_data = blob.download_as_text()
            state = json.loads(state_data)
            return state.get('last_timestamp')
    except Exception as e:
        logger.warning(f"Error reading state file: {str(e)}")

    return None

def get_latest_timestamp_from_records(records):
    """Get the latest timestamp from collected records to prevent missing events"""
    if not records:
        return None

    latest = None
    latest_time = None

    for record in records:
        try:
            # Handle both direct attributes and nested structure
            attrs = record.get('attributes', record)
            created = attrs.get('created')
            modTs = attrs.get('modTs')

            # Use created or modTs as fallback
            timestamp = created or modTs

            if timestamp:
                if latest_time is None or timestamp > latest_time:
                    latest_time = timestamp
                    latest = record
        except Exception as e:
            logger.debug(f"Error parsing timestamp from record: {e}")
            continue

    return latest_time

def update_state(bucket, state_key, timestamp):
    """Update the state file with the current timestamp"""
    try:
        state_data = {
            'last_timestamp': timestamp,
            'updated_at': datetime.utcnow().isoformat() + 'Z'
        }
        blob = bucket.blob(state_key)
        blob.upload_from_string(
            json.dumps(state_data),
            content_type='application/json'
        )
        logger.info(f"Updated state file with timestamp: {timestamp}")
    except Exception as e:
        logger.error(f"Error updating state file: {str(e)}")

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:
        logger.warning(f"Could not load state: {e}")

    return {}

2 つ目のファイル: requirements.txt:

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

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

注: Pub/Sub トリガー構成では、必要なサブスクリプションと権限が自動的に作成されます。APIC URL は、Cloud Run 関数のネットワーク環境から到達可能である必要があります。また、適切なファイアウォールルールで、関数から APIC への HTTPS（TCP/443）を許可する必要があります。

Cloud Scheduler ジョブの作成

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

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

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

設定	値
名前	`cisco-aci-collector-15m`
リージョン	Cloud Run functions の関数と同じリージョンを選択します。
周波数	`/15 * * *`（15 分ごと）
タイムゾーン	タイムゾーンを選択（UTC を推奨）
ターゲットタイプ	Pub/Sub
トピック	Pub/Sub トピック（`cisco-aci-trigger`）を選択します。
メッセージ本文	`{}`（空の JSON オブジェクト）

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

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

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

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

統合をテストする

Cloud Scheduler コンソールで、ジョブ（cisco-aci-collector-15m）を見つけます。
[強制実行] をクリックして、ジョブを手動でトリガーします。
数秒待ちます。
[Cloud Run] > [サービス] に移動します。
次の関数名（cisco-aci-collector）をクリックします。
[ログ] タブをクリックします。

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

Starting Cisco ACI data collection for bucket: cisco-aci-logs
Successfully authenticated to APIC
Collecting faultInst data
Collected X faultInst records
Collecting eventRecord data
Collected X eventRecord records
Collecting aaaModLR data
Collected X aaaModLR records
Total records collected: X
Uploaded X records to gs://cisco-aci-logs/cisco-aci-events/cisco_aci_events_YYYYMMDD_HHMMSS.ndjson
Successfully processed X records

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

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

HTTP 401: 環境変数で APIC 認証情報を確認します。
HTTP 403: APIC アカウントに faultInst、eventRecord、aaaModLR クラスに対する読み取り権限があることを確認します。
接続エラー: Cloud Run functions が TCP/443 で APIC URL にアクセスできることを確認します。
環境変数が設定されていない: 必要な変数がすべて設定されていることを確認します。

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

[設定] > [データ管理] > [フィード] に移動します。SIEM のみのプラットフォームを使用している場合は、[設定] > [フィード] に移動します。
[Add New Feed] をクリックします。
[単一フィードを設定] をクリックします。
[フィード名] フィールドに、フィードの名前を入力します（例: Cisco ACI JSON logs）。
[Source type] として [Google Cloud Storage V2] を選択します。
[Log type] として [Cisco Application Centric Infrastructure] を選択します。
[サービスアカウントを取得する] をクリックします。一意のサービスアカウントのメールアドレスが表示されます。例:
```
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
```
このメールアドレスをコピーします。次のステップでこれを使用します。

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

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

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

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

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

UDM マッピングテーブル

ログフィールド	UDM マッピング	ロジック
@timestamp	read_only_udm.metadata.event_timestamp	値は、未加工ログの「@timestamp」フィールドから取得され、タイムスタンプとして解析されます。
aci_tag	read_only_udm.metadata.product_log_id	値は、未加工ログのフィールド「aci_tag」から取得されます。
cisco_timestamp	-	マッピングされていません。
DIP	read_only_udm.target.ip	値は、未加工ログの「DIP」フィールドから取得されます。
DPort	read_only_udm.target.port	値は、未加工ログの「DPort」フィールドから取得され、整数に変換されます。
説明	read_only_udm.security_result.description	値は、未加工ログの「description」フィールドから取得されます。
fault_cause	read_only_udm.additional.fields.value.string_value	値は、未加工ログの「fault_cause」フィールドから取得されます。キーは「Fault Cause」に設定されます。
hostname	read_only_udm.principal.hostname	値は、未加工ログの「hostname」フィールドから取得されます。
lifecycle_state	read_only_udm.metadata.product_event_type	値は、未加工ログの「lifecycle_state」フィールドから取得されます。
log.source.address	-	マッピングされていません。
logstash.collect.host	-	マッピングされていません。
logstash.collect.timestamp	read_only_udm.metadata.collected_timestamp	値は、未加工ログのフィールド「logstash.collect.timestamp」から取得され、タイムスタンプとして解析されます。
logstash.ingest.host	read_only_udm.intermediary.hostname	値は、未加工ログのフィールド「logstash.ingest.host」から取得されます。
logstash.irm_environment	read_only_udm.additional.fields.value.string_value	値は、未加工ログのフィールド「logstash.irm_environment」から取得されます。キーは「IRM_Environment」に設定されます。
logstash.irm_region	read_only_udm.additional.fields.value.string_value	値は、未加工ログのフィールド「logstash.irm_region」から取得されます。キーは「IRM_Region」に設定されます。
logstash.irm_site	read_only_udm.additional.fields.value.string_value	値は、未加工ログの「logstash.irm_site」フィールドから取得されます。キーは「IRM_Site」に設定されます。
logstash.process.host	read_only_udm.intermediary.hostname	値は、未加工ログのフィールド「logstash.process.host」から取得されます。
メッセージ	-	マッピングされていません。
message_class	-	マッピングされていません。
message_code	-	マッピングされていません。
message_content	-	マッピングされていません。
message_dn	-	マッピングされていません。
message_type	read_only_udm.metadata.product_event_type	値は、角かっこを削除した後に、未加工ログの「message_type」フィールドから取得されます。
node_link	read_only_udm.principal.process.file.full_path	値は、未加工ログの「node_link」フィールドから取得されます。
PktLen	read_only_udm.network.received_bytes	値は、未加工ログフィールド「PktLen」から取得され、符号なし整数に変換されます。
プログラム	-	マッピングされていません。
.proto	read_only_udm.network.ip_protocol	値は、未加工ログのフィールド「Proto」から取得され、整数に変換され、対応する IP プロトコル名にマッピングされます（例: 6 -> TCP）。
SIP	read_only_udm.principal.ip	値は、未加工ログの「SIP」フィールドから取得されます。
SPort	read_only_udm.principal.port	値は、未加工ログの「SPort」フィールドから取得され、整数に変換されます。
syslog_facility	-	マッピングされていません。
syslog_facility_code	-	マッピングされていません。
syslog_host	read_only_udm.principal.ip, read_only_udm.observer.ip	値は、未加工ログの「syslog_host」フィールドから取得されます。
syslog_prog	-	マッピングされていません。
syslog_severity	read_only_udm.security_result.severity_details	値は、未加工ログの「syslog_severity」フィールドから取得されます。
syslog_severity_code	read_only_udm.security_result.severity	値は、未加工ログフィールド「syslog_severity_code」から取得され、対応する重大度レベルにマッピングされます。5、6、7 -> INFORMATIONAL、3、4 -> MEDIUM、0、1、2 -> HIGH。
syslog5424_pri	-	マッピングされていません。
Vlan-Id	read_only_udm.principal.resource.id	値は、未加工ログの「Vlan-Id」フィールドから取得されます。
-	read_only_udm.metadata.event_type	ロジック: 「SIP」または「hostname」が存在し、「Proto」が存在する場合は、「NETWORK_CONNECTION」に設定します。それ以外の場合、「SIP」、「hostname」、または「syslog_host」が存在する場合は「STATUS_UPDATE」に設定されます。それ以外の場合は「GENERIC_EVENT」に設定されます。
-	read_only_udm.metadata.log_type	ロジック: 「CISCO_ACI」に設定します。
-	read_only_udm.metadata.vendor_name	ロジック: 「Cisco」に設定します。
-	read_only_udm.metadata.product_name	ロジック: 「ACI」に設定します。
`principal_mac`	`event.idm.read_only_udm.principal.mac` と `event.idm.read_only_udm.principal.asset.mac`	変更履歴からマッピング
`context_id`	`event.idm.read_only_udm.additional.fields`	変更履歴からマッピング
`node_number`	`event.idm.read_only_udm.additional.fields`	変更履歴からマッピング
`code1`	`event.idm.read_only_udm.security_result.detection_fields`	変更履歴からマッピング
`code2`	`event.idm.read_only_udm.security_result.detection_fields`	変更履歴からマッピング
`pid`	`event.idm.read_only_udm.principal.process.pid`	変更履歴からマッピング
`src_intf`	`event.idm.read_only_udm.principal.resource.attribute.labels`	変更履歴からマッピング
`VXLAN`	`event.idm.read_only_udm.network.session_id`	変更履歴からマッピング
`VlanType`	`event.idm.read_only_udm.additional.fields`	変更履歴からマッピング
`flag`	`event.idm.read_only_udm.additional.fields`	変更履歴からマッピング
`Hitcnt`	`event.idm.read_only_udm.additional.fields`	変更履歴からマッピング
`syslog_facility`	`event.idm.read_only_udm.additional.fields`	変更履歴からマッピング
`SMac`	`event.idm.read_only_udm.principal.resource.attribute.labels`	変更履歴からマッピング
`DMac`	`event.idm.read_only_udm.target.resource.attribute.labels`	変更履歴からマッピング
`path`	`event.idm.read_only_udm.principal.process.file.full_path`	変更履歴からマッピング
`CName`	`event.idm.read_only_udm.security_result.rule_name`	変更履歴からマッピング
`i_ip`	`event.idm.read_only_udm.intermediary.ip`	変更履歴からマッピング
`i_ip`	`event.idm.read_only_udm.intermediary.asset.ip`	変更履歴からマッピング
`log_id`	`event.idm.read_only_udm.metadata.product_log_id`	変更履歴からマッピング
`product_event_type`	`event.idm.read_only_udm.metadata.product_event_type`	変更履歴からマッピング
`severity_action`	`event.idm.read_only_udm.security_result.severity`	変更履歴からマッピング
`session_id`	`event.idm.read_only_udm.network.session_id`	変更履歴からマッピング
`protocol`	`event.idm.read_only_udm.additional.fields`	変更履歴からマッピング
`message_code`	`event.idm.read_only_udm.additional.fields`	変更履歴からマッピング
`program`	`event.idm.read_only_udm.additional.fields`	変更履歴からマッピング
`syslog_prog`	`event.idm.read_only_udm.additional.fields`	変更履歴からマッピング
`message_dn`	`event.idm.read_only_udm.additional.fields`	変更履歴からマッピング
`message_class`	`event.idm.read_only_udm.additional.fields`	変更履歴からマッピング
`syslog_facility_code`	`event.idm.read_only_udm.additional.fields`	変更履歴からマッピング
`syslog5424_pri`	`event.idm.read_only_udm.additional.fields`	変更履歴からマッピング
`initial_code`	`event.idm.read_only_udm.additional.fields`	変更履歴からマッピング
`logstash.ingest.timestamp`	`event.idm.read_only_udm.additional.fields`	変更履歴からマッピング
`cisco_timestamp`	`event.idm.read_only_udm.additional.fields`	変更履歴からマッピング

変更ログ

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

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

Cisco Application Centric Infrastructure（ACI）ログを収集する

オプション 1: Bindplane エージェントを使用した Syslog

始める前に

Bindplane エージェントの Google SecOps 認証情報を取得する

オプション 1: Chronicle API のサービス アカウントを作成する（推奨）

オプション 2: 取り込み認証ファイルをダウンロードする（レガシー）

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

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

Windows へのインストール

Linux へのインストール

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

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

構成ファイルにアクセスする

config.yaml ファイルを編集する

構成パラメータ

以前の Ingestion API エクスポータ

構成ファイルを保存する

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

Linux

Windows

Cisco ACI で Syslog 転送を構成する

帯域外管理契約を構成する

管理契約を作成

Syslog 宛先グループを構成する

モニタリング ポリシーを構成する - ファブリック モニタリング

モニタリング ポリシーを構成する - アクセス モニタリング

システム Syslog メッセージ ポリシーを構成する

オプション 2: Google Cloud Storage を使用した JSON

始める前に

Cisco ACI APIC の前提条件を収集する

APIC 認証情報を取得する

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

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

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

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

Pub/Sub トピックの作成

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

関数コードを追加する

Cloud Scheduler ジョブの作成

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

統合をテストする

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

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

UDM マッピング テーブル

変更ログ

オプション 1: Chronicle API のサービスアカウントを作成する（推奨）

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

モニタリングポリシーを構成する - ファブリックモニタリング

モニタリングポリシーを構成する - アクセスモニタリング

システム Syslog メッセージポリシーを構成する

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

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

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

UDM マッピングテーブル