Recopila registros de BeyondTrust Endpoint Privilege Management (EPM)

Se admite en los siguientes sistemas operativos:

Google SecOps SIEM

En este documento, se explica cómo transferir registros de BeyondTrust Endpoint Privilege Management (EPM) a Google Security Operations con Cloud Storage. El analizador se enfoca en transformar los datos de registro JSON sin procesar de BeyondTrust Endpoint en un formato estructurado que cumpla con el UDM de Google SecOps. Primero, inicializa los valores predeterminados para varios campos y, luego, analiza la carga útil JSON, y, posteriormente, asigna campos específicos del registro sin procesar a los campos del UDM correspondientes dentro del objeto event.idm.read_only_udm.

Antes de comenzar

Asegúrate de cumplir con los siguientes requisitos previos:

• Una instancia de Google SecOps
• Un Google Cloud proyecto con la API de Cloud Storage habilitada
• Permisos para crear y administrar buckets de Cloud Storage
• Permisos para administrar políticas de Identity and Access Management (IAM) en buckets de Cloud Storage
• Permisos para crear servicios de Cloud Run Functions, temas de Pub/Sub y trabajos de Cloud Scheduler
• Acceso con privilegios al arrendatario o la API de Endpoint Privilege Management de BeyondTrust

Cree un bucket de Cloud Storage

