收集 ServiceNow 稽核記錄
本文說明如何使用多種方法,將 ServiceNow 稽核記錄擷取至 Google Security Operations。
選項 A:搭配 Lambda 的 AWS S3
這個方法會使用 AWS Lambda 定期查詢 ServiceNow REST API 的稽核記錄,並將記錄儲存在 S3 值區中。接著,Google Security Operations 會從 S3 bucket 收集記錄。
事前準備
- Google SecOps 執行個體
- ServiceNow 租戶或 API 的特殊權限
- AWS 的特殊存取權 (S3、IAM、Lambda、EventBridge)
收集 ServiceNow 必要條件 (ID、API 金鑰、機構 ID、權杖)
- 登入 ServiceNow 管理控制台。
- 依序前往「System Security」(系統安全性)「Users and Groups」(使用者和群組)「Users」(使用者)。
- 建立新使用者,或選取具備適當權限的現有使用者,以存取稽核記錄。
- 複製下列詳細資料並儲存在安全位置:
- 使用者名稱
- 密碼
- 執行個體網址 (例如 https://instance.service-now.com)
為 Google SecOps 設定 AWS S3 值區和 IAM
- 按照這份使用者指南建立 Amazon S3 bucket:建立 bucket
- 儲存 bucket 的「名稱」和「地區」,以供日後參考 (例如
servicenow-audit-logs)。 - 請按照這份使用者指南建立使用者:建立 IAM 使用者。
- 選取建立的「使用者」。
- 選取「安全憑證」分頁標籤。
- 在「Access Keys」部分中,按一下「Create Access Key」。
- 選取「第三方服務」做為「用途」。
- 點選「下一步」。
- 選用:新增說明標記。
- 按一下「建立存取金鑰」。
- 按一下「下載 CSV 檔案」,儲存「存取金鑰」和「私密存取金鑰」以供日後使用。
- 按一下 [完成]。
- 選取 [權限] 分頁標籤。
- 在「Permissions policies」(權限政策) 區段中,按一下「Add permissions」(新增權限)。
- 選取「新增權限」。
- 選取「直接附加政策」。
- 搜尋並選取 AmazonS3FullAccess 政策。
- 點選「下一步」。
- 按一下「Add permissions」。
設定 S3 上傳的身分與存取權管理政策和角色
- 在 AWS 控制台中,依序前往「IAM」>「Policies」>「Create policy」>「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" } ] }- 如果您輸入其他 bucket 名稱,請替換
servicenow-audit-logs。
- 如果您輸入其他 bucket 名稱,請替換
依序點選「下一步」>「建立政策」。
依序前往「IAM」>「Roles」>「Create role」>「AWS service」>「Lambda」。
附加新建立的政策。
為角色命名
servicenow-audit-lambda-role,然後按一下「建立角色」。
建立 Lambda 函式
- 在 AWS 控制台中,依序前往「Lambda」>「Functions」>「Create function」。
- 按一下「從頭開始撰寫」。
請提供下列設定詳細資料:
設定 值 名稱 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依序前往「Configuration」>「Environment variables」>「Edit」>「Add new environment variable」。
輸入下列環境變數,並將 換成您的值。
鍵 範例值 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」>「Functions」>「servicenow-audit-collector」)。
選取「設定」分頁標籤。
在「一般設定」面板中,按一下「編輯」。
將「Timeout」(逾時間隔) 變更為「5 minutes (300 seconds)」(5 分鐘 (300 秒)),然後按一下「Save」(儲存)。
建立 EventBridge 排程
- 依序前往「Amazon EventBridge」>「Scheduler」>「Create schedule」。
- 提供下列設定詳細資料:
- 週期性時間表:費率 (
1 hour)。 - 目標:您的 Lambda 函式
servicenow-audit-collector。 - 名稱:
servicenow-audit-collector-1h。
- 週期性時間表:費率 (
- 按一下「建立時間表」。
在 Google SecOps 中設定動態饋給,擷取 ServiceNow 稽核記錄
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「+ 新增動態消息」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
ServiceNow Audit logs)。 - 選取「Amazon S3 V2」做為「來源類型」。
- 選取「ServiceNow Audit」做為「記錄類型」。
- 點選「下一步」。
- 指定下列輸入參數的值:
- S3 URI:
s3://servicenow-audit-logs/audit-logs/ - 來源刪除選項:根據偏好設定選取刪除選項。
- 檔案存在時間上限:包含在過去天數內修改的檔案。預設值為 180 天。
- 存取金鑰 ID:具有 S3 儲存空間存取權的使用者存取金鑰。
- 存取密鑰:具有 S3 bucket 存取權的使用者私密金鑰。
- 資產命名空間:資產命名空間。
- 擷取標籤:套用至這個動態饋給事件的標籤。
- S3 URI:
- 點選「下一步」。
- 在「完成」畫面中檢查新的動態饋給設定,然後按一下「提交」。
選項 B:使用 syslog 的 Bindplane 代理程式
這個方法會使用 Bindplane 代理程式收集 ServiceNow 稽核記錄,並轉送至 Google Security Operations。由於 ServiceNow 原生不支援稽核記錄的系統記錄,我們會使用指令碼查詢 ServiceNow REST API,並透過系統記錄將記錄轉送至 Bindplane 代理程式。
事前準備
請確認您已完成下列事前準備事項:
- Google SecOps 執行個體
- 搭載
systemd的 Windows 2016 以上版本或 Linux 主機 - 如果透過 Proxy 執行,請確保防火牆通訊埠已根據 Bindplane 代理程式需求開啟
- ServiceNow 管理控制台或設備的特殊存取權
取得 Google SecOps 擷取驗證檔案
- 登入 Google SecOps 控制台。
- 依序前往「SIEM 設定」>「收集代理程式」。
- 下載擷取驗證檔案。將檔案安全地儲存在要安裝 Bindplane 的系統上。
取得 Google SecOps 客戶 ID
- 登入 Google SecOps 控制台。
- 依序前往「SIEM 設定」>「設定檔」。
- 複製並儲存「機構詳細資料」專區中的客戶 ID。
安裝 Bindplane 代理程式
請按照下列操作說明,在 Windows 或 Linux 作業系統上安裝 Bindplane 代理程式。
Linux 安裝
- 開啟具備根層級或 sudo 權限的終端機。
執行下列指令:
sudo sh -c "$(curl -fsSlL https://github.com/observiq/bindplane-agent/releases/latest/download/install_unix.sh)" install_unix.sh
其他安裝資源
- 如需其他安裝選項,請參閱這份安裝指南。
設定 Bindplane 代理程式,擷取系統記錄檔並傳送至 Google SecOps
存取設定檔:
- 找出
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-agent如要在 Windows 中重新啟動 Bindplane 代理程式,可以使用「服務」控制台,或輸入下列指令:
net stop BindPlaneAgent && net start BindPlaneAgent
建立指令碼,將 ServiceNow 稽核記錄轉送至 syslog
由於 ServiceNow 原生不支援稽核記錄的系統記錄檔,因此我們會建立指令碼,查詢 ServiceNow REST API 並將記錄轉送至系統記錄檔。您可以排定定期執行這項指令碼。
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.py建立 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
- 安全性選項:無論使用者是否登入,都會執行
- 前往「觸發條件」分頁。
- 按一下「新增」,然後將執行頻率設為每小時。
- 前往「動作」分頁。
- 按一下「新增」,然後設定:
- 動作:啟動程式
- 「程式/指令碼」:powershell.exe
- 引數:-ExecutionPolicy Bypass -File "C:\path\to\ServiceNow-Audit-To-Syslog.ps1"
- 按一下「確定」儲存工作。
還有其他問題嗎?向社群成員和 Google SecOps 專業人員尋求答案。