Recopilar registros de Censys

En este documento se explica cómo ingerir registros de Censys en Google Security Operations mediante Amazon S3. Censys ofrece una gestión integral de la superficie de ataque e inteligencia de Internet a través de su API. Esta integración te permite recoger eventos de descubrimiento de hosts, eventos de riesgo y cambios en los recursos de Censys ASM y reenviarlos a Google SecOps para analizarlos y monitorizarlos. El analizador transforma los registros sin procesar en un formato estructurado conforme al UDM de Google SecOps. Extrae campos del mensaje de registro sin procesar, convierte los tipos de datos y asigna la información extraída a los campos de UDM correspondientes, lo que enriquece los datos con contexto y etiquetas adicionales.

Antes de empezar

Asegúrate de que cumples los siguientes requisitos previos:

• Instancia de Google SecOps
• Acceso privilegiado a Censys ASM
• Acceso privilegiado a AWS (S3, IAM, Lambda y EventBridge)

Recoger los requisitos previos de Censys (credenciales de API)

1. Inicia sesión en la consola de Censys ASM en app.censys.io.
2. Ve a Integraciones, en la parte superior de la página.
3. Copia y guarda los valores de API Key (Clave de API) y Organization ID (ID de organización).
4. Anota la URL base de la API: https://api.platform.censys.io.

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, censys-logs).
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 como 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 usarlas más adelante.
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 y selecciona la política AmazonS3FullAccess.
18. Haz clic en Siguiente.
19. 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, vaya a IAM > Policies > Create policy > JSON tab (IAM > Políticas > Crear política > pestaña JSON).

2. Introduce la siguiente política:

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

   • Sustituye censys-logs si has introducido otro nombre de segmento.

3. Haz clic en Siguiente > Crear política.

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

5. Adjunta la política que acabas de crear y la política gestionada AWSLambdaBasicExecutionRole (para acceder a los registros de CloudWatch).

