Recopila registros de MFA de HYPR

Se admite en los siguientes sistemas operativos:

En este documento, se explica cómo transferir registros de HYPR MFA a Google Security Operations con webhooks o Google Cloud Storage V2.

La MFA de HYPR es una solución de autenticación de varios factores sin contraseña que proporciona autenticación resistente al phishing con llaves de acceso FIDO2, datos biométricos y acceso iniciado desde dispositivos móviles. HYPR reemplaza las contraseñas tradicionales por criptografía segura de clave pública para eliminar los ataques basados en credenciales y, al mismo tiempo, optimizar la autenticación de usuarios en estaciones de trabajo, aplicaciones web y servicios en la nube.

Antes de comenzar

Asegúrate de cumplir con los siguientes requisitos previos:

  • Una instancia de Google SecOps
  • Acceso de administrador al Centro de control de HYPR
  • Comunícate con el equipo de asistencia de HYPR para habilitar los Custom Event Hooks para la aplicación de RP que deseas supervisar.

Diferencias en el método de recopilación

La MFA de HYPR admite dos métodos para enviar registros a Google Security Operations:

  • Webhook (recomendado): HYPR envía eventos en tiempo real a Operaciones de seguridad de Google a través de Custom Event Hooks. Este método proporciona la entrega inmediata de eventos y no requiere infraestructura adicional.
  • Google Cloud Storage: Los eventos de HYPR se recopilan a través de la API y se almacenan en GCS, y luego Google Security Operations los ingiere. Este método proporciona procesamiento por lotes y retención de datos históricos.

Elige el método que mejor se adapte a tus requisitos:

Función Webhook Google Cloud Storage
Latencia En tiempo real (segundos) Lote (de minutos a horas)
Infraestructura No se requiere ninguna acción. Proyecto de GCP con función de Cloud Run
Datos históricos Limitado a la transmisión del evento Retención completa en GCS
Complejidad de la configuración Sencillo Moderado
Costo Mínimo Costos de procesamiento y almacenamiento de GCP

Opción 1: Configura la integración del webhook

Crea un feed de webhook en Google SecOps

Crea el feed

  1. Ve a Configuración de SIEM > Feeds.
  2. Haz clic en Agregar feed nuevo.
  3. En la siguiente página, haz clic en Configurar un solo feed.
  4. En el campo Nombre del feed, ingresa un nombre para el feed (por ejemplo, HYPR MFA Events).
  5. Selecciona Webhook como el Tipo de origen.
  6. Selecciona HYPR MFA como el Tipo de registro.
  7. Haz clic en Siguiente.
  8. Especifica valores para los siguientes parámetros de entrada:
    • Delimitador de división (opcional): Déjalo vacío. Cada solicitud de webhook contiene un solo evento JSON.
    • 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.

Genera y guarda la clave secreta

Después de crear el feed, debes generar una clave secreta para la autenticación:

  1. En la página de detalles del feed, haz clic en Generar clave secreta.
  2. Un diálogo muestra la clave secreta.
  3. Copia y guarda la clave secreta de forma segura.

Obtén la URL del extremo del feed

  1. Ve a la pestaña Detalles del feed.
  2. En la sección Endpoint Information, copia la URL del extremo del feed.

  3. El formato de la URL es el siguiente:

    https://malachiteingestion-pa.googleapis.com/v2/unstructuredlogentries:batchCreate

    o

    https://<REGION>-malachiteingestion-pa.googleapis.com/v2/unstructuredlogentries:batchCreate

  4. Guarda esta URL para los próximos pasos.

  5. Haz clic en Listo.

Crea una clave de API de Google Cloud

Chronicle requiere una clave de API para la autenticación. Crea una clave de API restringida en la Google Cloud Console.

Crea la clave de API

  1. Ve a la página Credenciales de la consola de Google Cloud.
  2. Selecciona tu proyecto (el proyecto asociado con tu instancia de Chronicle).
  3. Haz clic en Crear credenciales > Clave de API.
  4. Se creará una clave de API y se mostrará en un diálogo.
  5. Haz clic en Editar clave de API para restringir la clave.

Restringe la clave de API

  1. En la página de configuración de clave de API, haz lo siguiente:
    • Nombre: Ingresa un nombre descriptivo (por ejemplo, Chronicle Webhook API Key).
  2. En Restricciones de API, haz lo siguiente:
    1. Selecciona Restringir clave.
    2. En el menú desplegable Seleccionar APIs, busca y selecciona Google SecOps API (o Chronicle API).
  3. Haz clic en Guardar.
  4. Copia el valor de la clave de API del campo Clave de API en la parte superior de la página.
  5. Guarda la clave de API de forma segura.