1. Ve a la consola deGoogle Cloud .
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:

   Parámetro de configuración	Valor
   Asigna un nombre a tu bucket	Ingresa un nombre global único (por ejemplo, beyondtrust-epm-logs).
   Tipo de ubicación	Elige según tus necesidades (regional, birregional o 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.

Recopila las credenciales de la API de BeyondTrust EPM

1. Accede a la consola web de BeyondTrust Privilege Management como administrador.
2. Ve a Configuración > Configuración > Configuración de la API.
3. Haz clic en Create an API Account.
4. Proporciona los siguientes detalles de configuración:
   • Nombre: Ingresa Google SecOps Collector.
   • Acceso a la API: Habilita Informes (solo lectura). El extremo /management-api/v3/Events/FromStartDate que usa este recopilador pertenece a la categoría de permisos de Informes. Habilita Auditoría (solo lectura) solo si también deseas recopilar auditorías de actividad de la consola web.
5. Copia y guarda el ID de cliente y el secreto del cliente.

6. Copia la URL base de la API que se muestra en la parte superior de la página Configuración de la API. Por lo general, es https://<your-tenant>-services.pm.beyondtrustcloud.com. Lo usarás como BPT_API_URL.

   Nota: Copia el valor de la consola en lugar de construirlo. Los inquilinos administrados en app.beyondtrust.io usan un host diferente.

Crea una cuenta de servicio para la función de Cloud Run

La función de Cloud Run necesita una cuenta de servicio con permisos para escribir en el bucket de Cloud Storage y para que Pub/Sub la invoque.

Crea la cuenta de servicio

1. En la consola de Google Cloud , 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 beyondtrust-epm-collector-sa.
   • Descripción de la cuenta de servicio: Ingresa Service account for Cloud Run function to collect BeyondTrust EPM logs.
4. Haz clic en Crear y continuar.
5. En la sección Otorga acceso al proyecto a esta cuenta de servicio, agrega los siguientes roles:
   1. Haz clic en Selecciona un rol.
   2. Busca y selecciona Administrador de objetos de Storage.
   3. Haz clic en + Agregar otro rol.
   4. Busca y selecciona Invocador de Cloud Run.
   5. Haz clic en + Agregar otro rol.
   6. Busca y selecciona Invocador de Cloud Functions.
6. Haz clic en Continuar.
7. Haz clic en Listo.

Estos roles son necesarios para lo siguiente:

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

Otorga permisos de IAM en un bucket de Cloud Storage

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

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 entidades: Ingresa el correo electrónico de la cuenta de servicio. Por ejemplo, beyondtrust-epm-collector-sa@PROJECT_ID.iam.gserviceaccount.com
   • Asignar roles: Selecciona Administrador de objetos de Storage.
6. Haz clic en Guardar.

Crea un 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 la Google Cloud consola, ve a Pub/Sub > Temas.
2. Haz clic en Crear un tema.
3. Proporciona los siguientes detalles de configuración:
   • ID del tema: Ingresa beyondtrust-epm-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 mensajes de Pub/Sub de Cloud Scheduler para recuperar registros de la API de BeyondTrust EPM y los escribe en Cloud Storage.

1. En la consola de Google Cloud , 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:

   Parámetro de configuración	Valor
   Nombre del servicio	beyondtrust-epm-collector
   Región	Selecciona la región que coincida con tu bucket de Cloud Storage (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 beyondtrust-epm-trigger.
   4. Haz clic en Guardar.

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

   1. Selecciona Necesita autenticación.
   2. Verifica Identity and Access Management (IAM).
   Nota: Pub/Sub controla automáticamente la autenticación cuando se invoca la función.

7. Ve a Contenedores, redes y seguridad y expándelo.

8. Ve a la pestaña Seguridad:

   • Cuenta de servicio: Selecciona la cuenta de servicio beyondtrust-epm-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
   GCS_BUCKET	beyondtrust-epm-logs
   GCS_PREFIX	beyondtrust-epm/
   STATE_KEY	beyondtrust-epm-state.json
   BPT_API_URL	https://yourtenant-services.pm.beyondtrustcloud.com
   CLIENT_ID	your-client-id
   CLIENT_SECRET	your-client-secret
   RECORD_SIZE	1000
   MAX_BATCHES	50
   LOOKBACK_HOURS	24

   • RECORD_SIZE: Registros por solicitud. /Events/FromStartDate acepta valores de 1 a 1,000; la función limita los valores más altos.
   • MAX_BATCHES: Es el límite de seguridad para las solicitudes por ejecución. Si una ejecución lo alcanza, el cursor sigue avanzando hacia lo que se escribió, por lo que la siguiente ejecución continúa desde allí en lugar de repetir el trabajo.
   • LOOKBACK_HOURS: Indica qué tan atrás llega la primera ejecución. BeyondTrust conserva los eventos pasivos (códigos 106, 107, 603 y 706) durante 30 días y todos los demás eventos durante 90 días, por lo que una ventana de visualización más larga no muestra nada anterior a ese período.
   • LOOKBACK_HOURS: Indica qué tan atrás se debe llegar en la primera ejecución, antes de que exista cualquier cursor.
   Nota: El alcance de OAuth (urn:management:api) y las rutas de la API son valores fijos y se establecen en el código de la función, por lo que no son variables de entorno.
   Nota: STATE_KEY apunta fuera del prefijo de registro a propósito. El feed incorpora cada objeto en el URI de su bucket y, cuando se selecciona una opción de eliminación, borra lo que transfirió. Si mantienes el archivo de estado fuera de ese prefijo, evitarás que se ingiera como datos de registro y que se borre, lo que restablecería la recopilación y volvería a ingerir eventos duplicados.

10. En la pestaña Variables y Secrets, ve a Solicitudes:

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

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

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

12. Desplázate hasta Entorno de ejecución:

    • Selecciona Predeterminado (recomendado).

13. En la sección Escalamiento 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).

14. Haz clic en Crear.

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

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

Agrega un código de función

1. Ingresa main en Punto de entrada de la función.

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

   • Primer archivo:main.py:
   Nota: El extremo del token es /oauth/connect/token. El ejemplo de PowerShell publicado con la documentación de eventos de ECS muestra /oauth/token, pero ese ejemplo es incorrecto. Todas las guías de la API de BeyondTrust desde la versión 21.7 hasta la versión actual documentan /oauth/connect/token, y el conector Sentinel propio de BeyondTrust, el flujo de trabajo de QRadar y la app de Splunk llaman a /oauth/connect/token. Para confirmar el extremo de un arrendatario específico, solicita https://<tenant>-services.pm.beyondtrustcloud.com/oauth/.well-known/openid-configuration y lee token_endpoint.

   import hashlib
   import json
   import os
   import re
   import time
   import urllib3
   import uuid
   from datetime import datetime, timedelta, timezone
   from urllib.parse import urlencode
   
   import functions_framework
   from google.cloud import storage
   from google.cloud.exceptions import NotFound
   
   # Initialize HTTP client with timeouts
   http = urllib3.PoolManager(
       timeout=urllib3.Timeout(connect=5.0, read=60.0),
       retries=False,
   )
   
   # Initialize Storage client
   storage_client = storage.Client()
   
   TOKEN_PATH = '/oauth/connect/token'
   # Events/FromStartDate is the endpoint BeyondTrust documents for bulk extraction:
   # "A new API is exposed to extract the events in bulk." It takes only StartDate and
   # RecordSize. Events/search is not usable here: it requires an OperatingSystem value,
   # it is a single string with no documented "all" value, so it can only ever return one
   # operating system per call.
   EVENTS_PATH = '/management-api/v3/Events/FromStartDate'
   OAUTH_SCOPE = 'urn:management:api'
   # RecordSize accepts 1 to 1000.
   MAX_RECORD_SIZE = 1000
   
   class FetchError(Exception):
       """Raised when the BeyondTrust API call fails.
   
       The cursor must never advance on a failed fetch, otherwise the events in the
       failed window are skipped permanently.
       """
   
   def rfc3339(dt: datetime) -> str:
       """Render a datetime the way StartDate accepts it: milliseconds and a literal Z."""
       return dt.astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z'
   
   def parse_iso(stamp: str) -> datetime:
       """Parse any ISO 8601 timestamp shape the API emits.
   
       event.ingested comes back with a +00:00 offset and up to seven fractional
       digits, e.g. 2026-08-03T13:29:55.1109163+00:00. fromisoformat's tolerance for
       long fractions varies across Python versions, so the fraction is trimmed to
       microseconds first.
       """
       text = stamp.strip()
       if text.endswith('Z'):
           text = text[:-1] + '+00:00'
       text = re.sub(r'\.(\d{1,6})\d*', r'.\1', text, count=1)
       parsed = datetime.fromisoformat(text)
       if parsed.tzinfo is None:
           parsed = parsed.replace(tzinfo=timezone.utc)
       return parsed
   
   def canonical(stamp: str) -> str:
       """Re-render an API timestamp into the one form StartDate accepts.
   
       The API returns event.ingested as +00:00-offset ISO 8601 but rejects that
       same shape as a StartDate value with 400 "Invalid Start date format": it
       accepts only Z-suffixed values. Every timestamp that came from the API must
       pass through here before being sent back or persisted.
       """
       return rfc3339(parse_iso(stamp))
   
   def event_id(evt: dict) -> str:
       """Return the event's identity for deduplication.
   
       Falls back to a content hash when event.id is absent, so an id-less event
       still deduplicates instead of being re-ingested on every boundary re-read.
       """
       explicit = str((evt.get('event') or {}).get('id') or '')
       if explicit:
           return explicit
       digest = hashlib.sha256(
           json.dumps(evt, sort_keys=True, ensure_ascii=False).encode('utf-8')
       ).hexdigest()
       return f'sha256:{digest}'
   
   def event_ingested(evt: dict) -> str:
       """Return the Elastic ingestion timestamp, which is what StartDate filters on."""
       return str((evt.get('event') or {}).get('ingested') or '')
   
   def next_millisecond(stamp: str) -> str:
       """Return the canonical timestamp one millisecond later.
   
       Used only when a full batch fits inside a single ingestion millisecond. Without
       this the cursor cannot move and the collector stalls on that timestamp forever.
       """
       return rfc3339(parse_iso(stamp) + timedelta(milliseconds=1))
   
   @functions_framework.cloud_event
   def main(cloud_event):
       """Fetch BeyondTrust EPM events and write them to Cloud Storage as NDJSON.
   
       Args:
               cloud_event: CloudEvent object containing the Pub/Sub message.
       """
       bucket_name = os.environ.get('GCS_BUCKET')
       prefix = os.environ.get('GCS_PREFIX', 'beyondtrust-epm/')
       state_key = os.environ.get('STATE_KEY', 'beyondtrust-epm-state.json')
   
       api_url = (os.environ.get('BPT_API_URL') or '').rstrip('/')
       client_id = os.environ.get('CLIENT_ID')
       client_secret = os.environ.get('CLIENT_SECRET')
       # Clamp both ends: the API rejects RecordSize outside 1 to 1000, and a
       # misconfigured 0 would otherwise crash-loop on a 400 every run.
       record_size = min(max(int(os.environ.get('RECORD_SIZE', '1000')), 1), MAX_RECORD_SIZE)
       max_batches = int(os.environ.get('MAX_BATCHES', '50'))
       lookback_hours = int(os.environ.get('LOOKBACK_HOURS', '24'))
   
       if not all([bucket_name, api_url, client_id, client_secret]):
           raise RuntimeError(
               'Missing required environment variables: '
               'GCS_BUCKET, BPT_API_URL, CLIENT_ID, CLIENT_SECRET'
           )
   
       bucket = storage_client.bucket(bucket_name)
       state = load_state(bucket, state_key)
   
       # StartDate is a rising cursor on the Elastic ingestion timestamp, not a closed
       # window. There is no EndDate on this endpoint.
   
       if state.get('last_ingested'):
           # canonical() also repairs state written by the previous script revision,
           # which persisted the API's raw +00:00 form that StartDate rejects.
           try:
               start_date = canonical(state['last_ingested'])
           except (ValueError, TypeError, AttributeError) as e:
               raise RuntimeError(
                   f'Unparseable last_ingested in gs://{bucket_name}/{state_key}: '
                   f'{state["last_ingested"]!r}. Fix or delete that object; deleting '
                   f'restarts collection from the lookback window.'
               ) from e
       else:
           start_date = rfc3339(datetime.now(timezone.utc) - timedelta(hours=lookback_hours))
   
       seen_ids = set(state.get('seen_ids', []))
       print(f'Collecting events ingested from {start_date}')
   
       token = get_oauth_token(api_url, client_id, client_secret)
       fresh, cursor, drained = fetch_events(
           api_url, token, start_date, record_size, max_batches, seen_ids
       )
   
       if not fresh:
           print('No new events. Cursor left unchanged.')
           return
   
       # Everything that can fail is computed before the upload: a crash between the
       # upload and save_state replays the batch on the next run, so the window where
       # side effects exist without recorded state must stay minimal. The retained ids
       # share the cursor millisecond, because those are exactly the ones the
       # inclusive StartDate will return again.
       retained = {
           event_id(e) for e in fresh
           if event_ingested(e) and canonical(event_ingested(e)) == cursor
       }
       if cursor == start_date:
           # The cursor millisecond did not advance, so ids retained by earlier runs
           # are still on the boundary; dropping them would re-ingest their events.
           retained |= seen_ids
   
       timestamp = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
       # The random suffix keeps concurrent executions (Pub/Sub delivers at least
       # once) from overwriting each other's object within the same second.
       filename = (
           f'{prefix}beyondtrust-epm-events-{timestamp}-'
           f'{uuid.uuid4().hex[:8]}.ndjson'
       )
       ndjson = '\n'.join(json.dumps(e, ensure_ascii=False) for e in fresh) + '\n'
       bucket.blob(filename).upload_from_string(
           ndjson, content_type='application/x-ndjson'
       )
       print(f'Wrote {len(fresh)} events to gs://{bucket_name}/{filename}')
   
       save_state(bucket, state_key, {
           'last_ingested': cursor,
           'seen_ids': sorted(retained),
           'updated_at': rfc3339(datetime.now(timezone.utc)),
       })
   
       if not drained:
           print(
               f'Stopped after {max_batches} batches with more events available. '
               'The cursor advanced, so the next run continues from here.'
           )
   
   def fetch_events(api_url, token, start_date, record_size, max_batches, seen_ids):
       """Read forward from start_date until the API returns a short batch.
   
       The response envelope has no page count and no next-page token, so a batch
       shorter than record_size is the only documented end-of-data signal.
   
       StartDate is inclusive and the cursor lands on the newest event's timestamp, so
       every batch re-returns the events sharing it. Deduplication by event id therefore
       has to happen as batches arrive, not only between runs.
   
       Returns:
       Tuple of (new events, cursor to persist, whether the stream was drained).
   
       Raises:
       FetchError: on any API or transport failure, so the caller cannot mistake a
       failed fetch for an empty result and advance the cursor.
       """
       headers = {
           'Authorization': f'Bearer {token}',
           'Accept': 'application/json',
       }
   
       all_events = []
       seen = set(seen_ids)
       cursor = start_date
   
       for batch in range(1, max_batches + 1):
           query = urlencode({'StartDate': cursor, 'RecordSize': record_size})
           body = request_with_retry(f'{api_url}{EVENTS_PATH}?{query}', headers)
   
           events = body.get('events') or []
           new = [e for e in events if event_id(e) not in seen]
           seen.update(event_id(e) for e in events if event_id(e))
           all_events.extend(new)
           print(f'Batch {batch}: {len(events)} events, {len(new)} new')
   
           # Termination is judged on what the API returned, not on what survived
           # deduplication: a batch can be full and still be entirely duplicates.
           # The newest stamp in the batch, not events[-1]: ordering is a client-side
           # convention, not a documented guarantee, and a misordered tail would
           # regress the cursor and re-ingest events already written.
           stamps = [canonical(s) for s in (event_ingested(e) for e in events) if s]
   
           if len(events) < record_size:
               # Short batch: the stream is drained.
               if events and not stamps:
                   print(
                       'Warning: no event in the final batch carries event.ingested; '
                       'cursor left unchanged.'
                   )
               if stamps:
                   cursor = max(max(stamps), cursor)
               return all_events, cursor, True
   
           if not stamps:
               raise FetchError(
                   'No event in a full batch carries event.ingested, '
                   'so the cursor cannot advance'
               )
           batch_max = max(stamps)
           if batch_max <= cursor:
               # A full batch fits inside one ingestion millisecond. Stepping past it is the only
               # way to make progress; holding the cursor here stalls collection forever.
               # Events beyond record_size at that exact millisecond are unreachable, which
               # needs more than 1000 events in one millisecond.
               print(
                   f'Warning: a full batch shares ingestion timestamp {batch_max}; '
                   'stepping past it. Events beyond RecordSize at that timestamp are skipped.'
               )
               cursor = next_millisecond(cursor)
           else:
               cursor = batch_max
   
       return all_events, cursor, False
   
   def request_with_retry(url, headers, attempts=4):
       """GET with backoff on 429 and 5xx.
   
       The documented rate limit is 1000 requests per 100 seconds.
       """
       backoff = 1.0
       for attempt in range(1, attempts + 1):
           try:
               response = http.request('GET', url, headers=headers)
           except Exception as e:
               raise FetchError(f'Request to {url} failed: {e}') from e
   
           if response.status in (429, 500, 502, 503, 504) and attempt < attempts:
               retry_after = response.headers.get('Retry-After')
               try:
                   delay = int(retry_after) if retry_after else backoff
               except (TypeError, ValueError):
                   delay = backoff
               # Retry-After is server-controlled input: bound it so a bogus value
               # cannot sleep past the function timeout or crash time.sleep.
               delay = min(max(delay, 1.0), 60.0)
               print(f'HTTP {response.status}. Retrying in {delay}s...')
               time.sleep(delay)
               backoff = min(backoff * 2, 30.0)
               continue
   
           if response.status != 200:
               raise FetchError(
                   f'Request failed: {response.status} {response.data.decode("utf-8")}'
               )
   
           text = response.data.decode('utf-8')
           # BeyondTrust's sample folds "Owner" into "owner" before parsing, but that
           # workaround exists for PowerShell's case-insensitive ConvertFrom-Json.
           # Python parses case-sensitively, and the SecOps parser maps file.Owner.*
           # and file.owner to different UDM fields, so both keys must survive.
           try:
               return json.loads(text)
           except json.JSONDecodeError as e:
               raise FetchError(f'Malformed JSON response: {e}') from e
   
       raise FetchError(f'Giving up on {url} after {attempts} attempts')
   
   def get_oauth_token(api_url, client_id, client_secret):
       """Get an access token using the OAuth client credentials flow.
   
       The token is valid for one hour.
       """
       body = urlencode({
           'grant_type': 'client_credentials',
           'client_id': client_id,
           'client_secret': client_secret,
           'scope': OAUTH_SCOPE,
       })
       response = http.request(
           'POST',
           f'{api_url}{TOKEN_PATH}',
           body=body,
           headers={'Content-Type': 'application/x-www-form-urlencoded'},
       )
       if response.status != 200:
           raise FetchError(
               f'Token request failed: {response.status} '
               f'{response.data.decode("utf-8")}'
           )
       return json.loads(response.data.decode('utf-8'))['access_token']
   
   def load_state(bucket, key):
       """Read the collector state from Cloud Storage.
   
       Only a missing object is a cold start. Any other error is raised: swallowing it
       would reset collection to the lookback window and re-ingest that period.
       """
       blob = bucket.blob(key)
       try:
           return json.loads(blob.download_as_text())
       except NotFound:
           print('No state file found. Starting from the lookback window.')
           return {}
   
   def save_state(bucket, key, state):
       """Write the collector state to Cloud Storage.
   
       Failures are raised, not logged. A run that cannot record its cursor must fail,
       otherwise the next run repeats the same window.
       """
       bucket.blob(key).upload_from_string(
           json.dumps(state, indent=2), content_type='application/json'
       )
       print(f"Saved state: last_ingested={state.get('last_ingested')}")
   

   Nota: Este script se proporciona como ejemplo y no se admite oficialmente. Agradecemos tus comentarios sobre cómo podemos mejorarla. Para enviar comentarios, haz clic en Enviar comentarios > Comentarios sobre el producto de seguridad de Google SecOps en esta página.
   • 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).

   Nota: La configuración del activador de Pub/Sub crea automáticamente las suscripciones y los permisos necesarios.

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 la consola Google Cloud , ve a Cloud Scheduler.
2. Haz clic en Crear trabajo.

