Recoger registros de auditoría de ServiceNow

Disponible en:

En este documento se explica cómo ingerir registros de auditoría de ServiceNow en Google Security Operations mediante varios métodos.

Opción A: AWS S3 con Lambda

Este método usa AWS Lambda para consultar periódicamente la API REST de ServiceNow en busca de registros de auditoría y almacenarlos en un contenedor de S3. A continuación, Google Security Operations recoge los registros del segmento de S3.

Antes de empezar

  • Una instancia de Google SecOps
  • Acceso privilegiado al arrendatario o a la API de ServiceNow
  • Acceso con privilegios a AWS (S3, IAM, Lambda y EventBridge)

Recopilar los requisitos previos de ServiceNow (IDs, claves de API, IDs de organización y tokens)

  1. Inicia sesión en la consola de administración de ServiceNow.
  2. Ve a Seguridad del sistema > Usuarios y grupos > Usuarios.
  3. Crea un usuario o selecciona uno que ya tengas con los permisos adecuados para acceder a los registros de auditoría.
  4. Copia y guarda en un lugar seguro los siguientes detalles:
    • Username (Nombre de usuario)
    • Password (Contraseña)
    • URL de la instancia (por ejemplo, https://instance.service-now.com)

Configurar un segmento de AWS S3 y IAM para Google SecOps

  1. Crea un segmento de Amazon S3 siguiendo esta guía de usuario: Crear un segmento.
  2. Guarda el nombre y la región del segmento para consultarlos más adelante (por ejemplo, servicenow-audit-logs).
  3. Crea un usuario siguiendo esta guía: Crear un usuario de gestión de identidades y accesos.
  4. Selecciona el usuario creado.
  5. Selecciona la pestaña Credenciales de seguridad.
  6. En la sección Claves de acceso, haz clic en Crear clave de acceso.
  7. Selecciona Servicio de terceros como Caso práctico.
  8. Haz clic en Siguiente.
  9. Opcional: añade una etiqueta de descripción.
  10. Haz clic en Crear clave de acceso.
  11. Haz clic en Descargar archivo CSV para guardar la clave de acceso y la clave de acceso secreta para usarlas más adelante.
  12. Haz clic en Listo.
  13. Selecciona la pestaña Permisos.
  14. En la sección Políticas de permisos, haz clic en Añadir permisos .
  15. Selecciona Añadir permisos.
  16. Seleccione Adjuntar políticas directamente.
  17. Busca y selecciona la política AmazonS3FullAccess.
  18. Haz clic en Siguiente.
  19. Haz clic en Añadir permisos.

Configurar la política y el rol de gestión de identidades y accesos para las subidas de S3

  1. En la consola de AWS, ve a IAM > Policies > Create policy > JSON tab.
  2. Copia y pega la política que se indica más abajo.

    {
      "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"
        }
      ]
    }
    
    • Sustituye servicenow-audit-logs si has introducido otro nombre de segmento.
  3. Haz clic en Siguiente > Crear política.

  4. Ve a IAM > Roles > Crear rol > Servicio de AWS > Lambda.

  5. Adjunte la política que acaba de crear.

  6. Dale el nombre servicenow-audit-lambda-role al rol y haz clic en Crear rol.

