Registros de auditoría de ServiceNow

Se admite en los siguientes sistemas operativos:

En este documento, se explica cómo transferir registros de auditoría de ServiceNow a Google Security Operations con varios métodos.

Opción A: GCS con función de Cloud Run

Este método usa una función de Cloud Run para consultar periódicamente la API de REST de ServiceNow en busca de registros de auditoría y almacenarlos en un bucket de GCS. Luego, Google Security Operations recopila los registros del bucket de GCS.

Antes de comenzar

Asegúrate de cumplir con los siguientes requisitos previos:

  • Una instancia de Google SecOps
  • Acceso privilegiado al arrendatario o a la API de ServiceNow con los roles adecuados (por lo general, admin o usuario con acceso de lectura a la tabla sys_audit)
  • Un proyecto de GCP con la API de Cloud Storage habilitada
  • Permisos para crear y administrar buckets de GCS
  • Permisos para administrar políticas de IAM en buckets de GCS
  • Permisos para crear servicios de Cloud Run, temas de Pub/Sub y trabajos de Cloud Scheduler

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

  1. Accede a la Consola del administrador de ServiceNow.
  2. Ve a Seguridad del sistema > Usuarios y grupos > Usuarios.
  3. Crea un usuario nuevo o selecciona uno existente con los permisos adecuados para acceder a los registros de auditoría.

  4. Copia y guarda en una ubicación segura los siguientes detalles:

    • Nombre de usuario
    • Contraseña
    • URL de la instancia (por ejemplo, https://instance.service-now.com)

Configura la LCA para usuarios que no son administradores

Si deseas usar una cuenta de usuario que no sea de administrador, debes crear una lista de control de acceso (LCA) personalizada para otorgar acceso de lectura a la tabla sys_audit:

  1. Accede a la Consola del administrador de ServiceNow como administrador.
  2. Ve a Seguridad del sistema > Control de acceso (ACL).
  3. Haz clic en Nuevo.
  4. Proporciona los siguientes detalles de configuración:
    • Tipo: Selecciona registro.
    • Operación: Selecciona read.
    • Nombre: Ingresa sys_audit.
    • Descripción: ingresa Allow read access to sys_audit table for Chronicle integration.
  5. En el campo Requires role, agrega el rol asignado a tu usuario de integración (por ejemplo, chronicle_reader).
  6. Haz clic en Enviar.
  7. Verifica que la ACL esté activa y que el usuario pueda consultar la tabla sys_audit.

Crea un bucket de Google Cloud Storage

  1. Ve a Google Cloud Console.
  2. Selecciona tu proyecto o crea uno nuevo.
  3. En el menú de navegación, ve a Cloud Storage > Buckets.
  4. Haz clic en Crear bucket.

  5. Proporciona los siguientes detalles de configuración:

    Configuración Valor
    Asigna un nombre a tu bucket Ingresa un nombre global único (por ejemplo, servicenow-audit-logs).
    Tipo de ubicación Elige según tus necesidades (región, birregional, multirregional)
    Ubicación Selecciona la ubicación (por ejemplo, us-central1).
    Clase de almacenamiento Estándar (recomendado para los registros a los que se accede con frecuencia)
    Control de acceso Uniforme (recomendado)
    Herramientas de protección Opcional: Habilita el control de versiones de objetos o la política de retención

  6. Haz clic en Crear.

Crea una cuenta de servicio para la Cloud Run Function

La Cloud Run Function necesita una cuenta de servicio con permisos para escribir en el bucket de GCS y ser invocada por Pub/Sub.

Crear cuenta de servicio

  1. En GCP Console, ve a IAM y administración > Cuentas de servicio.
  2. Haz clic en Crear cuenta de servicio.
  3. Proporciona los siguientes detalles de configuración:
    • Nombre de la cuenta de servicio: Ingresa servicenow-audit-collector-sa.
    • Descripción de la cuenta de servicio: Ingresa Service account for Cloud Run function to collect ServiceNow audit logs.
  4. Haz clic en Crear y continuar.
  5. En la sección Otorga a esta cuenta de servicio acceso al proyecto, agrega los siguientes roles:
    1. Haz clic en Selecciona un rol.
    2. Busca y selecciona Administrador de objetos de almacenamiento.
    3. Haz clic en + Agregar otra función.
    4. Busca y selecciona Invocador de Cloud Run.
    5. Haz clic en + Agregar otra función.
    6. Busca y selecciona Cloud Functions Invoker.
  6. Haz clic en Continuar.
  7. Haz clic en Listo.

Estos roles son necesarios para las siguientes acciones:

  • Administrador de objetos de almacenamiento: Escribe registros en el bucket de GCS y administra archivos de estado
  • Invocador de Cloud Run: Permite que Pub/Sub invoque la función
  • Cloud Functions Invoker: Permite la invocación de funciones

Otorga permisos de IAM en el bucket de GCS

Otorga permisos de escritura a la cuenta de servicio en el bucket de GCS:

  1. Ve a Cloud Storage > Buckets.
  2. Haz clic en el nombre de tu bucket.
  3. Ve a la pestaña Permisos.
  4. Haz clic en Otorgar acceso.
  5. Proporciona los siguientes detalles de configuración:
    • Agregar principales: Ingresa el correo electrónico de la cuenta de servicio (por ejemplo, servicenow-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
    • Asignar roles: Selecciona Administrador de objetos de Storage.
  6. Haz clic en Guardar.

Crear tema de Pub/Sub

Crea un tema de Pub/Sub en el que Cloud Scheduler publicará y al que se suscribirá la función de Cloud Run.

  1. En GCP Console, ve a Pub/Sub > Temas.
  2. Haz clic en Crear un tema.
  3. Proporciona los siguientes detalles de configuración:
    • ID del tema: Ingresa servicenow-audit-trigger.
    • Deja el resto de la configuración con sus valores predeterminados.
  4. Haz clic en Crear.

Crea una función de Cloud Run para recopilar registros

La función de Cloud Run se activa con los mensajes de Pub/Sub de Cloud Scheduler para recuperar registros de la API de ServiceNow y escribirlos en GCS.

  1. En GCP Console, ve a Cloud Run.
  2. Haz clic en Crear servicio.
  3. Selecciona Función (usa un editor intercalado para crear una función).

  4. En la sección Configurar, proporciona los siguientes detalles de configuración:

    Configuración Valor
    Nombre del servicio servicenow-audit-collector
    Región Selecciona la región que coincida con tu bucket de GCS (por ejemplo, us-central1).
    Tiempo de ejecución Selecciona Python 3.12 o una versión posterior.

  5. En la sección Activador (opcional), haz lo siguiente:

    1. Haz clic en + Agregar activador.
    2. Selecciona Cloud Pub/Sub.
    3. En Selecciona un tema de Cloud Pub/Sub, elige el tema de Pub/Sub (servicenow-audit-trigger).
    4. Haz clic en Guardar.

  6. En la sección Autenticación, haz lo siguiente:

    1. Selecciona Solicitar autenticación.
    2. Verifica Identity and Access Management (IAM).

  7. Desplázate hacia abajo y expande Contenedores, redes y seguridad.

  8. Ve a la pestaña Seguridad:

    • Cuenta de servicio: Selecciona la cuenta de servicio (servicenow-audit-collector-sa).

  9. Ve a la pestaña Contenedores:

    1. Haz clic en Variables y secretos.
    2. Haz clic en + Agregar variable para cada variable de entorno:
    Nombre de la variable Valor de ejemplo Descripción
    GCS_BUCKET servicenow-audit-logs Nombre del bucket de GCS
    GCS_PREFIX audit-logs Prefijo para los archivos de registro
    STATE_KEY audit-logs/state.json Ruta de acceso al archivo de estado
    API_BASE_URL https://instance.service-now.com URL de la instancia de ServiceNow
    API_USERNAME your-username Nombre de usuario de ServiceNow
    API_PASSWORD your-password Contraseña de ServiceNow
    PAGE_SIZE 1000 Registros por página
    MAX_PAGES 1000 Cantidad máxima de páginas para recuperar

  10. En la sección Variables y Secrets, desplázate hacia abajo hasta Solicitudes:

    • Tiempo de espera de la solicitud: Ingresa 600 segundos (10 minutos).

  11. Ve a la pestaña Configuración:

    • En la sección Recursos, haz lo siguiente:
      • Memoria: Selecciona 512 MiB o más.
      • CPU: Selecciona 1.

  12. En la sección Ajuste de escala de revisión, haz lo siguiente:

    • Cantidad mínima de instancias: Ingresa 0.
    • Cantidad máxima de instancias: Ingresa 100 (o ajusta según la carga esperada).

  13. Haz clic en Crear.

  14. Espera a que se cree el servicio (de 1 a 2 minutos).

  15. Después de crear el servicio, se abrirá automáticamente el editor de código intercalado.

Agregar el código de función

  1. Ingresa main en el campo Punto de entrada.

  2. En el editor de código intercalado, crea dos archivos:

    • Primer archivo: main.py:

          import functions_framework
    from google.cloud import storage
    import json
    import os
    import urllib3
    from datetime import datetime, timezone, timedelta
    import time
    import base64

    # Initialize HTTP client with timeouts
    http = urllib3.PoolManager(
        timeout=urllib3.Timeout(connect=5.0, read=30.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', 'audit-logs')
    STATE_KEY = os.environ.get('STATE_KEY', 'audit-logs/state.json')
    API_BASE = os.environ.get('API_BASE_URL')
    USERNAME = os.environ.get('API_USERNAME')
    PASSWORD = os.environ.get('API_PASSWORD')
    PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '1000'))
    MAX_PAGES = int(os.environ.get('MAX_PAGES', '1000'))

    def parse_datetime(value: str) -> datetime:
        """Parse ServiceNow datetime string to datetime object."""
        # ServiceNow format: YYYY-MM-DD HH:MM:SS
        try:
            return datetime.strptime(value, '%Y-%m-%d %H:%M:%S').replace(tzinfo=timezone.utc)
        except ValueError:
            # Try ISO format as fallback
            if value.endswith("Z"):
                value = value[:-1] + "+00:00"
            return datetime.fromisoformat(value)

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

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

        if not all([GCS_BUCKET, API_BASE, USERNAME, PASSWORD]):
            print('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
            now = datetime.now(timezone.utc)
            last_time = None

            if isinstance(state, dict) and state.get("last_event_time"):
                try:
                    last_time = parse_datetime(state["last_event_time"])
                    # Overlap by 2 minutes to catch any delayed events
                    last_time = last_time - timedelta(minutes=2)
                except Exception as e:
                    print(f"Warning: Could not parse last_event_time: {e}")

            if last_time is None:
                last_time = now - timedelta(hours=24)

            print(f"Fetching logs from {last_time.strftime('%Y-%m-%d %H:%M:%S')} to {now.strftime('%Y-%m-%d %H:%M:%S')}")

            # Fetch logs
            records, newest_event_time = fetch_logs(
                api_base=API_BASE,
                username=USERNAME,
                password=PASSWORD,
                start_time=last_time,
                end_time=now,
                page_size=PAGE_SIZE,
                max_pages=MAX_PAGES,
            )

            if not records:
                print("No new log records found.")
                save_state(bucket, STATE_KEY, now.strftime('%Y-%m-%d %H:%M:%S'))
                return

            # Write to GCS as NDJSON
            timestamp = now.strftime('%Y%m%d_%H%M%S')
            object_key = f"{GCS_PREFIX}/logs_{timestamp}.ndjson"
            blob = bucket.blob(object_key)

            ndjson = '\n'.join([json.dumps(record, ensure_ascii=False) for record in records]) + '\n'
            blob.upload_from_string(ndjson, content_type='application/x-ndjson')

            print(f"Wrote {len(records)} records to gs://{GCS_BUCKET}/{object_key}")

            # Update state with newest event time
            if newest_event_time:
                save_state(bucket, STATE_KEY, newest_event_time)
            else:
                save_state(bucket, STATE_KEY, now.strftime('%Y-%m-%d %H:%M:%S'))

            print(f"Successfully processed {len(records)} records")

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

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

        return {}

    def save_state(bucket, key, last_event_time: str):
        """Save the last event timestamp to GCS state file."""
        try:
            state = {'last_event_time': last_event_time}
            blob = bucket.blob(key)
            blob.upload_from_string(
                json.dumps(state, indent=2),
                content_type='application/json'
            )
            print(f"Saved state: last_event_time={last_event_time}")
        except Exception as e:
            print(f"Warning: Could not save state: {e}")

    def fetch_logs(api_base: str, username: str, password: str, start_time: datetime, end_time: datetime, page_size: int, max_pages: int):
        """
        Fetch logs from ServiceNow sys_audit table with pagination and rate limiting.

        Args:
            api_base: ServiceNow instance URL
            username: ServiceNow username
            password: ServiceNow password
            start_time: Start time for log query
            end_time: End time for log query
            page_size: Number of records per page
            max_pages: Maximum total pages to fetch

        Returns:
            Tuple of (records list, newest_event_time string)
        """
        # Clean up base URL
        base_url = api_base.rstrip('/')

        endpoint = f"{base_url}/api/now/table/sys_audit"

        # Encode credentials using UTF-8
        auth_string = f"{username}:{password}"
        auth_bytes = auth_string.encode('utf-8')
        auth_b64 = base64.b64encode(auth_bytes).decode('utf-8')

        headers = {
            'Authorization': f'Basic {auth_b64}',
            'Accept': 'application/json',
            'Content-Type': 'application/json',
            'User-Agent': 'GoogleSecOps-ServiceNowCollector/1.0'
        }

        records = []
        newest_time = None
        page_num = 0
        backoff = 1.0
        offset = 0

        # Format timestamps for ServiceNow (YYYY-MM-DD HH:MM:SS)
        start_time_str = start_time.strftime('%Y-%m-%d %H:%M:%S')

        while True:
            page_num += 1

            if len(records) >= page_size * max_pages:
                print(f"Reached max_pages limit ({max_pages})")
                break

            # Build query parameters
            # Use >= operator for sys_created_on field (on or after)
            params = []
            params.append(f"sysparm_query=sys_created_on>={start_time_str}")
            params.append(f"sysparm_display_value=true")
            params.append(f"sysparm_limit={page_size}")
            params.append(f"sysparm_offset={offset}")

            url = f"{endpoint}?{'&'.join(params)}"

            try:
                response = http.request('GET', url, headers=headers)

                # Handle rate limiting with exponential backoff
                if response.status == 429:
                    retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
                    print(f"Rate limited (429). Retrying after {retry_after}s...")
                    time.sleep(retry_after)
                    backoff = min(backoff * 2, 30.0)
                    continue

                backoff = 1.0

                if response.status != 200:
                    print(f"HTTP Error: {response.status}")
                    response_text = response.data.decode('utf-8')
                    print(f"Response body: {response_text}")
                    return [], None

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

                if not page_results:
                    print(f"No more results (empty page)")
                    break

                print(f"Page {page_num}: Retrieved {len(page_results)} events")
                records.extend(page_results)

                # Track newest event time
                for event in page_results:
                    try:
                        event_time = event.get('sys_created_on')
                        if event_time:
                            if newest_time is None or parse_datetime(event_time) > parse_datetime(newest_time):
                                newest_time = event_time
                    except Exception as e:
                        print(f"Warning: Could not parse event time: {e}")

                # Check for more results
                if len(page_results) < page_size:
                    print(f"Reached last page (size={len(page_results)} < limit={page_size})")
                    break

                # Move to next page
                offset += page_size

                # Small delay to avoid rate limiting
                time.sleep(0.1)

            except Exception as e:
                print(f"Error fetching logs: {e}")
                return [], None

        print(f"Retrieved {len(records)} total records from {page_num} pages")
        return records, newest_time
    ```

    • Segundo archivo: requirements.txt:

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

  3. Haz clic en Implementar para guardar y, luego, implementar la función.

  4. Espera a que se complete la implementación (de 2 a 3 minutos).

Crea un trabajo de Cloud Scheduler

Cloud Scheduler publica mensajes en el tema de Pub/Sub a intervalos regulares, lo que activa la función de Cloud Run.

  1. En GCP Console, ve a Cloud Scheduler.
  2. Haz clic en Crear trabajo.

  3. Proporciona los siguientes detalles de configuración:

    Configuración Valor
    Nombre servicenow-audit-collector-hourly
    Región Selecciona la misma región que la función de Cloud Run
    Frecuencia 0 * * * * (cada hora, en punto)
    Zona horaria Selecciona la zona horaria (se recomienda UTC)
    Tipo de orientación Pub/Sub
    Tema Selecciona el tema de Pub/Sub (servicenow-audit-trigger).
    Cuerpo del mensaje {} (objeto JSON vacío)

  4. Haz clic en Crear.

Opciones de frecuencia de programación

  • Elige la frecuencia según los requisitos de latencia y volumen de registros:

    Frecuencia Expresión cron Caso de uso
    Cada 5 minutos */5 * * * * Alto volumen y baja latencia
    Cada 15 minutos */15 * * * * Volumen medio
    Cada 1 hora 0 * * * * Estándar (opción recomendada)
    Cada 6 horas 0 */6 * * * Procesamiento por lotes y volumen bajo
    Diario 0 0 * * * Recopilación de datos históricos

Prueba la integración

  1. En la consola de Cloud Scheduler, busca tu trabajo.
  2. Haz clic en Forzar ejecución para activar el trabajo de forma manual.
  3. Espera unos segundos.
  4. Ve a Cloud Run > Servicios.
  5. Haz clic en el nombre de tu función (servicenow-audit-collector).
  6. Haz clic en la pestaña Registros.

  7. Verifica que la función se haya ejecutado correctamente. Busca lo siguiente:

    Fetching logs from YYYY-MM-DD HH:MM:SS to YYYY-MM-DD HH:MM:SS
Page 1: Retrieved X events
Wrote X records to gs://bucket-name/audit-logs/logs_YYYYMMDD_HHMMSS.ndjson
Successfully processed X records

  8. Ve a Cloud Storage > Buckets.

  9. Haz clic en el nombre de tu bucket.

  10. Navega a la carpeta del prefijo (audit-logs/).

  11. Verifica que se haya creado un archivo .ndjson nuevo con la marca de tiempo actual.

Si ves errores en los registros, haz lo siguiente:

  • HTTP 401: Verifica las credenciales de la API en las variables de entorno
  • HTTP 403: Verifica que la cuenta tenga los permisos necesarios (rol de administrador o ACL personalizada para sys_audit)
  • HTTP 429: Limitación de frecuencia. La función volverá a intentarlo automáticamente con una espera exponencial.
  • Faltan variables de entorno: Verifica que estén configuradas todas las variables requeridas.

Recupera la cuenta de servicio de Google SecOps

Las Operaciones de seguridad de Google usan una cuenta de servicio única para leer datos de tu bucket de GCS. Debes otorgar acceso a tu bucket a esta cuenta de servicio.

Obtén el correo electrónico de la cuenta de servicio

  1. Ve a Configuración de SIEM > Feeds.
  2. Haz clic en Agregar feed nuevo.
  3. Haz clic en Configura un feed único.
  4. En el campo Nombre del feed, ingresa un nombre para el feed (por ejemplo, ServiceNow Audit logs).
  5. Selecciona Google Cloud Storage V2 como el Tipo de fuente.
  6. Selecciona ServiceNow Audit como el Tipo de registro.

  7. Haz clic en Obtener cuenta de servicio. Se muestra un correo electrónico único de la cuenta de servicio, por ejemplo:

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com

  8. Copia esta dirección de correo electrónico para usarla en el siguiente paso.

Otorga permisos de IAM a la cuenta de servicio de Google SecOps

La cuenta de servicio de Google SecOps necesita el rol de visualizador de objetos de almacenamiento en tu bucket de GCS.

  1. Ve a Cloud Storage > Buckets.
  2. Haz clic en el nombre de tu bucket.
  3. Ve a la pestaña Permisos.
  4. Haz clic en Otorgar acceso.
  5. Proporciona los siguientes detalles de configuración:
    • Agregar principales: Pega el correo electrónico de la cuenta de servicio de Google SecOps.
    • Asignar roles: Selecciona Visualizador de objetos de Storage.

  6. Haz clic en Guardar.

Configura un feed en Google SecOps para transferir registros de auditoría de ServiceNow

  1. Ve a Configuración de SIEM > Feeds.
  2. Haz clic en Agregar feed nuevo.
  3. Haz clic en Configura un feed único.
  4. En el campo Nombre del feed, ingresa un nombre para el feed (por ejemplo, ServiceNow Audit logs).
  5. Selecciona Google Cloud Storage V2 como el Tipo de fuente.
  6. Selecciona ServiceNow Audit como el Tipo de registro.
  7. Haz clic en Siguiente.

  8. Especifica valores para los siguientes parámetros de entrada:

    • URL del bucket de almacenamiento: Ingresa el URI del bucket de GCS con la ruta de acceso del prefijo:

      gs://servicenow-audit-logs/audit-logs/

      • Reemplaza lo siguiente:

        • servicenow-audit-logs: Es el nombre de tu bucket de GCS.
        • audit-logs: Es el prefijo o la ruta de acceso a la carpeta en la que se almacenan los registros.

    • Opción de borrado de la fuente: Selecciona la opción de borrado según tu preferencia:

      • Nunca: Nunca borra ningún archivo después de las transferencias (se recomienda para las pruebas).
      • Borrar archivos transferidos: Borra los archivos después de la transferencia exitosa.

      • Borrar los archivos transferidos y los directorios vacíos: Borra los archivos y los directorios vacíos después de la transferencia exitosa.

    • Antigüedad máxima del archivo: Incluye los archivos modificados en la cantidad de días especificada. El valor predeterminado es de 180 días.

    • Espacio de nombres del recurso: Es el espacio de nombres del recurso.

    • Etiquetas de transmisión: Es la etiqueta que se aplicará a los eventos de este feed.

  9. Haz clic en Siguiente.

  10. Revisa la nueva configuración del feed en la pantalla Finalizar y, luego, haz clic en Enviar.

Opción B: Agente de Bindplane con syslog

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

Antes de comenzar

Asegúrate de cumplir con los siguientes requisitos previos:

  • Una instancia de Google SecOps
  • Windows Server 2016 o versiones posteriores, o host de Linux con systemd
  • Conectividad de red entre el agente de Bindplane y ServiceNow
  • Si se ejecuta detrás de un proxy, asegúrate de que los puertos de firewall estén abiertos según los requisitos del agente de Bindplane.
  • Acceso con privilegios a la consola o el dispositivo de administración de ServiceNow con los roles adecuados (por lo general, admin o usuario con acceso de lectura a la tabla sys_audit)

Obtén el archivo de autenticación de transferencia de Google SecOps

  1. Accede a la consola de Google SecOps.
  2. Ve a Configuración de SIEM > Agente de recopilación.
  3. Haz clic en Descargar para descargar el archivo de autenticación de la transferencia.

  4. Guarda el archivo de forma segura en el sistema en el que se instalará Bindplane.

Obtén el ID de cliente de Google SecOps

  1. Accede a 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.

Instala el agente de BindPlane

Instala el agente de Bindplane en tu sistema operativo Windows o Linux según las siguientes instrucciones.

Instalación en Windows

  1. Abre el símbolo del sistema o PowerShell como administrador.

  2. Ejecuta el siguiente comando:

    msiexec /i "https://github.com/observIQ/bindplane-agent/releases/latest/download/observiq-otel-collector.msi" /quiet

  3. Espera a que se complete la instalación.

  4. Ejecute el siguiente comando para verificar la instalación:

    sc query observiq-otel-collector

El servicio debería mostrarse como RUNNING.

Instalación en Linux

  1. Abre una terminal con privilegios de administrador 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

  3. Espera a que se complete la instalación.

  4. Ejecute el siguiente comando para verificar la instalación:

    sudo systemctl status observiq-otel-collector

El servicio debería mostrarse como activo (en ejecución).

Recursos de instalación adicionales

Para obtener más opciones de instalación y solucionar problemas, consulta la guía de instalación del agente de Bindplane.

Configura el agente de BindPlane para transferir el syslog y enviarlo a Google SecOps

Ubica el archivo de configuración

Linux: bash sudo nano /etc/bindplane-agent/config.yaml

Windows: cmd notepad "C:\Program Files\observIQ OpenTelemetry Collector\config.yaml"

Edita el archivo de configuración

Reemplaza todo el contenido de config.yaml con la siguiente configuración:

```yaml
receivers:
  udplog:
    listen_address: "0.0.0.0:514"

exporters:
  chronicle/servicenow_audit:
    compression: gzip
    creds_file_path: '/path/to/ingestion-authentication-file.json'
    customer_id: '<YOUR_CUSTOMER_ID>'
    endpoint: <CUSTOMER_REGION_ENDPOINT>
    log_type: 'SERVICENOW_AUDIT'
    raw_log_field: body
    ingestion_labels:
      service: servicenow

service:
  pipelines:
    logs/servicenow_to_chronicle:
      receivers:
        - udplog
      exporters:
        - chronicle/servicenow_audit
```

Parámetros de configuración

Reemplaza los marcadores de posición que se indican más abajo:

  • listen_address: Es la dirección IP y el puerto que se escucharán. Usa 0.0.0.0:514 para escuchar en todas las interfaces del puerto 514.
  • creds_file_path: Ruta de acceso completa al archivo de autenticación de la transferencia:
    • Linux: /etc/bindplane-agent/ingestion-auth.json
    • Windows: C:\Program Files\observIQ OpenTelemetry Collector\ingestion-auth.json
  • <YOUR_CUSTOMER_ID>: Es el ID de cliente del paso anterior.
  • <CUSTOMER_REGION_ENDPOINT>: URL del extremo regional:
    • EE.UU.: malachiteingestion-pa.googleapis.com
    • Europa: europe-malachiteingestion-pa.googleapis.com
    • Asia: asia-southeast1-malachiteingestion-pa.googleapis.com
    • Consulta Extremos regionales para obtener la lista completa.

Guarda el archivo de configuración

Después de editarlo, guarda el archivo: * Linux: Presiona Ctrl+O, luego Enter y, por último, Ctrl+X. * Windows: Haz clic en Archivo > Guardar.

Reinicia el agente de Bindplane para aplicar los cambios

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

    sudo systemctl restart observiq-otel-collector

    1. Verifica que el servicio esté en ejecución:

      sudo systemctl status observiq-otel-collector

    2. Revisa los registros en busca de errores:

      sudo journalctl -u observiq-otel-collector -f

  • Para reiniciar el agente de Bindplane en Windows, elige una de las siguientes opciones:

    • Con el símbolo del sistema o PowerShell como administrador, haz lo siguiente:

      net stop observiq-otel-collector && net start observiq-otel-collector

    • Usa la consola de Services:

    1. Presiona Win+R, escribe services.msc y presiona Intro.
    2. Busca observIQ OpenTelemetry Collector.

    3. Haz clic con el botón derecho y selecciona Reiniciar.

    4. Verifica que el servicio esté en ejecución:

      sc query observiq-otel-collector

    5. Revisa los registros en busca de errores:

      type "C:\Program Files\observIQ OpenTelemetry Collector\log\collector.log"

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

Dado que ServiceNow no admite de forma nativa syslog para los registros de auditoría, crearemos una secuencia de comandos que consulte la API de 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-01 00: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 using UTF-8
    auth_string = f"{USERNAME}:{PASSWORD}"
    auth_bytes = auth_string.encode('utf-8')
    auth_encoded = base64.b64encode(auth_bytes).decode('utf-8')

    # Setup HTTP client
    http = urllib3.PoolManager()
    headers = {
        'Authorization': f'Basic {auth_encoded}',
        'Accept': 'application/json'
    }

    results = []
    offset = 0

    # Format timestamp for ServiceNow (YYYY-MM-DD HH:MM:SS format)
    # Convert ISO format to ServiceNow format if needed
    if 'T' in last_run_timestamp:
        last_run_timestamp = last_run_timestamp.replace('T', ' ').split('.')[0]

    for page in range(MAX_PAGES):
        # Build query with pagination
        # Use >= operator for sys_created_on field (on or after)
        query_params = (
            f"sysparm_query=sys_created_on>={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().strftime('%Y-%m-%d %H:%M:%S')

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

Configura la ejecución programada (Linux)

  • Haz que la secuencia de comandos sea ejecutable:

    chmod +x servicenow_audit_to_syslog.py

  • Crea un trabajo cron para ejecutar la secuencia de comandos cada hora:

    crontab -e

  • Agrega 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-01 00:00:00'
        }
    } catch {
        return '1970-01-01 00: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 using UTF-8 encoding
    $Auth = [System.Convert]::ToBase64String([System.Text.Encoding]::UTF8.GetBytes("${Username}:${Password}"))
    $Headers = @{
        Authorization = "Basic ${Auth}"
        Accept = 'application/json'
    }

    $Results = @()
    $Offset = 0

    # Format timestamp for ServiceNow (YYYY-MM-DD HH:MM:SS format)
    # Convert ISO format to ServiceNow format if needed
    if ($LastRunTimestamp -match 'T') {
        $LastRunTimestamp = $LastRunTimestamp -replace 'T', ' '
        $LastRunTimestamp = $LastRunTimestamp -replace '\.\d+', ''
    }

    for ($page = 0; $page -lt $MaxPages; $page++) {
        # Build query with pagination
        # Use >= operator for sys_created_on field (on or after)
        $QueryParams = "sysparm_query=sys_created_on>=${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-dd HH: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"
}

Configura la ejecución programada (Windows)

  1. Abre el Programador de tareas.
  2. Haz clic en Crear tarea.
  3. Proporciona la siguiente configuración:
    • Nombre: ServiceNowAuditToSyslog
    • Opciones de seguridad: Se ejecuta independientemente de si el usuario accedió 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 establece lo siguiente:
    • Acción: Iniciar un programa
    • Programa o secuencia de comandos: powershell.exe
    • Argumentos: -ExecutionPolicy Bypass -File "C:\path\to\ServiceNow-Audit-To-Syslog.ps1"
  8. Haz clic en Aceptar para guardar la tarea.

¿Necesitas más ayuda? Obtén respuestas de miembros de la comunidad y profesionales de Google SecOps.