6. Dale el nombre censys-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:

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

   import json
   import boto3
   import urllib3
   import gzip
   import logging
   import os
   from datetime import datetime, timedelta, timezone
   from typing import Dict, List, Any, Optional
   from urllib.parse import urlencode
   
   # Configure logging
   logger = logging.getLogger()
   logger.setLevel(logging.INFO)
   
   # AWS S3 client
   s3_client = boto3.client('s3')
   # HTTP client
   http = urllib3.PoolManager()
   
   # Environment variables
   S3_BUCKET = os.environ['S3_BUCKET']
   S3_PREFIX = os.environ['S3_PREFIX']
   STATE_KEY = os.environ['STATE_KEY']
   CENSYS_API_KEY = os.environ['CENSYS_API_KEY']
   CENSYS_ORG_ID = os.environ['CENSYS_ORG_ID']
   API_BASE = os.environ.get('API_BASE', 'https://api.platform.censys.io')
   
   class CensysCollector:
       def __init__(self):
           self.headers = {
               'Authorization': f'Bearer {CENSYS_API_KEY}',
               'X-Organization-ID': CENSYS_ORG_ID,
               'Content-Type': 'application/json'
           }
   
       def get_last_collection_time(self) -> Optional[datetime]:
           """Get the last collection timestamp from S3 state file."""
           try:
               response = s3_client.get_object(Bucket=S3_BUCKET, Key=STATE_KEY)
               state = json.loads(response['Body'].read().decode('utf-8'))
               return datetime.fromisoformat(state.get('last_collection_time', '2024-01-01T00:00:00Z'))
           except Exception as e:
               logger.info(f"No state file found or error reading state: {e}")
               return datetime.now(timezone.utc) - timedelta(hours=1)
   
       def save_collection_time(self, collection_time: datetime):
           """Save the current collection timestamp to S3 state file."""
           state = {'last_collection_time': collection_time.strftime('%Y-%m-%dT%H:%M:%SZ')}
           s3_client.put_object(
               Bucket=S3_BUCKET,
               Key=STATE_KEY,
               Body=json.dumps(state),
               ContentType='application/json'
           )
   
       def collect_logbook_events(self, cursor: str = None) -> List[Dict[str, Any]]:
           """Collect logbook events from Censys ASM API using cursor-based pagination."""
           events = []
           url = f"{API_BASE}/v3/logbook"
   
           # Use cursor-based pagination as per Censys API documentation
           params = {}
           if cursor:
               params['cursor'] = cursor
   
           try:
               query_string = urlencode(params) if params else ''
               full_url = f"{url}?{query_string}" if query_string else url
   
               response = http.request('GET', full_url, headers=self.headers)
   
               if response.status != 200:
                   logger.error(f"API request failed with status {response.status}: {response.data}")
                   return []
   
               data = json.loads(response.data.decode('utf-8'))
               events.extend(data.get('logbook_entries', []))
   
               # Handle cursor-based pagination
               next_cursor = data.get('next_cursor')
               if next_cursor:
                   events.extend(self.collect_logbook_events(next_cursor))
   
               logger.info(f"Collected {len(events)} logbook events")
               return events
   
           except Exception as e:
               logger.error(f"Error collecting logbook events: {e}")
               return []
   
       def collect_risks_events(self) -> List[Dict[str, Any]]:
           """Collect risk events from Censys ASM API."""
           events = []
           url = f"{API_BASE}/v3/risks"
   
           try:
               response = http.request('GET', url, headers=self.headers)
   
               if response.status != 200:
                   logger.error(f"API request failed with status {response.status}: {response.data}")
                   return []
   
               data = json.loads(response.data.decode('utf-8'))
               events.extend(data.get('risks', []))
   
               logger.info(f"Collected {len(events)} risk events")
               return events
   
           except Exception as e:
               logger.error(f"Error collecting risk events: {e}")
               return []
   
       def save_events_to_s3(self, events: List[Dict[str, Any]], event_type: str):
           """Save events to S3 in compressed NDJSON format."""
           if not events:
               return
   
           timestamp = datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')
           filename = f"{S3_PREFIX}{event_type}_{timestamp}.json.gz"
   
           try:
               # Convert events to newline-delimited JSON
               ndjson_content = 'n'.join(json.dumps(event, separators=(',', ':')) for event in events)
   
               # Compress with gzip
               gz_bytes = gzip.compress(ndjson_content.encode('utf-8'))
   
               s3_client.put_object(
                   Bucket=S3_BUCKET,
                   Key=filename,
                   Body=gz_bytes,
                   ContentType='application/gzip',
                   ContentEncoding='gzip'
               )
   
               logger.info(f"Saved {len(events)} {event_type} events to {filename}")
   
           except Exception as e:
               logger.error(f"Error saving {event_type} events to S3: {e}")
               raise
   
   def lambda_handler(event, context):
       """AWS Lambda handler function."""
       try:
           collector = CensysCollector()
   
           # Get last collection time for cursor state management
           last_collection_time = collector.get_last_collection_time()
           current_time = datetime.now(timezone.utc)
   
           logger.info(f"Collecting events since {last_collection_time}")
   
           # Collect different types of events
           logbook_events = collector.collect_logbook_events()
           risk_events = collector.collect_risks_events()
   
           # Save events to S3
           collector.save_events_to_s3(logbook_events, 'logbook')
           collector.save_events_to_s3(risk_events, 'risks')
   
           # Update state
           collector.save_collection_time(current_time)
   
           return {
               'statusCode': 200,
               'body': json.dumps({
                   'message': 'Censys data collection completed successfully',
                   'logbook_events': len(logbook_events),
                   'risk_events': len(risk_events),
                   'collection_time': current_time.strftime('%Y-%m-%dT%H:%M:%SZ')
               })
           }
   
       except Exception as e:
           logger.error(f"Lambda execution failed: {str(e)}")
           return {
               'statusCode': 500,
               'body': json.dumps({
                   'error': str(e)
               })
           }
   

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

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

   Clave	Valor de ejemplo
   S3_BUCKET	censys-logs
   S3_PREFIX	censys/
   STATE_KEY	censys/state.json
   CENSYS_API_KEY	<your-censys-api-key>
   CENSYS_ORG_ID	<your-organization-id>
   API_BASE	https://api.platform.censys.io

4. Una vez creada la función, permanece en su página (o abre Lambda > Funciones > tu-función).

5. Seleccione la pestaña Configuración.

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

7. 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 censys-data-collector.
   • Nombre: censys-data-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. En la consola de AWS, ve a IAM > Usuarios > Añadir usuarios.
2. Haz clic en Add users (Añadir usuarios).
3. Proporcione los siguientes detalles de configuración:
   • Usuario: secops-reader.
   • Tipo de acceso: Clave de acceso (acceso programático).
4. Haz clic en Crear usuario.
5. Asigna una política de lectura mínima (personalizada): Usuarios > secops-reader > 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:::censys-logs/*"
       },
       {
         "Effect": "Allow",
         "Action": ["s3:ListBucket"],
         "Resource": "arn:aws:s3:::censys-logs"
       }
     ]
   }
   

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 Censys

1. Ve a Configuración de SIEM > Feeds.
2. Haz clic en + Añadir nuevo feed.
3. En el campo Nombre del feed, introduce un nombre para el feed (por ejemplo, Censys logs).
4. Selecciona Amazon S3 V2 como Tipo de fuente.
5. Seleccione CENSYS como Tipo de registro.
6. Haz clic en Siguiente.
7. Especifique los valores de los siguientes parámetros de entrada:
   • URI de S3: s3://censys-logs/censys/
   • 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 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 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.
8. Haz clic en Siguiente.
9. Revise la configuración de la nueva fuente en la pantalla Finalizar y, a continuación, haga clic en Enviar.

Tabla de asignación de UDM

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