Esta página se ha traducido con Cloud Translation API.

Recopilar registros de gestión de identidades y accesos de SailPoint

Disponible en:

SecOps de Google SIEM

En este documento se explica cómo ingerir registros de gestión de identidades y accesos (IAM) de SailPoint en Google Security Operations mediante Amazon S3.

Antes de empezar

Asegúrate de que cumples los siguientes requisitos previos:

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

Recopilar los requisitos previos de gestión de identidades y accesos de SailPoint (IDs, claves de API, IDs de organización y tokens)

Inicia sesión en la consola de administración de SailPoint Identity Security Cloud como administrador.
Ve a Global > Security Settings > API Management.
Haz clic en Crear cliente de API.
Elige Credenciales de cliente como tipo de autorización.
Proporcione los siguientes detalles de configuración:
- Nombre: introduce un nombre descriptivo (por ejemplo, Chronicle Export API).
- Descripción: introduce una descripción para el cliente de la API.
- Ámbitos: selecciona sp:scopes:all (o los ámbitos de lectura adecuados para los eventos de auditoría).
Haz clic en Crear y copia las credenciales de API generadas de forma segura.
Anota la URL base del cliente de SailPoint (por ejemplo, https://tenant.api.identitynow.com).
Copia y guarda en un lugar seguro los siguientes datos:
- IDN_CLIENT_ID
- IDN_CLIENT_SECRET
- IDN_BASE

Configurar un segmento de AWS S3 y IAM para Google SecOps

Crea un segmento de Amazon S3 siguiendo esta guía de usuario: Crear un segmento.
Guarda el nombre y la región del segmento para consultarlos más adelante (por ejemplo, sailpoint-iam-logs).
Crea un usuario siguiendo esta guía: Crear un usuario de gestión de identidades y accesos.
Selecciona el usuario creado.
Selecciona la pestaña Credenciales de seguridad.
En la sección Claves de acceso, haz clic en Crear clave de acceso.
Selecciona Servicio de terceros como Caso práctico.
Haz clic en Siguiente.
Opcional: añade una etiqueta de descripción.
Haz clic en Crear clave de acceso.
Haz clic en Descargar archivo CSV para guardar la clave de acceso y la clave de acceso secreta para usarlas más adelante.
Haz clic en Listo.
Selecciona la pestaña Permisos.
En la sección Políticas de permisos, haz clic en Añadir permisos.
Selecciona Añadir permisos.
Seleccione Adjuntar políticas directamente.
Busca y selecciona la política AmazonS3FullAccess.
Haz clic en Siguiente.
Haz clic en Añadir permisos.

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

En la consola de AWS, ve a IAM > Policies > Create policy > JSON tab.

Copia y pega la siguiente política:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowPutObjects",
      "Effect": "Allow",
      "Action": "s3:PutObject",
      "Resource": "arn:aws:s3:::sailpoint-iam-logs/*"
    },
    {
      "Sid": "AllowGetStateObject",
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::sailpoint-iam-logs/sailpoint/iam/state.json"
    }
  ]
}

Sustituye sailpoint-iam-logs si has introducido otro nombre de segmento:

Haz clic en Siguiente > Crear política.
Ve a IAM > Roles > Crear rol > Servicio de AWS > Lambda.
Adjunte la política que acaba de crear.
Dale el nombre SailPointIamToS3Role al rol y haz clic en Crear rol.

Crear la función Lambda

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

Ajuste Valor

Nombre sailpoint_iam_to_s3

Tiempo de ejecución Python 3.13

Arquitectura x86_64

Rol de ejecución SailPointIamToS3Role

Ajuste	Valor
Nombre	`sailpoint_iam_to_s3`
Tiempo de ejecución	Python 3.13
Arquitectura	x86_64
Rol de ejecución	`SailPointIamToS3Role`

Una vez creada la función, abra la pestaña Código, elimine el stub e introduzca el siguiente código (sailpoint_iam_to_s3.py):

#!/usr/bin/env python3
# Lambda: Pull SailPoint Identity Security Cloud audit events and store raw JSONL payloads to S3
# - Uses /v3/search API with pagination for audit events.
# - Preserves vendor-native JSON format for identity events.
# - Retries with exponential backoff; unique S3 keys to avoid overwrites.
# - Outputs JSONL format (one event per line) for optimal Chronicle ingestion.

import os, json, time, uuid, urllib.parse
from urllib.request import Request, urlopen
from urllib.error import URLError, HTTPError

import boto3

