Raccogliere i log di Censys

Questo documento spiega come importare i log di Censys in Google Security Operations utilizzando Amazon S3. Censys fornisce una gestione completa della superficie di attacco e informazioni su internet tramite la sua API. Questa integrazione ti consente di raccogliere eventi di rilevamento degli host, eventi di rischio e modifiche agli asset da Censys ASM e inoltrarli a Google SecOps per l'analisi e il monitoraggio. Il parser trasforma i log non elaborati in un formato strutturato conforme a UDM di Google SecOps. Estrae i campi dal messaggio di log non elaborato, esegue conversioni del tipo di dati e mappa le informazioni estratte ai campi UDM corrispondenti, arricchendo i dati con contesto ed etichette aggiuntivi.

Prima di iniziare

Assicurati di soddisfare i seguenti prerequisiti:

• Istanza Google SecOps
• Accesso privilegiato a Censys ASM
• Accesso con privilegi ad AWS (S3, IAM, Lambda, EventBridge)

Raccogli i prerequisiti di Censys (credenziali API)

1. Accedi alla console Censys ASM all'indirizzo app.censys.io.
2. Vai a Integrazioni nella parte superiore della pagina.
3. Copia e salva la chiave API e l'ID organizzazione.
4. Prendi nota dell'URL di base dell'API: https://api.platform.censys.io

Configura il bucket AWS S3 e IAM per Google SecOps

1. Crea un bucket Amazon S3 seguendo questa guida utente: Creazione di un bucket
2. Salva il nome e la regione del bucket per riferimento futuro (ad esempio, censys-logs).
3. Crea un utente seguendo questa guida: Creazione di un utente IAM.
4. Seleziona l'utente creato.
5. Seleziona la scheda Credenziali di sicurezza.
6. Fai clic su Crea chiave di accesso nella sezione Chiavi di accesso.
7. Seleziona Servizio di terze parti come Caso d'uso.
8. Fai clic su Avanti.
9. (Facoltativo) Aggiungi un tag di descrizione.
10. Fai clic su Crea chiave di accesso.
11. Fai clic su Scarica file CSV per salvare la chiave di accesso e la chiave di accesso segreta per un utilizzo successivo.
12. Fai clic su Fine.
13. Seleziona la scheda Autorizzazioni.
14. Fai clic su Aggiungi autorizzazioni nella sezione Criteri per le autorizzazioni.
15. Seleziona Aggiungi autorizzazioni.
16. Seleziona Allega direttamente i criteri.
17. Cerca e seleziona il criterio AmazonS3FullAccess.
18. Fai clic su Avanti.
19. Fai clic su Aggiungi autorizzazioni.

Configura il ruolo e il criterio IAM per i caricamenti S3

1. Nella console AWS, vai a IAM > Policy > Crea policy > Scheda JSON.

2. Inserisci la seguente policy:

   {
     "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"
       }
     ]
   }
   

   • Sostituisci censys-logs se hai inserito un nome bucket diverso.

3. Fai clic su Avanti > Crea criterio.

4. Vai a IAM > Ruoli > Crea ruolo > Servizio AWS > Lambda.

5. Collega il criterio appena creato e il criterio gestito AWSLambdaBasicExecutionRole (per l'accesso a CloudWatch Logs).

6. Assegna al ruolo il nome censys-lambda-role e fai clic su Crea ruolo.

Crea la funzione Lambda

1. Nella console AWS, vai a Lambda > Funzioni > Crea funzione.
2. Fai clic su Crea autore da zero.
3. Fornisci i seguenti dettagli di configurazione:

1. Dopo aver creato la funzione, apri la scheda Codice, elimina lo stub e inserisci il seguente codice (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. Vai a Configurazione > Variabili di ambiente > Modifica > Aggiungi nuova variabile di ambiente.

3. Inserisci le seguenti variabili di ambiente, sostituendole con i tuoi valori:

   Chiave	Valore di esempio
   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. Dopo aver creato la funzione, rimani sulla relativa pagina (o apri Lambda > Funzioni > la tua funzione).

5. Seleziona la scheda Configurazione.

6. Nel riquadro Configurazione generale, fai clic su Modifica.

7. Modifica Timeout impostandolo su 5 minuti (300 secondi) e fai clic su Salva.

Creare una pianificazione EventBridge

1. Vai a Amazon EventBridge > Scheduler > Crea pianificazione.
2. Fornisci i seguenti dettagli di configurazione:
   • Programma ricorrente: Tariffa (1 hour).
   • Destinazione: la tua funzione Lambda censys-data-collector.
   • Nome: censys-data-collector-1h
3. Fai clic su Crea pianificazione.

(Facoltativo) Crea chiavi e utenti IAM di sola lettura per Google SecOps

1. Nella console AWS, vai a IAM > Utenti > Aggiungi utenti.
2. Fai clic su Add users (Aggiungi utenti).
3. Fornisci i seguenti dettagli di configurazione:
   • Utente: secops-reader.
   • Tipo di accesso: Chiave di accesso - Accesso programmatico.
4. Fai clic su Crea utente.
5. Collega la criterio per la lettura minima (personalizzata): Utenti > secops-reader > Autorizzazioni > Aggiungi autorizzazioni > Collega le norme direttamente > Crea norma.

6. Nell'editor JSON, inserisci la seguente policy:

   {
     "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. Imposta il nome su secops-reader-policy.

8. Vai a Crea criterio > cerca/seleziona > Avanti > Aggiungi autorizzazioni.

9. Vai a Credenziali di sicurezza > Chiavi di accesso > Crea chiave di accesso.

10. Scarica il file CSV (questi valori vengono inseriti nel feed).

Configura un feed in Google SecOps per importare i log di Censys

1. Vai a Impostazioni SIEM > Feed.
2. Fai clic su + Aggiungi nuovo feed.
3. Nel campo Nome feed, inserisci un nome per il feed (ad esempio, Censys logs).
4. Seleziona Amazon S3 V2 come Tipo di origine.
5. Seleziona CENSYS come Tipo di log.
6. Fai clic su Avanti.
7. Specifica i valori per i seguenti parametri di input:
   • URI S3: s3://censys-logs/censys/
   • Opzioni di eliminazione dell'origine: seleziona l'opzione di eliminazione in base alle tue preferenze.
   • Età massima del file: includi i file modificati nell'ultimo numero di giorni. Il valore predefinito è 180 giorni.
   • ID chiave di accesso: chiave di accesso utente con accesso al bucket S3.
   • Chiave di accesso segreta: chiave segreta dell'utente con accesso al bucket S3.
   • Spazio dei nomi dell'asset: lo spazio dei nomi dell'asset.
   • Etichette di importazione: l'etichetta applicata agli eventi di questo feed.
8. Fai clic su Avanti.
9. Controlla la nuova configurazione del feed nella schermata Finalizza e poi fai clic su Invia.

Tabella di mappatura UDM

Hai bisogno di ulteriore assistenza? Ricevi risposte dai membri della community e dai professionisti di Google SecOps.