Configura el gancho de eventos personalizados de MFA de HYPR

Construye la URL del webhook con encabezados

HYPR admite encabezados personalizados para la autenticación. Usa el método de autenticación de encabezados para mejorar la seguridad.

  • URL del extremo (sin parámetros):

    <ENDPOINT_URL>

  • Encabezados:

    x-goog-chronicle-auth: <API_KEY>
x-chronicle-auth: <SECRET_KEY>
    • Reemplaza:
      • <ENDPOINT_URL>: Es la URL del extremo del feed del paso anterior.
      • <API_KEY>: Es la clave de API de Google Cloud que creaste.
      • <SECRET_KEY>: Es la clave secreta de la creación del feed de Chronicle.

Prepara la configuración JSON del gancho de eventos personalizados

  • Los hooks de eventos personalizados de HYPR se configuran con JSON. Prepara la siguiente configuración JSON y reemplaza los valores de marcador de posición:

    {
  "name": "Chronicle SIEM Integration",
  "eventType": "ALL",
  "invocationEndpoint": "<ENDPOINT_URL>",
  "httpMethod": "POST",
  "authType": "API_KEY",
  "authParams": {
    "apiKeyAuthParameters": {
      "apiKeyName": "x-goog-chronicle-auth",
      "apiKeyValue": "<API_KEY>"
    },
    "invocationHttpParameters": {
      "headerParameters": [
        {
          "key": "Content-Type",
          "value": "application/json",
          "isValueSecret": false
        },
        {
          "key": "x-chronicle-auth",
          "value": "<SECRET_KEY>",
          "isValueSecret": true
        }
      ]
    }
  }
}

    • Reemplaza lo siguiente:

      • <ENDPOINT_URL>: Es la URL del extremo del feed de Chronicle.
      • <API_KEY>: Es la clave de la API de Google Cloud.
      • <SECRET_KEY>: Es la clave secreta de Chronicle.

    • Parámetros de configuración:

    • name: Es un nombre descriptivo para el gancho de eventos (por ejemplo, Chronicle SIEM Integration).

    • eventType: Establece el valor en ALL para enviar todos los eventos de HYPR o especifica etiquetas de eventos específicos, como AUTHENTICATION, REGISTRATION o ACCESS_TOKEN.

    • invocationEndpoint: Es la URL del extremo del feed de Chronicle.

    • httpMethod: Se configura como POST.

    • authType: Se establece en API_KEY para la autenticación de la clave de API.

    • apiKeyName: Es el nombre del encabezado de la clave de API (x-goog-chronicle-auth).

    • apiKeyValue: Es el valor de la clave de API de Google Cloud.

    • headerParameters: Encabezados adicionales, incluido Content-Type: application/json y la clave secreta de Chronicle en el encabezado x-chronicle-auth.

Crea el gancho de evento personalizado en el Centro de control de HYPR

  1. Accede al Centro de control de HYPR como administrador.
  2. En el menú de navegación de la izquierda, haz clic en Integraciones.
  3. En la página Integraciones, haz clic en Agregar nueva integración.
  4. El Centro de control de HYPR muestra las integraciones disponibles.
  5. Haz clic en la tarjeta de Eventos personalizados en Event Hooks.
  6. Haz clic en Agregar nuevo hook de evento.
  7. En el diálogo Add New Event Hook, pega el contenido JSON que preparaste en el campo de texto.
  8. Haz clic en Add Event Hook.
  9. El Centro de control de HYPR vuelve a la página Event Hooks.

El gancho de eventos personalizados ahora está configurado y comenzará a enviar eventos a Google SecOps.

Verifica que el webhook funcione

Verifica el estado del gancho de eventos del Centro de control de HYPR

  1. Accede al Centro de control de HYPR.
  2. Ve a Integraciones.
  3. Haz clic en la integración de Eventos personalizados.
  4. En la tabla Event Hooks, verifica que aparezca tu hook de eventos.
  5. Haz clic en el nombre del gancho del evento para ver los detalles.
  6. Verifica que la configuración coincida con tus parámetros.

Verifica el estado del feed de Chronicle

  1. Ve a Configuración de SIEM > Feeds en Chronicle.
  2. Ubica tu feed de webhook.
  3. Verifica la columna Estado (debe ser Activo).
  4. Verifica el recuento de Eventos recibidos (debe aumentar).
  5. Verifica la marca de tiempo de Last succeeded on (debe ser reciente).

