ServiceNow の監査ログを収集する
このドキュメントでは、複数の方法を使用して ServiceNow 監査ログを Google Security Operations に取り込む方法について説明します。
オプション A: AWS S3 と Lambda
この方法では、AWS Lambda を使用して ServiceNow REST API を定期的にクエリし、監査ログを S3 バケットに保存します。Google Security Operations は、S3 バケットからログを収集します。
始める前に
- Google SecOps インスタンス
- ServiceNow テナントまたは API への特権アクセス
- AWS(S3、IAM、Lambda、EventBridge)への特権アクセス
ServiceNow の前提条件(ID、API キー、組織 ID、トークン)を収集する
- ServiceNow 管理コンソールにログインします。
- [System Security]> [Users and Groups] > [Users] に移動します。
- 監査ログにアクセスするための適切な権限を持つ新しいユーザーを作成するか、既存のユーザーを選択します。
- 次の詳細をコピーして安全な場所に保存します。
- ユーザー名
- パスワード
- インスタンス URL(例: https://instance.service-now.com)
Google SecOps 用に AWS S3 バケットと IAM を構成する
- バケットの作成のユーザーガイドに沿って、Amazon S3 バケットを作成します。
- 後で参照できるように、バケットの名前とリージョンを保存します(例:
servicenow-audit-logs)。 - IAM ユーザーの作成のユーザーガイドに沿って、ユーザーを作成します。
- 作成したユーザーを選択します。
- [セキュリティ認証情報] タブを選択します。
- [アクセスキー] セクションで [アクセスキーを作成] をクリックします。
- [ユースケース] として [サードパーティ サービス] を選択します。
- [次へ] をクリックします。
- 省略可: 説明タグを追加します。
- [アクセスキーを作成] をクリックします。
- [CSV ファイルをダウンロード] をクリックして、[アクセスキー] と [シークレット アクセスキー] を保存し、後で使用できるようにします。
- [完了] をクリックします。
- [権限] タブを選択します。
- [権限ポリシー] セクションで、[権限を追加] をクリックします。
- [権限を追加] を選択します。
- [ポリシーを直接アタッチする] を選択します。
- AmazonS3FullAccess ポリシーを検索して選択します。
- [次へ] をクリックします。
- [権限を追加] をクリックします。
S3 アップロードの IAM ポリシーとロールを構成する
- AWS コンソールで、[IAM] > [ポリシー] > [ポリシーの作成] > [JSON] タブに移動します。
以下のポリシーをコピーして貼り付けます。
{ "Version": "2012-10-17", "Statement": [ { "Sid": "AllowPutObjects", "Effect": "Allow", "Action": "s3:PutObject", "Resource": "arn:aws:s3:::servicenow-audit-logs/*" }, { "Sid": "AllowGetStateObject", "Effect": "Allow", "Action": "s3:GetObject", "Resource": "arn:aws:s3:::servicenow-audit-logs/audit-logs/state.json" } ] }- 別のバケット名を入力した場合は、
servicenow-audit-logsを置き換えます。
- 別のバケット名を入力した場合は、
[次へ] > [ポリシーを作成] をクリックします。
[IAM] > [ロール] > [ロールの作成] > [AWS サービス] > [Lambda] に移動します。
新しく作成したポリシーを関連付けます。
ロールに「
servicenow-audit-lambda-role」という名前を付けて、[ロールを作成] をクリックします。
Lambda 関数を作成する
- AWS コンソールで、[Lambda] > [Functions] > [Create function] に移動します。
- [Author from scratch] をクリックします。
次の構成情報を提供してください。
設定 値 名前 servicenow-audit-collectorランタイム Python 3.13 アーキテクチャ x86_64 実行ロール servicenow-audit-lambda-role関数を作成したら、[コード] タブを開き、スタブを削除して次のコード(
servicenow-audit-collector.py)を入力します。import urllib3 import json import os import datetime import boto3 import base64 def lambda_handler(event, context): # ServiceNow API details base_url = os.environ['API_BASE_URL'] # e.g., https://instance.service-now.com username = os.environ['API_USERNAME'] password = os.environ['API_PASSWORD'] # S3 details s3_bucket = os.environ['S3_BUCKET'] s3_prefix = os.environ['S3_PREFIX'] # State management state_key = os.environ.get('STATE_KEY', f"{s3_prefix}/state.json") # Pagination settings page_size = int(os.environ.get('PAGE_SIZE', '1000')) max_pages = int(os.environ.get('MAX_PAGES', '1000')) # Initialize S3 client s3 = boto3.client('s3') # Get last run timestamp from state file last_run_timestamp = get_last_run_timestamp(s3, s3_bucket, state_key) # Current timestamp for this run current_timestamp = datetime.datetime.now().isoformat() # Query ServiceNow API for audit logs with pagination audit_logs = get_audit_logs(base_url, username, password, last_run_timestamp, page_size, max_pages) if audit_logs: # Write logs to S3 in NDJSON format (newline-delimited JSON) timestamp = datetime.datetime.now().strftime('%Y-%m-%d-%H-%M-%S') s3_key = f"{s3_prefix}/servicenow-audit-{timestamp}.ndjson" # Format as NDJSON: one JSON object per line body = "\n".join(json.dumps(log) for log in audit_logs) + "\n" s3.put_object( Bucket=s3_bucket, Key=s3_key, Body=body, ContentType='application/x-ndjson' ) # Update state file update_state_file(s3, s3_bucket, state_key, current_timestamp) return { 'statusCode': 200, 'body': json.dumps(f'Successfully exported {len(audit_logs)} audit logs to S3') } else: return { 'statusCode': 200, 'body': json.dumps('No new audit logs to export') } def get_last_run_timestamp(s3, bucket, key): try: response = s3.get_object(Bucket=bucket, Key=key) state = json.loads(response['Body'].read().decode('utf-8')) return state.get('last_run_timestamp', '1970-01-01T00:00:00') except: return '1970-01-01T00:00:00' def update_state_file(s3, bucket, key, timestamp): state = {'last_run_timestamp': timestamp} s3.put_object( Bucket=bucket, Key=key, Body=json.dumps(state), ContentType='application/json' ) def get_audit_logs(base_url, username, password, last_run_timestamp, page_size=1000, max_pages=1000): """ Query ServiceNow sys_audit table with proper pagination. Uses sys_created_on field for timestamp filtering. """ # Encode credentials auth_string = f"{username}:{password}" auth_bytes = auth_string.encode('ascii') auth_encoded = base64.b64encode(auth_bytes).decode('ascii') # Setup HTTP client http = urllib3.PoolManager() headers = { 'Authorization': f'Basic {auth_encoded}', 'Accept': 'application/json' } results = [] offset = 0 for page in range(max_pages): # Build query with pagination # Use sys_created_on (not created_on) for timestamp filtering query_params = ( f"sysparm_query=sys_created_onAFTER{last_run_timestamp}" f"&sysparm_display_value=true" f"&sysparm_limit={page_size}" f"&sysparm_offset={offset}" ) url = f"{base_url}/api/now/table/sys_audit?{query_params}" try: response = http.request('GET', url, headers=headers) if response.status == 200: data = json.loads(response.data.decode('utf-8')) chunk = data.get('result', []) results.extend(chunk) # Stop if we got fewer records than page_size (last page) if len(chunk) < page_size: break # Move to next page offset += page_size else: print(f"Error querying ServiceNow API: {response.status} - {response.data.decode('utf-8')}") break except Exception as e: print(f"Exception querying ServiceNow API: {str(e)}") break return results[構成> 環境変数 > 編集 > 新しい環境変数を追加] に移動します。
次の環境変数を入力し、実際の値に置き換えます。
キー 値の例 S3_BUCKETservicenow-audit-logsS3_PREFIXaudit-logs/STATE_KEYaudit-logs/state.jsonAPI_BASE_URLhttps://instance.service-now.comAPI_USERNAME<your-username>API_PASSWORD<your-password>PAGE_SIZE1000MAX_PAGES1000関数が作成されたら、そのページにとどまるか、[Lambda] > [関数] > [servicenow-audit-collector] を開きます。
[CONFIGURATION] タブを選択します。
[全般設定] パネルで、[編集] をクリックします。
[Timeout] を [5 minutes (300 seconds)] に変更し、[Save] をクリックします。
EventBridge スケジュールを作成する
- [Amazon EventBridge] > [Scheduler] > [スケジュールの作成] に移動します。
- 次の構成の詳細を入力します。
- 定期的なスケジュール: レート(
1 hour)。 - ターゲット: Lambda 関数
servicenow-audit-collector。 - 名前:
servicenow-audit-collector-1h
- 定期的なスケジュール: レート(
- [スケジュールを作成] をクリックします。
ServiceNow の監査ログを取り込むように Google SecOps でフィードを構成する
- [SIEM 設定] > [フィード] に移動します。
- [+ 新しいフィードを追加] をクリックします。
- [フィード名] フィールドに、フィードの名前を入力します(例:
ServiceNow Audit logs)。 - [ソースタイプ] として [Amazon S3 V2] を選択します。
- [ログタイプ] として [ServiceNow Audit] を選択します。
- [次へ] をクリックします。
- 次の入力パラメータの値を指定します。
- S3 URI:
s3://servicenow-audit-logs/audit-logs/ - Source deletion options: 必要に応じて削除オプションを選択します。
- ファイルの最大経過日数: 指定した日数以内に変更されたファイルを含めます。デフォルトは 180 日です。
- アクセスキー ID: S3 バケットにアクセスできるユーザー アクセスキー。
- シークレット アクセスキー: S3 バケットにアクセスできるユーザーのシークレット キー。
- アセットの名前空間: アセットの名前空間。
- Ingestion labels: このフィードのイベントに適用されるラベル。
- S3 URI:
- [次へ] をクリックします。
- [Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。
オプション B: syslog を使用した Bindplane エージェント
この方法では、Bindplane エージェントを使用して ServiceNow 監査ログを収集し、Google Security Operations に転送します。ServiceNow は監査ログの syslog をネイティブにサポートしていないため、スクリプトを使用して ServiceNow REST API をクエリし、syslog 経由で Bindplane エージェントにログを転送します。
始める前に
次の前提条件を満たしていることを確認してください。
- Google SecOps インスタンス
- Windows 2016 以降、または
systemdを使用する Linux ホスト - プロキシの背後で実行している場合は、Bindplane エージェントの要件に従ってファイアウォール ポートが開いていることを確認します。
- ServiceNow 管理コンソールまたはアプライアンスへの特権アクセス
Google SecOps の取り込み認証ファイルを取得する
- Google SecOps コンソールにログインします。
- [SIEM 設定] > [コレクション エージェント] に移動します。
- Ingestion Authentication File をダウンロードします。Bindplane をインストールするシステムにファイルを安全に保存します。
Google SecOps のお客様 ID を取得する
- Google SecOps コンソールにログインします。
- [SIEM 設定] > [プロファイル] に移動します。
- [組織の詳細情報] セクションから [お客様 ID] をコピーして保存します。
Bindplane エージェントをインストールする
次の手順に沿って、Windows または Linux オペレーティング システムに Bindplane エージェントをインストールします。
Linux のインストール
- root 権限または sudo 権限でターミナルを開きます。
次のコマンドを実行します。
sudo sh -c "$(curl -fsSlL https://github.com/observiq/bindplane-agent/releases/latest/download/install_unix.sh)" install_unix.sh
その他のインストール リソース
- その他のインストール オプションについては、こちらのインストール ガイドをご覧ください。
Syslog を取り込んで Google SecOps に送信するように Bindplane エージェントを構成する
構成ファイルにアクセスします。
config.yamlファイルを見つけます。通常、Linux では/etc/bindplane-agent/ディレクトリに、Windows ではインストール ディレクトリにあります。- テキスト エディタ(
nano、vi、メモ帳など)を使用してファイルを開きます。
config.yamlファイルを次のように編集します。receivers: udplog: # Replace the port and IP address as required listen_address: "0.0.0.0:514" exporters: chronicle/chronicle_w_labels: compression: gzip # Adjust the path to the credentials file you downloaded in Step 1 creds_file_path: '/path/to/ingestion-authentication-file.json' # Replace with your actual customer ID from Step 2 customer_id: <YOUR_CUSTOMER_ID> # Replace with the appropriate regional endpoint endpoint: <CUSTOMER_REGION_ENDPOINT> # Add optional ingestion labels for better organization log_type: 'SERVICENOW_AUDIT' raw_log_field: body ingestion_labels: service: pipelines: logs/source0__chronicle_w_labels-0: receivers: - udplog exporters: - chronicle/chronicle_w_labels
- 自社のインフラストラクチャでの必要性に応じて、ポートと IP アドレスを置き換えます。
<YOUR_CUSTOMER_ID>は、実際の顧客 ID に置き換えます。<CUSTOMER_REGION_ENDPOINT>は、リージョン エンドポイントのドキュメントから適切なリージョン エンドポイントに置き換えます。/path/to/ingestion-authentication-file.jsonの値を、Google SecOps の取り込み認証ファイルを取得するで認証ファイルを保存したパスに更新します。
Bindplane エージェントを再起動して変更を適用する
Linux で Bindplane エージェントを再起動するには、次のコマンドを実行します。
sudo systemctl restart bindplane-agentWindows で Bindplane エージェントを再起動するには、サービス コンソールを使用するか、次のコマンドを入力します。
net stop BindPlaneAgent && net start BindPlaneAgent
ServiceNow 監査ログを syslog に転送するスクリプトを作成する
ServiceNow は監査ログの syslog をネイティブでサポートしていないため、ServiceNow REST API をクエリしてログを syslog に転送するスクリプトを作成します。このスクリプトは定期的に実行するようにスケジュールできます。
Python スクリプトの例(Linux)
次の内容で
servicenow_audit_to_syslog.pyという名前のファイルを作成します。import urllib3 import json import datetime import base64 import socket import time import os # ServiceNow API details BASE_URL = 'https://instance.service-now.com' # Replace with your ServiceNow instance URL USERNAME = 'admin' # Replace with your ServiceNow username PASSWORD = 'password' # Replace with your ServiceNow password # Syslog details SYSLOG_SERVER = '127.0.0.1' # Replace with your Bindplane agent IP SYSLOG_PORT = 514 # Replace with your Bindplane agent port # State file to keep track of last run STATE_FILE = '/tmp/servicenow_audit_last_run.txt' # Pagination settings PAGE_SIZE = 1000 MAX_PAGES = 1000 def get_last_run_timestamp(): try: with open(STATE_FILE, 'r') as f: return f.read().strip() except: return '1970-01-01T00:00:00' def update_state_file(timestamp): with open(STATE_FILE, 'w') as f: f.write(timestamp) def send_to_syslog(message): sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.sendto(message.encode(), (SYSLOG_SERVER, SYSLOG_PORT)) sock.close() def get_audit_logs(last_run_timestamp): """ Query ServiceNow sys_audit table with proper pagination. Uses sys_created_on field for timestamp filtering. """ # Encode credentials auth_string = f"{USERNAME}:{PASSWORD}" auth_bytes = auth_string.encode('ascii') auth_encoded = base64.b64encode(auth_bytes).decode('ascii') # Setup HTTP client http = urllib3.PoolManager() headers = { 'Authorization': f'Basic {auth_encoded}', 'Accept': 'application/json' } results = [] offset = 0 for page in range(MAX_PAGES): # Build query with pagination # Use sys_created_on (not created_on) for timestamp filtering query_params = ( f"sysparm_query=sys_created_onAFTER{last_run_timestamp}" f"&sysparm_display_value=true" f"&sysparm_limit={PAGE_SIZE}" f"&sysparm_offset={offset}" ) url = f"{BASE_URL}/api/now/table/sys_audit?{query_params}" try: response = http.request('GET', url, headers=headers) if response.status == 200: data = json.loads(response.data.decode('utf-8')) chunk = data.get('result', []) results.extend(chunk) # Stop if we got fewer records than PAGE_SIZE (last page) if len(chunk) < PAGE_SIZE: break # Move to next page offset += PAGE_SIZE else: print(f"Error querying ServiceNow API: {response.status} - {response.data.decode('utf-8')}") break except Exception as e: print(f"Exception querying ServiceNow API: {str(e)}") break return results def main(): # Get last run timestamp last_run_timestamp = get_last_run_timestamp() # Current timestamp for this run current_timestamp = datetime.datetime.now().isoformat() # Query ServiceNow API for audit logs audit_logs = get_audit_logs(last_run_timestamp) if audit_logs: # Send each log to syslog for log in audit_logs: # Format the log as JSON log_json = json.dumps(log) # Send to syslog send_to_syslog(log_json) # Sleep briefly to avoid flooding time.sleep(0.01) # Update state file update_state_file(current_timestamp) print(f"Successfully forwarded {len(audit_logs)} audit logs to syslog") else: print("No new audit logs to forward") if __name__ == "__main__": main()
スケジュールされた実行を設定する(Linux)
スクリプトを実行可能にします。
chmod +x servicenow_audit_to_syslog.py1 時間ごとにスクリプトを実行する cron ジョブを作成します。
crontab -e次の行を追加します。
0 * * * * /usr/bin/python3 /path/to/servicenow_audit_to_syslog.py >> /tmp/servicenow_audit_to_syslog.log 2>&1
PowerShell スクリプトの例(Windows)
次の内容で
ServiceNow-Audit-To-Syslog.ps1という名前のファイルを作成します。# ServiceNow API details $BaseUrl = 'https://instance.service-now.com' # Replace with your ServiceNow instance URL $Username = 'admin' # Replace with your ServiceNow username $Password = 'password' # Replace with your ServiceNow password # Syslog details $SyslogServer = '127.0.0.1' # Replace with your Bindplane agent IP $SyslogPort = 514 # Replace with your Bindplane agent port # State file to keep track of last run $StateFile = "$env:TEMP\ServiceNowAuditLastRun.txt" # Pagination settings $PageSize = 1000 $MaxPages = 1000 function Get-LastRunTimestamp { try { if (Test-Path $StateFile) { return Get-Content $StateFile } else { return '1970-01-01T00:00:00' } } catch { return '1970-01-01T00:00:00' } } function Update-StateFile { param ( [string]$Timestamp ) Set-Content -Path $StateFile -Value $Timestamp } function Send-ToSyslog { param ( [string]$Message ) $UdpClient = New-Object System.Net.Sockets.UdpClient $UdpClient.Connect($SyslogServer, $SyslogPort) $Encoding = [System.Text.Encoding]::ASCII $Bytes = $Encoding.GetBytes($Message) $UdpClient.Send($Bytes, $Bytes.Length) $UdpClient.Close() } function Get-AuditLogs { param ( [string]$LastRunTimestamp ) # Create auth header $Auth = [System.Convert]::ToBase64String([System.Text.Encoding]::ASCII.GetBytes("${Username}:${Password}")) $Headers = @{ Authorization = "Basic ${Auth}" Accept = 'application/json' } $Results = @() $Offset = 0 for ($page = 0; $page -lt $MaxPages; $page++) { # Build query with pagination # Use sys_created_on (not created_on) for timestamp filtering $QueryParams = "sysparm_query=sys_created_onAFTER${LastRunTimestamp}&sysparm_display_value=true&sysparm_limit=${PageSize}&sysparm_offset=${Offset}" $Url = "${BaseUrl}/api/now/table/sys_audit?${QueryParams}" try { $Response = Invoke-RestMethod -Uri $Url -Headers $Headers -Method Get $Chunk = $Response.result $Results += $Chunk # Stop if we got fewer records than PageSize (last page) if ($Chunk.Count -lt $PageSize) { break } # Move to next page $Offset += $PageSize } catch { Write-Error "Error querying ServiceNow API: $_" break } } return $Results } # Main execution $LastRunTimestamp = Get-LastRunTimestamp $CurrentTimestamp = (Get-Date).ToString('yyyy-MM-ddTHH:mm:ss') $AuditLogs = Get-AuditLogs -LastRunTimestamp $LastRunTimestamp if ($AuditLogs -and $AuditLogs.Count -gt 0) { # Send each log to syslog foreach ($Log in $AuditLogs) { # Format the log as JSON $LogJson = $Log | ConvertTo-Json -Compress # Send to syslog Send-ToSyslog -Message $LogJson # Sleep briefly to avoid flooding Start-Sleep -Milliseconds 10 } # Update state file Update-StateFile -Timestamp $CurrentTimestamp Write-Output "Successfully forwarded $($AuditLogs.Count) audit logs to syslog" } else { Write-Output "No new audit logs to forward" }
スケジュールされた実行を設定する(Windows)
- [タスク スケジューラ] を開きます。
- [タスクを作成] をクリックします。
- 次の構成を指定します。
- 名前: ServiceNowAuditToSyslog
- セキュリティ オプション: ユーザーがログインしているかどうかに関係なく実行する
- [トリガー] タブに移動します。
- [新規作成] をクリックし、1 時間ごとに実行するように設定します。
- [アクション] タブに移動します。
- [新規] をクリックして、次の項目を設定します。
- アクション: プログラムの開始
- プログラム/スクリプト: powershell.exe
- 引数: -ExecutionPolicy Bypass -File "C:\path\to\ServiceNow-Audit-To-Syslog.ps1"
- [OK] をクリックしてタスクを保存します。
さらにサポートが必要な場合 コミュニティ メンバーや Google SecOps のプロフェッショナルから回答を得ることができます。