Recoger registros de URLScan IO

Disponible en:

En este documento se explica cómo ingerir registros de URLScan.io 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 con privilegios al arrendatario de URLScan IO
  • Acceso con privilegios a AWS (S3, IAM, Lambda y EventBridge)

Obtener los requisitos previos de URLScan.io

  1. Inicia sesión en URLScan IO.
  2. Haz clic en el icono de tu perfil.
  3. Selecciona Clave de API en el menú.
  4. Si aún no tienes una clave de API, sigue estos pasos:
    • Haz clic en el botón Crear clave de API.
    • Introduce una descripción para la clave de API (por ejemplo, Google SecOps Integration).
    • Selecciona los permisos de la clave (para obtener acceso de solo lectura, selecciona los permisos Leer).
    • Haz clic en Generar clave de API.
  5. Copia y guarda en un lugar seguro los siguientes detalles:
    • API_KEY la cadena de clave de API generada (formato: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx)
    • URL base de la API: https://urlscan.io/api/v1 (es constante para todos los usuarios)
  6. Anota los límites de tu cuota de API:
    • Cuentas gratuitas: tienen un límite de 1000 llamadas a la API al día y 60 por minuto.
    • Cuentas Pro: límites más altos en función del nivel de suscripción
  7. Si necesitas restringir las búsquedas solo a las exploraciones de tu organización, anota lo siguiente:
    • Identificador de usuario: tu nombre de usuario o correo electrónico (para usarlo con el filtro de búsqueda user:)
    • Identificador de equipo: si se usa la función de equipos (para usarla con el filtro de búsqueda team:)

Configurar un segmento de AWS S3 y IAM para Google SecOps

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

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

  1. En la consola de AWS, ve a IAM > Políticas.
  2. Haz clic en Crear política > pestaña JSON.
  3. Introduce la siguiente política:

    {
      "Version": "2012-10-17",
      "Statement": [
        {
          "Sid": "AllowPutObjects",
          "Effect": "Allow",
          "Action": "s3:PutObject",
          "Resource": "arn:aws:s3:::urlscan-logs-bucket/*"
        },
        {
          "Sid": "AllowGetStateObject",
          "Effect": "Allow",
          "Action": "s3:GetObject",
          "Resource": "arn:aws:s3:::urlscan-logs-bucket/urlscan/state.json"
        }
      ]
    }
    
    • Sustituye urlscan-logs-bucket si has introducido otro nombre de segmento.
  4. Haz clic en Siguiente > Crear política.

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

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

  7. Dale el nombre urlscan-lambda-role al rol y haz clic en Crear rol.