Verifica los registros en Chronicle

  1. Ve a Búsqueda > Búsqueda de UDM.

  2. Usa la siguiente consulta:

    metadata.vendor_name = "HYPR" AND metadata.product_name = "MFA"

  3. Ajusta el intervalo de tiempo a la última hora.

  4. Verifica que los eventos aparezcan en los resultados.

Referencia de métodos de autenticación

Los hooks de eventos personalizados de HYPR admiten varios métodos de autenticación. El método recomendado para Chronicle es la autenticación con clave de API y encabezados personalizados.

  • Configuración:

    {
  "authType": "API_KEY",
  "authParams": {
    "apiKeyAuthParameters": {
      "apiKeyName": "x-goog-chronicle-auth",
      "apiKeyValue": "<API_KEY>"
    },
    "invocationHttpParameters": {
      "headerParameters": [
        {
          "key": "Content-Type",
          "value": "application/json",
          "isValueSecret": false
        },
        {
          "key": "x-chronicle-auth",
          "value": "<SECRET_KEY>",
          "isValueSecret": true
        }
      ]
    }
  }
}

  • Ventajas:

    • La clave y el secreto de la API se envían en encabezados (más seguros que los parámetros de URL).
    • Admite varios encabezados de autenticación.
    • Los encabezados no se registran en los registros de acceso del servidor web.

Autenticación básica

  • Configuración:

    {
  "authType": "BASIC",
  "authParams": {
    "basicAuthParameters": {
      "username": "your-username",
      "password": "your-password"
    },
    "invocationHttpParameters": {
      "headerParameters": [
        {
          "key": "Content-Type",
          "value": "application/json",
          "isValueSecret": false
        }
      ]
    }
  }
}
    • Caso de uso: Cuando el sistema de destino requiere autenticación HTTP básica.

Credenciales de cliente de OAuth 2.0

  • Configuración:

    {
  "authType": "OAUTH_CLIENT_CREDENTIALS",
  "authParams": {
    "oauthParameters": {
      "clientParameters": {
        "clientId": "your-client-id",
        "clientSecret": "your-client-secret"
      },
      "authorizationEndpoint": "https://login.example.com/oauth2/v2.0/token",
      "httpMethod": "POST",
      "oauthHttpParameters": {
        "bodyParameters": [
          {
            "key": "scope",
            "value": "api://your-api/.default",
            "isValueSecret": false
          },
          {
            "key": "grant_type",
            "value": "client_credentials",
            "isValueSecret": false
          }
        ]
      }
    },
    "invocationHttpParameters": {
      "headerParameters": [
        {
          "key": "Content-Type",
          "value": "application/json",
          "isValueSecret": false
        }
      ]
    }
  }
}
    • Caso de uso: Cuando el sistema de destino requiere autenticación de OAuth 2.0

Tipos de eventos y filtrado

Los eventos de HYPR se agrupan con el parámetro eventTags. Puedes configurar el gancho de eventos personalizados para que envíe todos los eventos o filtrar por tipos de eventos específicos.

Etiquetas de eventos

  • AUTHENTICATION: Eventos de autenticación del usuario (acceso, desbloqueo).
  • REGISTRATION: Eventos de registro de dispositivos (vinculación de dispositivos móviles, llaves de seguridad).
  • ACCESS_TOKEN: Eventos de generación y uso de tokens de acceso.
  • AUDIT: Eventos de registro de auditoría (acciones administrativas, cambios de configuración).

Configura el filtrado de eventos

Para enviar solo tipos de eventos específicos, modifica el parámetro eventType en la configuración JSON:

  • Enviar todos los eventos:

    {
  "eventType": "ALL"
}

  • Enviar solo eventos de autenticación:

    {
  "eventType": "AUTHENTICATION"
}

  • Envía solo eventos de registro:

    {
  "eventType": "REGISTRATION"
}

Opción 2: Configura la integración de Google Cloud Storage

Requisitos previos adicionales para la integración de GCS

Además de los requisitos previos que se indican en la sección "Antes de comenzar", necesitarás lo siguiente:

  • 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
  • Credenciales de la API de HYPR (comunícate con el equipo de asistencia de HYPR para obtener acceso a la API)

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, hypr-mfa-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.

Recopila credenciales de la API de HYPR

