收集 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、權杖)

  1. 登入 ServiceNow 管理控制台
  2. 依序前往「System Security」(系統安全性)「Users and Groups」(使用者和群組)「Users」(使用者)。
  3. 建立新使用者,或選取具備適當權限的現有使用者,以存取稽核記錄。
  4. 複製下列詳細資料並儲存在安全位置:
    • 使用者名稱
    • 密碼
    • 執行個體網址 (例如 https://instance.service-now.com)

為 Google SecOps 設定 AWS S3 值區和 IAM

  1. 按照這份使用者指南建立 Amazon S3 bucket建立 bucket
  2. 儲存 bucket 的「名稱」和「地區」,以供日後參考 (例如 servicenow-audit-logs)。
  3. 請按照這份使用者指南建立使用者建立 IAM 使用者
  4. 選取建立的「使用者」
  5. 選取「安全憑證」分頁標籤。
  6. 在「Access Keys」部分中,按一下「Create Access Key」
  7. 選取「第三方服務」做為「用途」
  8. 點選「下一步」
  9. 選用:新增說明標記。
  10. 按一下「建立存取金鑰」
  11. 按一下「下載 CSV 檔案」,儲存「存取金鑰」和「私密存取金鑰」以供日後使用。
  12. 按一下 [完成]
  13. 選取 [權限] 分頁標籤。
  14. 在「Permissions policies」(權限政策) 區段中,按一下「Add permissions」(新增權限)
  15. 選取「新增權限」
  16. 選取「直接附加政策」
  17. 搜尋並選取 AmazonS3FullAccess 政策。
  18. 點選「下一步」
  19. 按一下「Add permissions」。

設定 S3 上傳的身分與存取權管理政策和角色

  1. 在 AWS 控制台中,依序前往「IAM」>「Policies」>「Create policy」>「JSON」分頁標籤
  2. 複製並貼上下列政策。

    {
      "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
  3. 依序點選「下一步」>「建立政策」

  4. 依序前往「IAM」>「Roles」>「Create role」>「AWS service」>「Lambda」

  5. 附加新建立的政策。

  6. 為角色命名 servicenow-audit-lambda-role,然後按一下「建立角色」

建立 Lambda 函式

  1. AWS 控制台中,依序前往「Lambda」>「Functions」>「Create function」
  2. 按一下「從頭開始撰寫」
  3. 請提供下列設定詳細資料:

    設定
    名稱 servicenow-audit-collector
    執行階段 Python 3.13
    架構 x86_64
    執行角色 servicenow-audit-lambda-role
  4. 建立函式後,開啟「程式碼」分頁,刪除存根並輸入下列程式碼 (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
    
  5. 依序前往「Configuration」>「Environment variables」>「Edit」>「Add new environment variable」

  6. 輸入下列環境變數,並將 換成您的值。

    範例值
    S3_BUCKET servicenow-audit-logs
    S3_PREFIX audit-logs/
    STATE_KEY audit-logs/state.json
    API_BASE_URL https://instance.service-now.com
    API_USERNAME <your-username>
    API_PASSWORD <your-password>
    PAGE_SIZE 1000
    MAX_PAGES 1000
  7. 建立函式後,請留在函式頁面 (或依序開啟「Lambda」>「Functions」>「servicenow-audit-collector」)。

  8. 選取「設定」分頁標籤。

  9. 在「一般設定」面板中,按一下「編輯」

  10. 將「Timeout」(逾時間隔) 變更為「5 minutes (300 seconds)」(5 分鐘 (300 秒)),然後按一下「Save」(儲存)

建立 EventBridge 排程

  1. 依序前往「Amazon EventBridge」>「Scheduler」>「Create schedule」
  2. 提供下列設定詳細資料:
    • 週期性時間表費率 (1 hour)。
    • 目標:您的 Lambda 函式 servicenow-audit-collector
    • 名稱servicenow-audit-collector-1h
  3. 按一下「建立時間表」

在 Google SecOps 中設定動態饋給,擷取 ServiceNow 稽核記錄

  1. 依序前往「SIEM 設定」>「動態饋給」
  2. 按一下「+ 新增動態消息」
  3. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 ServiceNow Audit logs)。
  4. 選取「Amazon S3 V2」做為「來源類型」
  5. 選取「ServiceNow Audit」做為「記錄類型」
  6. 點選「下一步」
  7. 指定下列輸入參數的值:
    • S3 URIs3://servicenow-audit-logs/audit-logs/
    • 來源刪除選項:根據偏好設定選取刪除選項。
    • 檔案存在時間上限:包含在過去天數內修改的檔案。預設值為 180 天。
    • 存取金鑰 ID:具有 S3 儲存空間存取權的使用者存取金鑰。
    • 存取密鑰:具有 S3 bucket 存取權的使用者私密金鑰。
    • 資產命名空間資產命名空間
    • 擷取標籤:套用至這個動態饋給事件的標籤。
  8. 點選「下一步」
  9. 在「完成」畫面中檢查新的動態饋給設定,然後按一下「提交」

選項 B:使用 syslog 的 Bindplane 代理程式

這個方法會使用 Bindplane 代理程式收集 ServiceNow 稽核記錄,並轉送至 Google Security Operations。由於 ServiceNow 原生不支援稽核記錄的系統記錄,我們會使用指令碼查詢 ServiceNow REST API,並透過系統記錄將記錄轉送至 Bindplane 代理程式。

事前準備

請確認您已完成下列事前準備事項:

  • Google SecOps 執行個體
  • 搭載 systemd 的 Windows 2016 以上版本或 Linux 主機
  • 如果透過 Proxy 執行,請確保防火牆通訊埠已根據 Bindplane 代理程式需求開啟
  • ServiceNow 管理控制台或設備的特殊存取權

取得 Google SecOps 擷取驗證檔案

  1. 登入 Google SecOps 控制台。
  2. 依序前往「SIEM 設定」>「收集代理程式」
  3. 下載擷取驗證檔案。將檔案安全地儲存在要安裝 Bindplane 的系統上。

取得 Google SecOps 客戶 ID

  1. 登入 Google SecOps 控制台。
  2. 依序前往「SIEM 設定」>「設定檔」
  3. 複製並儲存「機構詳細資料」專區中的客戶 ID

安裝 Bindplane 代理程式

請按照下列操作說明,在 Windows 或 Linux 作業系統上安裝 Bindplane 代理程式。

Linux 安裝

  1. 開啟具備根層級或 sudo 權限的終端機。
  2. 執行下列指令:

    sudo sh -c "$(curl -fsSlL https://github.com/observiq/bindplane-agent/releases/latest/download/install_unix.sh)" install_unix.sh
    

其他安裝資源

設定 Bindplane 代理程式,擷取系統記錄檔並傳送至 Google SecOps

  1. 存取設定檔:

    1. 找出 config.yaml 檔案。通常位於 Linux 的 /etc/bindplane-agent/ 目錄,或 Windows 的安裝目錄。
    2. 使用文字編輯器 (例如 nanovi 或記事本) 開啟檔案。
  2. 按照下列方式編輯 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)

  1. 將指令碼設定為可執行:

    chmod +x servicenow_audit_to_syslog.py
    
  2. 建立 Cron 工作,每小時執行一次指令碼:

    crontab -e
    
  3. 新增下列程式碼:

    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)

  1. 開啟「工作排程器」
  2. 按一下「建立工作」
  3. 提供下列設定:
    • 名稱:ServiceNowAuditToSyslog
    • 安全性選項:無論使用者是否登入,都會執行
  4. 前往「觸發條件」分頁。
  5. 按一下「新增」,然後將執行頻率設為每小時。
  6. 前往「動作」分頁。
  7. 按一下「新增」,然後設定:
    • 動作:啟動程式
    • 「程式/指令碼」:powershell.exe
    • 引數:-ExecutionPolicy Bypass -File "C:\path\to\ServiceNow-Audit-To-Syslog.ps1"
  8. 按一下「確定」儲存工作。

還有其他問題嗎?向社群成員和 Google SecOps 專業人員尋求答案。