Crear la función Lambda

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

    Ajuste Valor
    Nombre urlscan-collector
    Tiempo de ejecución Python 3.13
    Arquitectura x86_64
    Rol de ejecución urlscan-lambda-role
  4. Una vez creada la función, abra la pestaña Código, elimine el stub e introduzca el siguiente código (urlscan-collector.py):

    import json
    import os
    import boto3
    from datetime import datetime, timedelta
    import urllib3
    import base64
    
    s3 = boto3.client('s3')
    http = urllib3.PoolManager()
    
    def lambda_handler(event, context):
        # Environment variables
        bucket = os.environ['S3_BUCKET']
        prefix = os.environ['S3_PREFIX']
        state_key = os.environ['STATE_KEY']
        api_key = os.environ['API_KEY']
        api_base = os.environ['API_BASE']
        search_query = os.environ.get('SEARCH_QUERY', 'date:>now-1h')
        page_size = int(os.environ.get('PAGE_SIZE', '100'))
        max_pages = int(os.environ.get('MAX_PAGES', '10'))
    
        # Load state
        state = load_state(bucket, state_key)
        last_run = state.get('last_run')
    
        # Prepare search query
        if last_run:
            # Adjust search query based on last run
            search_time = datetime.fromisoformat(last_run)
            time_diff = datetime.utcnow() - search_time
            hours = int(time_diff.total_seconds() / 3600) + 1
            search_query = f'date:>now-{hours}h'
    
        # Search for scans
        headers = {'API-Key': api_key}
        all_results = []
    
        for page in range(max_pages):
            search_url = f"{api_base}/search/"
            params = {
                'q': search_query,
                'size': page_size,
                'offset': page * page_size
            }
    
            # Make search request
            response = http.request(
                'GET',
                search_url,
                fields=params,
                headers=headers
            )
    
            if response.status != 200:
                print(f"Search failed: {response.status}")
                break
    
            search_data = json.loads(response.data.decode('utf-8'))
            results = search_data.get('results', [])
    
            if not results:
                break
    
            # Fetch full result for each scan
            for result in results:
                uuid = result.get('task', {}).get('uuid')
                if uuid:
                    result_url = f"{api_base}/result/{uuid}/"
                    result_response = http.request(
                        'GET',
                        result_url,
                        headers=headers
                    )
    
                    if result_response.status == 200:
                        full_result = json.loads(result_response.data.decode('utf-8'))
                        all_results.append(full_result)
                    else:
                        print(f"Failed to fetch result for {uuid}: {result_response.status}")
    
            # Check if we have more pages
            if len(results) < page_size:
                break
    
        # Write results to S3
        if all_results:
            now = datetime.utcnow()
            file_key = f"{prefix}year={now.year}/month={now.month:02d}/day={now.day:02d}/hour={now.hour:02d}/urlscan_{now.strftime('%Y%m%d_%H%M%S')}.json"
    
            # Create NDJSON content
            ndjson_content = '\n'.join([json.dumps(r, separators=(',', ':')) for r in all_results])
    
            # Upload to S3
            s3.put_object(
                Bucket=bucket,
                Key=file_key,
                Body=ndjson_content.encode('utf-8'),
                ContentType='application/x-ndjson'
            )
    
            print(f"Uploaded {len(all_results)} results to s3://{bucket}/{file_key}")
    
        # Update state
        state['last_run'] = datetime.utcnow().isoformat()
        save_state(bucket, state_key, state)
    
        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': f'Processed {len(all_results)} scan results',
                'location': f"s3://{bucket}/{prefix}"
            })
        }
    
    def load_state(bucket, key):
        try:
            response = s3.get_object(Bucket=bucket, Key=key)
            return json.loads(response['Body'].read())
        except s3.exceptions.NoSuchKey:
            return {}
        except Exception as e:
            print(f"Error loading state: {e}")
            return {}
    
    def save_state(bucket, key, state):
        try:
            s3.put_object(
                Bucket=bucket,
                Key=key,
                Body=json.dumps(state),
                ContentType='application/json'
            )
        except Exception as e:
            print(f"Error saving state: {e}")
    
  5. Vaya a Configuración > Variables de entorno.

  6. Haz clic en Editar > Añadir nueva variable de entorno.

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

    Clave Valor de ejemplo
    S3_BUCKET urlscan-logs-bucket
    S3_PREFIX urlscan/
    STATE_KEY urlscan/state.json
    API_KEY <your-api-key>
    API_BASE https://urlscan.io/api/v1
    SEARCH_QUERY date:>now-1h
    PAGE_SIZE 100
    MAX_PAGES 10
  8. Una vez creada la función, permanece en su página (o abre Lambda > Funciones > tu-función).

  9. Seleccione la pestaña Configuración.

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

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

Crear una programación de EventBridge

  1. Ve a Amazon EventBridge > Scheduler > Create schedule (Amazon EventBridge > Programador > Crear programación).
  2. Proporcione los siguientes detalles de configuración:
    • Programación periódica: Precio (1 hour).
    • Destino: tu función Lambda urlscan-collector.
    • Nombre: urlscan-collector-1h.
  3. 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

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

    {
      "Version": "2012-10-17",
      "Statement": [
        {
          "Effect": "Allow",
          "Action": ["s3:GetObject"],
          "Resource": "arn:aws:s3:::urlscan-logs-bucket/*"
        },
        {
          "Effect": "Allow",
          "Action": ["s3:ListBucket"],
          "Resource": "arn:aws:s3:::urlscan-logs-bucket"
        }
      ]
    }
    
  7. Asigna el nombre secops-reader-policy.

  8. Ve a Crear política > busca o selecciona > Siguiente > Añadir permisos.

  9. Ve a Credenciales de seguridad > Claves de acceso > Crear clave de acceso.

  10. Descarga el archivo CSV (estos valores se introducen en el feed).

Configurar un feed en Google SecOps para ingerir registros de URLScan.io

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

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