S3_BUCKET   = os.environ["S3_BUCKET"]
S3_PREFIX   = os.environ.get("S3_PREFIX", "sailpoint/iam/")
STATE_KEY   = os.environ.get("STATE_KEY", "sailpoint/iam/state.json")
WINDOW_SEC  = int(os.environ.get("WINDOW_SECONDS", "3600"))  # default 1h
HTTP_TIMEOUT= int(os.environ.get("HTTP_TIMEOUT", "60"))
IDN_BASE    = os.environ["IDN_BASE"]  # e.g. https://tenant.api.identitynow.com
CLIENT_ID   = os.environ["IDN_CLIENT_ID"]
CLIENT_SECRET = os.environ["IDN_CLIENT_SECRET"]
SCOPE       = os.environ.get("IDN_SCOPE", "sp:scopes:all")
PAGE_SIZE   = int(os.environ.get("PAGE_SIZE", "250"))
MAX_PAGES   = int(os.environ.get("MAX_PAGES", "20"))
MAX_RETRIES = int(os.environ.get("MAX_RETRIES", "3"))
USER_AGENT  = os.environ.get("USER_AGENT", "sailpoint-iam-to-s3/1.0")

s3 = boto3.client("s3")

def _load_state():
    try:
        obj = s3.get_object(Bucket=S3_BUCKET, Key=STATE_KEY)
        return json.loads(obj["Body"].read())
    except Exception:
        return {}

def _save_state(st):
    s3.put_object(
        Bucket=S3_BUCKET,
        Key=STATE_KEY,
        Body=json.dumps(st, separators=(",", ":")).encode("utf-8"),
        ContentType="application/json",
    )

def _iso(ts: float) -> str:
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(ts))

def _get_oauth_token() -> str:
    """Get OAuth2 access token using Client Credentials flow"""
    token_url = f"{IDN_BASE.rstrip('/')}/oauth/token"

    data = urllib.parse.urlencode({
        'grant_type': 'client_credentials',
        'client_id': CLIENT_ID,
        'client_secret': CLIENT_SECRET,
        'scope': SCOPE
    }).encode('utf-8')

    req = Request(token_url, data=data, method="POST")
    req.add_header("Content-Type", "application/x-www-form-urlencoded")
    req.add_header("User-Agent", USER_AGENT)

    with urlopen(req, timeout=HTTP_TIMEOUT) as r:
        response = json.loads(r.read())
        return response["access_token"]

def _search_events(access_token: str, created_from: str, search_after: list = None) -> list:
    """Search for audit events using SailPoint's /v3/search API

    IMPORTANT: SailPoint requires colons in ISO8601 timestamps to be escaped with backslashes.
    Example: 2024-01-15T10:30:00Z must be sent as 2024-01-15T10\:30\:00Z
    Reference: https://developer.sailpoint.com/discuss/t/datetime-searches/6609
    """
    search_url = f"{IDN_BASE.rstrip('/')}/v3/search"

    # Escape colons in timestamp for SailPoint search query
    # SailPoint requires: created:>=2024-01-15T10\:30\:00Z (colons must be escaped)
    escaped_timestamp = created_from.replace(":", "\\:")
    query_str = f'created:>={escaped_timestamp}'

    payload = {
        "indices": ["events"],
        "query": {"query": query_str},
        "sort": ["created", "+id"],
        "limit": PAGE_SIZE
    }

    if search_after:
        payload["searchAfter"] = search_after

    attempt = 0
    while True:
        req = Request(search_url, data=json.dumps(payload).encode('utf-8'), method="POST")
        req.add_header("Content-Type", "application/json")
        req.add_header("Accept", "application/json")
        req.add_header("Authorization", f"Bearer {access_token}")
        req.add_header("User-Agent", USER_AGENT)

        try:
            with urlopen(req, timeout=HTTP_TIMEOUT) as r:
                response = json.loads(r.read())
                # Handle different response formats
                if isinstance(response, list):
                    return response
                return response.get("results", response.get("data", []))
        except (HTTPError, URLError) as e:
            attempt += 1
            print(f"HTTP error on attempt {attempt}: {e}")
            if attempt > MAX_RETRIES:
                raise
            # exponential backoff with jitter
            time.sleep(min(60, 2 ** attempt) + (time.time() % 1))