3. Proporciona los siguientes detalles de configuración:

   Parámetro de configuración	Valor
   Nombre	beyondtrust-epm-collector-hourly
   Región	Selecciona la misma región que la Cloud Run Function.
   Frecuencia	0 * * * * (cada hora en punto)
   Zona horaria	Selecciona la zona horaria (se recomienda UTC)
   Tipo de objetivo	Pub/Sub
   Tema	Selecciona el tema beyondtrust-epm-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 volumen y latencia de los 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 (recomendado)
Cada 6 horas	0 */6 * * *	Procesamiento por lotes de bajo volumen
Diario	0 0 * * *	Recopilación de datos históricos

Prueba el trabajo del programador

1. En la consola de Cloud Scheduler, busca tu trabajo.
2. Haz clic en Forzar ejecución para activar el DAG de forma manual.
3. Espera unos segundos y ve a Cloud Run > Servicios > beyondtrust-epm-collector > Registros.
4. Verifica que la función se haya ejecutado correctamente.
5. Revisa el bucket de Cloud Storage para confirmar que se escribieron los registros.

Recupera la cuenta de servicio de Google SecOps

Google SecOps usa una cuenta de servicio única para leer datos de tu bucket de Cloud Storage. 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 Add New Feed.
3. Haz clic en Configura un feed único.
4. En el campo Nombre del feed, ingresa un nombre para el feed (por ejemplo, BeyondTrust EPM logs).
5. Selecciona Google Cloud Storage V2 como el Tipo de fuente.
6. Selecciona BeyondTrust Endpoint Privilege Management como el Tipo de registro.

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

   [e-mail]
   

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

   Nota: Cada instancia de Google SecOps tiene una cuenta de servicio única. No uses cuentas de servicio de otros ejemplos o documentación.

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