Comunícate con el equipo de asistencia de HYPR para obtener credenciales de la API y acceder a los datos de eventos de HYPR. Necesitarás:

  • URL base de la API: Es la URL de tu instancia de HYPR (por ejemplo, https://your-tenant.hypr.com).
  • Token de API: Es el token de autenticación para el acceso a la API.
  • ID de la app de RP: ID de la aplicación del usuario de confianza que se supervisará

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 hypr-logs-collector-sa.
    • Descripción de la cuenta de servicio: Ingresa Service account for Cloud Run function to collect HYPR MFA 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 (hypr-logs-collector-sa) en el bucket de GCS:

  1. Ve a Cloud Storage > Buckets.
  2. Haz clic en el nombre de tu bucket (por ejemplo, hypr-mfa-logs).
  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, hypr-logs-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 hypr-logs-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 activará con mensajes de Pub/Sub de Cloud Scheduler para recuperar registros de la API de HYPR 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 hypr-logs-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 (hypr-logs-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 (hypr-logs-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 hypr-mfa-logs Nombre del bucket de GCS
    GCS_PREFIX hypr-events Prefijo para los archivos de registro
    STATE_KEY hypr-events/state.json Ruta de acceso al archivo de estado
    HYPR_API_URL https://your-tenant.hypr.com URL base de la API de HYPR
    HYPR_API_TOKEN your-api-token Token de autenticación de la API de HYPR
    HYPR_RP_APP_ID your-rp-app-id ID de la aplicación de HYPR RP
    MAX_RECORDS 1000 Cantidad máxima de registros por ejecución
    PAGE_SIZE 100 Registros por página
    LOOKBACK_HOURS 24 Período de visualización inicial

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

    • 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', 'hypr-events')
STATE_KEY = os.environ.get('STATE_KEY', 'hypr-events/state.json')
HYPR_API_URL = os.environ.get('HYPR_API_URL')
HYPR_API_TOKEN = os.environ.get('HYPR_API_TOKEN')
HYPR_RP_APP_ID = os.environ.get('HYPR_RP_APP_ID')
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '1000'))
PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '100'))
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))

def to_unix_millis(dt: datetime) -> int:
    """Convert datetime to Unix epoch milliseconds."""
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    dt = dt.astimezone(timezone.utc)
    return int(dt.timestamp() * 1000)

def parse_datetime(value: str) -> datetime:
    """Parse ISO datetime string to datetime object."""
    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 HYPR MFA logs and write to GCS.

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

    if not all([GCS_BUCKET, HYPR_API_URL, HYPR_API_TOKEN, HYPR_RP_APP_ID]):
        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=LOOKBACK_HOURS)

        print(f"Fetching logs from {last_time.isoformat()} to {now.isoformat()}")

        # Convert to Unix milliseconds for HYPR API
        start_millis = to_unix_millis(last_time)
        end_millis = to_unix_millis(now)

        # Fetch logs
        records, newest_event_time = fetch_logs(
            api_url=HYPR_API_URL,
            api_token=HYPR_API_TOKEN,
            rp_app_id=HYPR_RP_APP_ID,
            start_time_ms=start_millis,
            end_time_ms=end_millis,
            page_size=PAGE_SIZE,
            max_records=MAX_RECORDS,
        )

        if not records:
            print("No new log records found.")
            save_state(bucket, STATE_KEY, now.isoformat())
            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.isoformat())

        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_iso: str):
    """Save the last event timestamp to GCS state file."""
    try:
        state = {'last_event_time': last_event_time_iso}
        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_iso}")
    except Exception as e:
        print(f"Warning: Could not save state: {e}")