def _put_events_data(events: list, from_ts: float, to_ts: float, page_num: int) -> str:
    """Write events to S3 in JSONL format (one JSON object per line)

    JSONL format is preferred for Chronicle ingestion as it allows:
    - Line-by-line processing
    - Better error recovery
    - Lower memory footprint
    """
    # Create unique S3 key for events data
    ts_path = time.strftime("%Y/%m/%d", time.gmtime(to_ts))
    uniq = f"{int(time.time()*1e6)}_{uuid.uuid4().hex[:8]}"
    key = f"{S3_PREFIX}{ts_path}/sailpoint_iam_{int(from_ts)}_{int(to_ts)}_p{page_num:03d}_{uniq}.jsonl"

    # Convert events list to JSONL format (one JSON object per line)
    jsonl_lines = [json.dumps(event, separators=(",", ":")) for event in events]
    jsonl_content = "\n".join(jsonl_lines)

    s3.put_object(
        Bucket=S3_BUCKET, 
        Key=key, 
        Body=jsonl_content.encode("utf-8"), 
        ContentType="application/x-ndjson",  # JSONL MIME type
        Metadata={
            'source': 'sailpoint-iam',
            'from_timestamp': str(int(from_ts)),
            'to_timestamp': str(int(to_ts)),
            'page_number': str(page_num),
            'events_count': str(len(events)),
            'format': 'jsonl'
        }
    )
    return key

def _get_item_id(item: dict) -> str:
    """Extract ID from event item, trying multiple possible fields"""
    for field in ("id", "uuid", "eventId", "_id"):
        if field in item and item[field]:
            return str(item[field])
    return ""

def lambda_handler(event=None, context=None):
    st = _load_state()
    now = time.time()
    from_ts = float(st.get("last_to_ts") or (now - WINDOW_SEC))
    to_ts = now

    # Get OAuth token
    access_token = _get_oauth_token()

    created_from = _iso(from_ts)
    print(f"Fetching SailPoint IAM events from: {created_from}")

    # Handle pagination state
    last_created = st.get("last_created")
    last_id = st.get("last_id")
    search_after = [last_created, last_id] if (last_created and last_id) else None

    pages = 0
    total_events = 0
    written_keys = []
    newest_created = last_created or created_from
    newest_id = last_id or ""

    while pages < MAX_PAGES:
        events = _search_events(access_token, created_from, search_after)

        if not events:
            break

        # Write page to S3 in JSONL format
        key = _put_events_data(events, from_ts, to_ts, pages + 1)
        written_keys.append(key)
        total_events += len(events)

        # Update pagination state from last item
        last_event = events[-1]
        last_event_created = last_event.get("created") or last_event.get("metadata", {}).get("created")
        last_event_id = _get_item_id(last_event)

        if last_event_created:
            newest_created = last_event_created
        if last_event_id:
            newest_id = last_event_id

        search_after = [newest_created, newest_id]
        pages += 1

        # If we got less than page size, we're done
        if len(events) < PAGE_SIZE:
            break

    print(f"Successfully retrieved {total_events} events across {pages} pages")

    # Save state for next run
    st["last_to_ts"] = to_ts
    st["last_created"] = newest_created
    st["last_id"] = newest_id
    st["last_successful_run"] = now
    _save_state(st)

    return {
        "statusCode": 200,
        "body": {
            "success": True,
            "pages": pages,
            "total_events": total_events,
            "s3_keys": written_keys,
            "from_timestamp": from_ts,
            "to_timestamp": to_ts,
            "last_created": newest_created,
            "last_id": newest_id,
            "format": "jsonl"
        }
    }

if __name__ == "__main__":
    print(lambda_handler())

Ve a Configuración > Variables de entorno > Editar > Añadir nueva variable de entorno.

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

Variables de entorno

Clave	Valor de ejemplo
`S3_BUCKET`	`sailpoint-iam-logs`
`S3_PREFIX`	`sailpoint/iam/`
`STATE_KEY`	`sailpoint/iam/state.json`
`WINDOW_SECONDS`	`3600`
`HTTP_TIMEOUT`	`60`
`MAX_RETRIES`	`3`
`USER_AGENT`	`sailpoint-iam-to-s3/1.0`
`IDN_BASE`	`https://tenant.api.identitynow.com`
`IDN_CLIENT_ID`	`your-client-id` (desde el paso 2)
`IDN_CLIENT_SECRET`	`your-client-secret` (desde el paso 2)
`IDN_SCOPE`	`sp:scopes:all`
`PAGE_SIZE`	`250`
`MAX_PAGES`	`20`

Una vez creada la función, permanece en su página (o abre Lambda > Funciones > tu-función).
Seleccione la pestaña Configuración.
En el panel Configuración general, haz clic en Editar.
Cambia Tiempo de espera a 5 minutos (300 segundos) y haz clic en Guardar.

Nota: El tiempo de espera es un límite superior estricto, lo que significa que, si la función sigue ejecutándose cuando el temporizador llegue a los 5 minutos, AWS Lambda la finalizará. Si la ejecución es más corta, se completará con normalidad y solo se te cobrará por el tiempo de ejecución real.