La cuenta de servicio de Google SecOps necesita dos roles en tu bucket de Cloud Storage: Visualizador de objetos de Storage para leer los objetos de registro y un rol a nivel del bucket para leer los metadatos del bucket.

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 entidades: Pega el correo electrónico de la cuenta de servicio de Google SecOps.
   • Asignar roles: Selecciona las siguientes opciones:
   • Visualizador de objetos de Storage: Lee los objetos de registro.
   • Storage Legacy Bucket Reader: Lee los metadatos del bucket. Si seleccionaste la opción de borrado Borrar los archivos transferidos y los directorios vacíos, selecciona Storage Legacy Bucket Writer, que también otorga el permiso de borrado.

6. Haz clic en Guardar.

   Nota: Otorga ambos roles. Visualizador de objetos de Storage no otorga el permiso a nivel del bucket que necesita el feed, y los roles de Bucket heredado de Storage no otorgan permiso para leer el contenido de los objetos.

Configura un feed en Google SecOps para transferir registros de BeyondTrust EPM

1. Ve a Configuración de SIEM > Feeds.
2. Haz clic en Add New Feed.
3. Haz clic en Configura un feed único.
4. En el campo Nombre del feed, ingresa un nombre para el feed (por ejemplo, BeyondTrust EPM logs).
5. Selecciona Google Cloud Storage V2 como el Tipo de fuente.
6. Selecciona BeyondTrust Endpoint Privilege Management 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 Cloud Storage con la ruta de acceso del prefijo:

     gs://beyondtrust-epm-logs/beyondtrust-epm/
     

     • Reemplaza lo siguiente:

       • beyondtrust-epm-logs: Es el nombre de tu bucket de Cloud Storage.
       • beyondtrust-epm/: Es el prefijo o la ruta de carpeta opcional en la que se almacenan los registros (déjalo vacío para la raíz).

     • Ejemplos:

       • Bucket raíz: gs://beyondtrust-epm-logs/
       • Con prefijo: gs://beyondtrust-epm-logs/beyondtrust-epm/
     Nota: Siempre incluye la barra diagonal final (/) al final del URI.

   • Opción de eliminación del código fuente: Selecciona la opción de eliminación según tu preferencia:

     • No borrar archivos nunca: Nunca borra archivos de la fuente. Esto se recomienda para fines de prueba.

     • Borrar los archivos transferidos y los directorios vacíos: Borra los archivos y los directorios vacíos de la fuente después de que se complete una recuperación exitosa.

       Nota: Otorga ambos roles. Visualizador de objetos de Storage no otorga el permiso a nivel del bucket que necesita el feed, y los roles de Bucket heredado de Storage no otorgan permiso para leer el contenido de los objetos.

   • 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 transferencia: 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.