Crear la función Lambda

  1. En la consola de AWS, ve a Lambda > Funciones > Crear función.
  2. Haz clic en Crear desde cero.
  3. Proporciona los siguientes detalles de configuración:

    Ajuste Valor
    Nombre servicenow-audit-collector
    Tiempo de ejecución Python 3.13
    Arquitectura x86_64
    Rol de ejecución servicenow-audit-lambda-role
  4. Una vez creada la función, abra la pestaña Código, elimine el stub e introduzca el siguiente código (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. Ve a Configuración > Variables de entorno > Editar > Añadir nueva variable de entorno.

  6. Introduce las siguientes variables de entorno y sustituye los valores por los tuyos.

    Clave Valor de ejemplo
    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. Una vez creada la función, permanece en su página (o abre Lambda > Funciones > servicenow-audit-collector).

  8. Seleccione la pestaña Configuración.

  9. En el panel Configuración general, haz clic en Editar.

  10. Cambia Tiempo de espera a 5 minutos (300 segundos) y haz clic en Guardar.

Crear una programación de EventBridge

  1. Ve a Amazon EventBridge > Scheduler > Create schedule (Amazon EventBridge > Programador > Crear programación).
  2. Proporcione los siguientes detalles de configuración:
    • Programación periódica: Precio (1 hour).
    • Destino: tu función Lambda servicenow-audit-collector.
    • Nombre: servicenow-audit-collector-1h.
  3. Haz clic en Crear programación.

Configurar un feed en Google SecOps para ingerir registros de auditoría de ServiceNow

  1. Ve a Configuración de SIEM > Feeds.
  2. Haz clic en + Añadir nuevo feed.
  3. En el campo Nombre del feed, introduce un nombre para el feed (por ejemplo, ServiceNow Audit logs).
  4. Selecciona Amazon S3 V2 como Tipo de fuente.
  5. Selecciona Auditoría de ServiceNow como Tipo de registro.
  6. Haz clic en Siguiente.
  7. Especifique valores para los siguientes parámetros de entrada:
    • URI de S3: s3://servicenow-audit-logs/audit-logs/
    • Opciones de eliminación de la fuente: selecciona la opción de eliminación que prefieras.
    • Antigüedad máxima del archivo: incluye los archivos modificados en los últimos días. El valor predeterminado es de 180 días.
    • ID de clave de acceso: clave de acceso de usuario con acceso al bucket de S3.
    • Clave de acceso secreta: clave secreta del usuario con acceso al segmento de S3.
    • Espacio de nombres de recursos: el espacio de nombres de recursos.
    • Etiquetas de ingestión: la etiqueta aplicada a los eventos de este feed.
  8. Haz clic en Siguiente.
  9. Revise la nueva configuración del feed en la pantalla Finalizar y, a continuación, haga clic en Enviar.

Opción B: agente de Bindplane con syslog

Este método utiliza un agente de Bindplane para recoger los registros de auditoría de ServiceNow y reenviarlos a Google Security Operations. Como ServiceNow no admite syslog de forma nativa para los registros de auditoría, usaremos una secuencia de comandos para consultar la API REST de ServiceNow y reenviar los registros al agente de Bindplane a través de syslog.

Antes de empezar

Asegúrate de que cumples los siguientes requisitos previos:

  • Una instancia de Google SecOps
  • Un host Windows 2016 o posterior, o Linux con systemd
  • Si se ejecuta a través de un proxy, asegúrese de que los puertos del cortafuegos estén abiertos según los requisitos del agente Bindplane.
  • Acceso con privilegios a la consola de gestión o al dispositivo de ServiceNow

Obtener el archivo de autenticación de ingestión de Google SecOps

  1. Inicia sesión en la consola de Google SecOps.
  2. Ve a Configuración de SIEM > Agentes de recogida.
  3. Descarga el archivo de autenticación de ingestión. Guarda el archivo de forma segura en el sistema en el que se instalará Bindplane.

Obtener el ID de cliente de Google SecOps

  1. Inicia sesión en la consola de Google SecOps.
  2. Ve a Configuración de SIEM > Perfil.
  3. Copia y guarda el ID de cliente de la sección Detalles de la organización.

Instalar el agente de Bindplane

Instala el agente Bindplane en tu sistema operativo Windows o Linux siguiendo las instrucciones que se indican a continuación.

Instalación de Linux

  1. Abre un terminal con privilegios de root o sudo.
  2. Ejecuta el siguiente comando:

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

Recursos de instalación adicionales

Configurar el agente de Bindplane para ingerir Syslog y enviarlo a Google SecOps

  1. Accede al archivo de configuración:

    1. Busca el archivo config.yaml. Normalmente, se encuentra en el directorio /etc/bindplane-agent/ en Linux o en el directorio de instalación en Windows.
    2. Abre el archivo con un editor de texto (por ejemplo, nano, vi o Bloc de notas).
  2. Edita el archivo config.yaml de la siguiente manera:

    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
    

Reinicia el agente de Bindplane para aplicar los cambios

  • Para reiniciar el agente de Bindplane en Linux, ejecuta el siguiente comando:

    sudo systemctl restart bindplane-agent
    
  • Para reiniciar el agente de Bindplane en Windows, puedes usar la consola Servicios o introducir el siguiente comando:

    net stop BindPlaneAgent && net start BindPlaneAgent
    

Crear una secuencia de comandos para reenviar los registros de auditoría de ServiceNow a syslog

Como ServiceNow no admite de forma nativa syslog para los registros de auditoría, crearemos una secuencia de comandos que consulte la API REST de ServiceNow y reenvíe los registros a syslog. Esta secuencia de comandos se puede programar para que se ejecute periódicamente.

Ejemplo de secuencia de comandos de Python (Linux)

  • Crea un archivo llamado servicenow_audit_to_syslog.py con el siguiente contenido:

    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()
    

Configurar la ejecución programada (Linux)

  1. Haz que la secuencia de comandos sea ejecutable:

    chmod +x servicenow_audit_to_syslog.py
    
  2. Crea una tarea cron para ejecutar la secuencia de comandos cada hora:

    crontab -e
    
  3. Añade la siguiente línea:

    0 * * * * /usr/bin/python3 /path/to/servicenow_audit_to_syslog.py >> /tmp/servicenow_audit_to_syslog.log 2>&1
    

Ejemplo de secuencia de comandos de PowerShell (Windows)

  • Crea un archivo llamado ServiceNow-Audit-To-Syslog.ps1 con el siguiente contenido:

    # 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"
    }
    

Configurar la ejecución programada (Windows)

  1. Abre Programador de tareas.
  2. Haz clic en Crear tarea.
  3. Proporcione la siguiente configuración:
    • Nombre: ServiceNowAuditToSyslog
    • Opciones de seguridad: ejecutar si el usuario ha iniciado sesión o no
  4. Ve a la pestaña Activadores.
  5. Haz clic en Nuevo y configúralo para que se ejecute cada hora.
  6. Ve a la pestaña Acciones.
  7. Haz clic en Nuevo y define lo siguiente:
    • Acción: iniciar un programa
    • Programa o secuencia de comandos: powershell.exe
    • Argumentos: -ExecutionPolicy Bypass -File "C:\ruta\a\ServiceNow-Audit-To-Syslog.ps1"
  8. Haz clic en Aceptar para guardar la tarea.

¿Necesitas más ayuda? Recibe respuestas de los miembros de la comunidad y de los profesionales de Google SecOps.