Crear una programación de EventBridge

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

Opcional: Crear un usuario y claves de gestión de identidades y accesos de solo lectura para Google SecOps

Ve a Consola de AWS > IAM > Usuarios > Añadir usuarios.
Haz clic en Add users (Añadir usuarios).
Proporcione los siguientes detalles de configuración:
- Usuario: introduce secops-reader.
- Tipo de acceso: selecciona Clave de acceso – Acceso programático.
Haz clic en Crear usuario.
Asigna una política de lectura mínima (personalizada): Usuarios > lector-secops > Permisos > Añadir permisos > Asignar políticas directamente > Crear política.

En el editor de JSON, introduce la siguiente política:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:GetObject"],
      "Resource": "arn:aws:s3:::sailpoint-iam-logs/*"
    },
    {
      "Effect": "Allow",
      "Action": ["s3:ListBucket"],
      "Resource": "arn:aws:s3:::sailpoint-iam-logs"
    }
  ]
}

Asigna el nombre secops-reader-policy.
Ve a Crear política > busca o selecciona > Siguiente > Añadir permisos.
Ve a Credenciales de seguridad > Claves de acceso > Crear clave de acceso.
Descarga el archivo CSV (estos valores se introducen en el feed).

Configurar un feed en Google SecOps para ingerir registros de gestión de identidades y accesos de SailPoint

Ve a Configuración de SIEM > Feeds.
Haz clic en + Añadir nuevo feed.
En la página siguiente, haga clic en Configurar un solo feed.
En el campo Nombre del feed, introduce un nombre para el feed (por ejemplo, SailPoint IAM logs).
Selecciona Amazon S3 V2 como Tipo de fuente.
Seleccione SailPoint IAM como Tipo de registro.
Haz clic en Siguiente.
Especifique los valores de los siguientes parámetros de entrada:
- URI de S3: s3://sailpoint-iam-logs/sailpoint/iam/
- Opciones de eliminación de la fuente: selecciona la opción de eliminación que prefieras.
- Antigüedad máxima de los archivos: 180 días de forma predeterminada.
- ID de clave de acceso: clave de acceso de usuario con acceso al bucket de S3.
- Clave de acceso secreta: clave secreta del usuario con acceso al bucket de S3.
- Espacio de nombres de recursos: el espacio de nombres de recursos.
- Etiquetas de ingestión: la etiqueta aplicada a los eventos de este feed.
Haz clic en Siguiente.
Revise la configuración de su nuevo feed en la pantalla Finalizar y, a continuación, haga clic en Enviar.

Tabla de asignación de UDM