Tabla de asignación de UDM

Campo de registro	Asignación de UDM	Lógica
DomainIdentifier_label	additional.fields	Combinado
WatsonEventType_label	additional.fields	Combinado
action_label	additional.fields	Combinado
activity_id_label	additional.fields	Combinado
app_id_label	additional.fields	Combinado
app_insights_instrumentation_key_label	additional.fields	Combinado
asimov_instrumentation_key_label	additional.fields	Combinado
authorization_request_control_authorization_label	additional.fields	Combinado
bundle_name_label	additional.fields	Combinado
changedBy_label	additional.fields	Combinado
code_signature_subject_name_label	additional.fields	Combinado
collector_api_key_label	additional.fields	Combinado
command_label	additional.fields	Combinado
configuration_application_group_description_label	additional.fields	Combinado
configuration_application_group_identifier_label	additional.fields	Combinado
configuration_application_group_name_label	additional.fields	Combinado
configuration_application_identifier_label	additional.fields	Combinado
configuration_application_type_label	additional.fields	Combinado
configuration_identifier_label	additional.fields	Combinado
configuration_message_identifier_label	additional.fields	Combinado
configuration_message_name_label	additional.fields	Combinado
configuration_message_type_label	additional.fields	Combinado
configuration_name_label	additional.fields	Combinado
configuration_revision_number_label	additional.fields	Combinado
configuration_rule_identifier_label	additional.fields	Combinado
configuration_rule_on_demand_label	additional.fields	Combinado
configuration_rule_script_outcome_rule_affected_label	additional.fields	Combinado
configuration_token_identifier_label	additional.fields	Combinado
configuration_token_name_label	additional.fields	Combinado
configuration_workstyle_description_label	additional.fields	Combinado
configuration_workstyle_identifier_label	additional.fields	Combinado
configuration_workstyle_name_label	additional.fields	Combinado
content_length_label	additional.fields	Combinado
content_type_label	additional.fields	Combinado
domainNetBIOSName_label	additional.fields	Combinado
domain_label	additional.fields	Combinado
entity_label	additional.fields	Combinado
entity_name_label	additional.fields	Combinado
event_action_label	additional.fields	Combinado
exitstatus_label	additional.fields	Combinado
file_hash_md5_label	additional.fields	Combinado
file_hash_sha1_label	additional.fields	Combinado
file_hash_sha256_label	additional.fields	Combinado
file_version_label	additional.fields	Combinado
gid_label	additional.fields	Combinado
group_data_id_label	additional.fields	Combinado
group_id_label	additional.fields	Combinado
group_label	additional.fields	Combinado
handle_label	additional.fields	Combinado
host_name_label	additional.fields	Combinado
host_uptime_labels	additional.fields	Combinado
http_host_label	additional.fields	Combinado
id_label	additional.fields	Combinado
iolog_label	additional.fields	Combinado
is_opted_in_label	additional.fields	Combinado
linenum_label	additional.fields	Combinado
local_identifier_label	additional.fields	Combinado
locale_label	additional.fields	Combinado
master_utcoffset_label	additional.fields	Combinado
masterlocale_label	additional.fields	Combinado
owner_identifier_label	additional.fields	Combinado
parent_entity_id_label	additional.fields	Combinado
parent_process_exec_label	additional.fields	Combinado
parent_process_label	additional.fields	Combinado
pbmasterdnodename_label	additional.fields	Combinado
pipeName_label	additional.fields	Combinado
process_entity_id_label	additional.fields	Combinado
process_hash_label	additional.fields	Combinado
process_name_label	additional.fields	Combinado
process_parent_name_label	additional.fields	Combinado
process_start_time_label	additional.fields	Combinado
processexe_label	additional.fields	Combinado
product_label	additional.fields	Combinado
product_type_label	additional.fields	Combinado
product_version_label	additional.fields	Combinado
requestuser_label	additional.fields	Combinado
runargv_label	additional.fields	Combinado
runcwd_label	additional.fields	Combinado
runeffectivegroup_label	additional.fields	Combinado
runeffectiveuser_label	additional.fields	Combinado
runhost_label	additional.fields	Combinado
schema_version_label	additional.fields	Combinado
sku_name_label	additional.fields	Combinado
telemetry_level_label	additional.fields	Combinado
tenant_id_label	additional.fields	Combinado
type_label	additional.fields	Combinado
uid_label	additional.fields	Combinado
user_id_label	additional.fields	Combinado
user_name_label	additional.fields	Combinado
vs_exe_version_label	additional.fields	Combinado
inter_host	intermediary.hostname	Asignada directamente
Processes.description	metadata.description	Asignada directamente
details	metadata.description	Asignada directamente
event_data.reason	metadata.description	Asignada directamente
file.pe.description	metadata.description	Asignada directamente
created	metadata.event_timestamp	Se analizó como ISO8601
datetime	metadata.event_timestamp	Se analizó como MMM dd HH:mm:ss
has_principal	metadata.event_type	Asignado: true → STATUS_UPDATE
has_user	metadata.event_type	Asignado: true → USER_UNCATEGORIZED
parent_working_directory_label	metadata.ingestion_labels	Combinado
working_directory_label	metadata.ingestion_labels	Combinado
auditType	metadata.product_event_type	Asignada directamente
event_datas.ActionId	metadata.product_log_id	Asignada directamente
labels.related_item_id	metadata.product_log_id	Asignada directamente
uniqueid	metadata.product_log_id	Asignada directamente
masterdversion	metadata.product_version	Asignada directamente
headers.http_version	network.application_protocol_version	Asignada directamente
headers.request_method	network.http.method	Asignada directamente
host.os.platform	principal.administrative_domain	Asignada directamente
Processes.process	principal.application	Asignada directamente
agent_ephemeral_id_label	principal.asset.attribute.labels	Combinado
agent_id_label	principal.asset.attribute.labels	Combinado
agent_version_label	principal.asset.attribute.labels	Combinado
ecs_version_label	principal.asset.attribute.labels	Combinado
_hardware	principal.asset.hardware	Combinado
host.hostname	principal.asset.hostname	Asignada directamente
submithost	principal.asset.hostname	Asignada directamente
ip_address	principal.asset.ip	Combinado
masterhostip	principal.asset.ip	Combinado
submithostip	principal.asset.ip	Combinado
file.path	principal.file.full_path	Asignada directamente
lineinfile	principal.file.full_path	Asignada directamente
host.hostname	principal.hostname	Asignada directamente
submithost	principal.hostname	Asignada directamente
ip_address	principal.ip	Combinado
masterhostip	principal.ip	Combinado
submithostip	principal.ip	Combinado
mac	principal.mac	Combinado
host.os.name	principal.platform_version	Asignada directamente
host.os.version	principal.platform_version	Asignada directamente
process.command_line	principal.process.command_line	Asignada directamente
runcommand	principal.process.command_line	Asignada directamente
process.executable	principal.process.file.full_path	Asignada directamente
cmd	principal.process.parent_process.command_line	Asignada directamente
Processes.process_path	principal.process.parent_process.file.full_path	Asignada directamente
process.parent.executable	principal.process.parent_process.file.full_path	Asignada directamente
Processes.parent_process_id	principal.process.parent_process.pid	Asignada directamente
Processes.process_id	principal.process.pid	Asignada directamente
logpid	principal.process.pid	Asignada directamente
file.Owner.DomainName	principal.user.company_name	Asignada directamente
file.Owner.Name	principal.user.user_display_name	Asignada directamente
runuser	principal.user.user_display_name	Asignada directamente
userName	principal.user.user_display_name	Asignada directamente
Processes.user	principal.user.userid	Asignada directamente
userId	principal.user.userid	Asignada directamente
userid	principal.user.userid	Asignada directamente
Processes.user_id	principal.user.windows_sid	Asignada directamente
EPMWinMac.Configuration.Rule.Action	security_result.action	Combinado
security_result_action	security_result.action	Combinado
event_data.outcome	security_result.category_details	Combinado
host.os.version	security_result.category_details	Combinado
EPMWinMac.Configuration.Application.Description	security_result.description	Asignada directamente
EPMWinMac.Configuration.Message.Description	security_result.description	Asignada directamente
host.os.type	src.administrative_domain	Asignada directamente
file.name	src.file.names	Combinado
host.os.full	src.platform_version	Asignada directamente
host.os.family	target.administrative_domain	Asignada directamente
Processes.dest	target.asset.hostname	Asignada directamente
file.extension	target.file.mime_type	Asignada directamente
Processes.dest	target.hostname	Asignada directamente
host.domain	target.hostname	Asignada directamente
file_DriveType_label	target.resource.attribute.labels	Combinado
file_drive_letter_label	target.resource.attribute.labels	Combinado
owner_label	target.resource.attribute.labels	Combinado
N/A	metadata.event_type	Constante: USER_UNCATEGORIZED
N/A	metadata.product_name	Constante: Beyondtrust Privilege Management
N/A	metadata.vendor_name	Constante: Beyondtrust Privilege Management
N/A	network.application_protocol	Constante: HTTP
N/A	principal.platform	Constante: MAC

Registro de cambios

Consulta el registro de cambios de este analizador

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