def fetch_logs(api_url: str, api_token: str, rp_app_id: str, start_time_ms: int, end_time_ms: int, page_size: int, max_records: int):
    """
    Fetch logs from HYPR API with pagination and rate limiting.

    Args:
        api_url: HYPR API base URL
        api_token: HYPR API authentication token
        rp_app_id: HYPR RP application ID
        start_time_ms: Start time in Unix milliseconds
        end_time_ms: End time in Unix milliseconds
        page_size: Number of records per page
        max_records: Maximum total records to fetch

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

    endpoint = f"{base_url}/rp/api/versioned/events"

    # Bearer token authentication
    headers = {
        'Authorization': f'Bearer {api_token}',
        'Accept': 'application/json',
        'Content-Type': 'application/json',
        'User-Agent': 'GoogleSecOps-HYPRCollector/1.0'
    }

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

    # Offset-based pagination
    start_index = 0

    while True:
        page_num += 1

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

        # Build request parameters
        params = []
        params.append(f"rpAppId={rp_app_id}")
        params.append(f"startDate={start_time_ms}")
        params.append(f"endDate={end_time_ms}")
        params.append(f"start={start_index}")
        params.append(f"limit={min(page_size, max_records - len(records))}")
        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'))

            # Extract results
            page_results = data.get('data', [])

            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:
                    # HYPR uses LOGGEDTIMEINUTC field with Unix milliseconds
                    event_time_ms = event.get('LOGGEDTIMEINUTC')
                    if event_time_ms:
                        event_dt = datetime.fromtimestamp(event_time_ms / 1000, tz=timezone.utc)
                        event_time = event_dt.isoformat()
                        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
            current_size = data.get('size', 0)
            if current_size < page_size:
                print(f"Reached last page (size={current_size} < limit={page_size})")
                break

            start_index += current_size

        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 publicará mensajes en el tema de Pub/Sub (hypr-logs-trigger) a intervalos regulares, lo que activará 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 hypr-logs-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 destino Pub/Sub
    Tema Selecciona el tema de Pub/Sub (hypr-logs-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 (recomendado)
Cada 6 horas 0 */6 * * * Procesamiento por lotes de bajo volumen
Diario 0 0 * * * Recopilación de datos históricos

Prueba la integración

  1. En la consola de Cloud Scheduler, busca tu trabajo (hypr-logs-collector-hourly).
  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 (hypr-logs-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-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
Page 1: Retrieved X events
Wrote X records to gs://bucket-name/prefix/logs_YYYYMMDD_HHMMSS.ndjson
Successfully processed X records

  8. Ve a Cloud Storage > Buckets.

  9. Haz clic en el nombre de tu bucket (por ejemplo, hypr-mfa-logs).

  10. Navega a la carpeta de prefijo (por ejemplo, hypr-events/).

  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 el token de la API de HYPR tenga los permisos necesarios y que el ID de la app de RP sea correcto
  • HTTP 429: Limitación de frecuencia. La función volverá a intentarlo automáticamente con una espera.
  • 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.

Configura un feed en Google SecOps para transferir registros de MFA de HYPR

  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, HYPR MFA Logs from GCS).
  5. Selecciona Google Cloud Storage V2 como el Tipo de fuente.

  6. Selecciona HYPR MFA 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:

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

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

  9. Haz clic en Siguiente.

  10. 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://hypr-mfa-logs/hypr-events/
      • Reemplaza:
        • hypr-mfa-logs: Es el nombre de tu bucket de GCS.
        • hypr-events: Es el prefijo o la ruta de carpeta opcionales en los que se almacenan los registros (déjalo vacío para la raíz).

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

  11. Haz clic en Siguiente.

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

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 (por ejemplo, hypr-mfa-logs).
  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.

Tabla de asignación de UDM

Campo de registro Asignación de UDM Lógica
extensions.auth.type Tipo de autenticación (p.ej., SSO, MFA)
metadata.event_type Tipo de evento (p.ej., USER_LOGIN, NETWORK_CONNECTION)
EVENTNAME metadata.product_event_type Tipo de evento específico del producto
ID metadata.product_log_id ID de registro específico del producto
USERAGENT network.http.parsed_user_agent Usuario-agente HTTP analizado
USERAGENT network.http.user_agent Cadena de usuario-agente HTTP
SESSIONID network.session_id ID de sesión
DEVICEMODEL principal.asset.hardware.model Modelo de hardware del activo
COMPANION,MACHINEDOMAIN principal.asset.hostname Nombre de host del activo
REMOTEIP principal.asset.ip Dirección IP del activo
DEVICEID principal.asset_id Identificador único del activo
COMPANION,MACHINEDOMAIN principal.hostname Nombre de host asociado con la principal
REMOTEIP principal.ip Dirección IP asociada con la entidad principal
DEVICEOS principal.platform Plataforma (p.ej., WINDOWS Y LINUX
DEVICEOSVERSION principal.platform_version Versión de la plataforma
ISSUCCESSFUL security_result.action Acción que tomó el sistema de seguridad (p.ej., PERMITIR, BLOQUEAR)
MENSAJE security_result.description Descripción del resultado de seguridad
MACHINEUSERNAME target.user.user_display_name Nombre visible del usuario
FIDOUSER target.user.userid ID de usuario
metadata.product_name Nombre del producto
metadata.vendor_name Nombre del proveedor o la empresa

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