Campo de registro	Asignación de UDM	Lógica
`action`	`metadata.description`	Valor del campo `action` del registro sin procesar.
`actor.name`	`principal.user.user_display_name`	Valor del campo `actor.name` del registro sin procesar.
`attributes.accountName`	`principal.user.group_identifiers`	Valor del campo `attributes.accountName` del registro sin procesar.
`attributes.appId`	`target.asset_id`	"ID de aplicación: " concatenado con el valor del campo `attributes.appId` del registro sin procesar.
`attributes.attributeName`	`additional.fields[0].value.string_value`	Valor del campo `attributes.attributeName` del registro sin procesar, colocado en un objeto `additional.fields`. La clave es "Nombre del atributo".
`attributes.attributeValue`	`additional.fields[1].value.string_value`	Valor del campo `attributes.attributeValue` del registro sin procesar, colocado en un objeto `additional.fields`. La clave es "Valor del atributo".
`attributes.cloudAppName`	`target.application`	Valor del campo `attributes.cloudAppName` del registro sin procesar.
`attributes.hostName`	`target.hostname`, `target.asset.hostname`	Valor del campo `attributes.hostName` del registro sin procesar.
`attributes.interface`	`additional.fields[2].value.string_value`	Valor del campo `attributes.interface` del registro sin procesar, colocado en un objeto `additional.fields`. La clave es "Interface".
`attributes.operation`	`security_result.action_details`	Valor del campo `attributes.operation` del registro sin procesar.
`attributes.previousValue`	`additional.fields[3].value.string_value`	Valor del campo `attributes.previousValue` del registro sin procesar, colocado en un objeto `additional.fields`. La clave se ha definido como "Previous Value".
`attributes.provisioningResult`	`security_result.detection_fields.value`	El valor del campo `attributes.provisioningResult` del registro sin procesar, colocado en un objeto `security_result.detection_fields`. La clave es "Provisioning Result".
`attributes.sourceId`	`principal.labels[0].value`	El valor del campo `attributes.sourceId` del registro sin procesar, colocado en un objeto `principal.labels`. La clave es "Source Id".
`attributes.sourceName`	`principal.labels[1].value`	El valor del campo `attributes.sourceName` del registro sin procesar, colocado en un objeto `principal.labels`. La clave es "Nombre de la fuente".
`auditClassName`	`metadata.product_event_type`	Valor del campo `auditClassName` del registro sin procesar.
`created`	`metadata.event_timestamp.seconds`, `metadata.event_timestamp.nanos`	Valor del campo `created` del registro sin procesar, convertido en marca de tiempo si `instant.epochSecond` no está presente.
`id`	`metadata.product_log_id`	Valor del campo `id` del registro sin procesar.
`instant.epochSecond`	`metadata.event_timestamp.seconds`	Valor del campo `instant.epochSecond` del registro sin procesar, que se usa para la marca de tiempo.
`ipAddress`	`principal.asset.ip`, `principal.ip`	Valor del campo `ipAddress` del registro sin procesar.
`interface`	`additional.fields[0].value.string_value`	Valor del campo `interface` del registro sin procesar, colocado en un objeto `additional.fields`. La clave es "interface".
`loggerName`	`intermediary.application`	Valor del campo `loggerName` del registro sin procesar.
`message`	`metadata.description`, `security_result.description`	Se usa para varios fines, como definir la descripción en los metadatos y security_result, y extraer contenido XML.
`name`	`security_result.description`	Valor del campo `name` del registro sin procesar.
`operation`	`target.resource.attribute.labels[0].value`, `metadata.product_event_type`	El valor del campo `operation` del registro sin procesar, colocado en un objeto `target.resource.attribute.labels`. La clave es "operation". También se usa para `metadata.product_event_type`.
`org`	`principal.administrative_domain`	Valor del campo `org` del registro sin procesar.
`pod`	`principal.location.name`	Valor del campo `pod` del registro sin procesar.
`referenceClass`	`additional.fields[1].value.string_value`	Valor del campo `referenceClass` del registro sin procesar, colocado en un objeto `additional.fields`. La clave se define como "referenceClass".
`referenceId`	`additional.fields[2].value.string_value`	Valor del campo `referenceId` del registro sin procesar, colocado en un objeto `additional.fields`. La clave es "referenceId".
`sailPointObjectName`	`additional.fields[3].value.string_value`	Valor del campo `sailPointObjectName` del registro sin procesar, colocado en un objeto `additional.fields`. La clave es "sailPointObjectName".
`serverHost`	`principal.hostname`, `principal.asset.hostname`	Valor del campo `serverHost` del registro sin procesar.
`stack`	`additional.fields[4].value.string_value`	Valor del campo `stack` del registro sin procesar, colocado en un objeto `additional.fields`. La clave se ha definido como "Pila".
`status`	`security_result.severity_details`	Valor del campo `status` del registro sin procesar.
`target`	`additional.fields[4].value.string_value`	Valor del campo `target` del registro sin procesar, colocado en un objeto `additional.fields`. La clave es "target".
`target.name`	`principal.user.userid`	Valor del campo `target.name` del registro sin procesar.
`technicalName`	`security_result.summary`	Valor del campo `technicalName` del registro sin procesar.
`thrown.cause.message`	`xml_body`, `detailed_message`	Valor del campo `thrown.cause.message` del registro sin procesar, que se usa para extraer contenido XML.
`thrown.message`	`xml_body`, `detailed_message`	Valor del campo `thrown.message` del registro sin procesar, que se usa para extraer contenido XML.
`trackingNumber`	`additional.fields[5].value.string_value`	Valor del campo `trackingNumber` del registro sin procesar, colocado en un objeto `additional.fields`. La clave es "Tracking Number".
`type`	`metadata.product_event_type`	Valor del campo `type` del registro sin procesar.
`_version`	`metadata.product_version`	Valor del campo `_version` del registro sin procesar.
N/A	`metadata.event_timestamp`	Derivado de los campos `instant.epochSecond` o `created`.
N/A	`metadata.event_type`	Se determina mediante la lógica del analizador en función de varios campos, como `has_principal_user`, `has_target_application`, `technicalName` y `action`. El valor predeterminado es "GENERIC_EVENT".
N/A	`metadata.log_type`	Se ha definido como "SAILPOINT_IAM".
N/A	`metadata.product_name`	Selecciona "IAM".
N/A	`metadata.vendor_name`	Se ha asignado el valor "SAILPOINT".
N/A	`extensions.auth.type`	Se define como "AUTHTYPE_UNSPECIFIED" en determinadas condiciones.
N/A	`target.resource.attribute.labels[0].key`	Asigna el valor "operation".

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