Raccogli i log di HCL BigFix

Supportata in:

Google SecOps SIEM

Questo documento spiega come importare i log di HCL BigFix in Google Security Operations utilizzando Google Cloud Storage V2 tramite una funzione Cloud Run.

HCL BigFix è una piattaforma di gestione degli endpoint che fornisce funzionalità unificate di gestione, sicurezza e conformità degli endpoint. Consente alle organizzazioni di rilevare, gestire e correggere gli endpoint nella loro infrastruttura, tra cui la gestione delle patch, la distribuzione del software, la gestione delle configurazioni e la conformità della sicurezza. La funzione Cloud Run esegue il polling dell'API REST HCL BigFix utilizzando query di rilevanza della sessione, scrive i risultati come NDJSON in un bucket Cloud Storage e Google SecOps li importa tramite un feed Cloud Storage V2.

Prima di iniziare

Assicurati di soddisfare i seguenti prerequisiti:

• Un'istanza Google SecOps
• Un progetto Google Cloud con fatturazione abilitata
• Accesso alla console HCL BigFix con privilegi di operatore principale
• HCL BigFix Server versione 9.5 o successive
• Servizio WebReports in esecuzione sul server BigFix
• Connettività di rete dalla funzione Cloud Run al server BigFix sulla porta 52311 (porta API REST predefinita)
• Google Cloud API abilitate: Cloud Run Functions, Cloud Storage, Cloud Scheduler, Pub/Sub e Identity and Access Management (IAM)

Crea un bucket di Google Cloud Storage

1. Vai alla console Google Cloud.
2. Seleziona il tuo progetto o creane uno nuovo.
3. Nel menu di navigazione, vai a Cloud Storage > Bucket.
4. Fai clic su Crea bucket.

5. Fornisci i seguenti dettagli di configurazione:

   Impostazione	Valore
   Assegna un nome al bucket	Inserisci un nome globalmente univoco (ad esempio hcl-bigfix-logs)
   Tipo di località	Scegli in base alle tue esigenze (Regione, a due regioni, multiregionale)
   Località	Seleziona la località più vicina alla tua istanza Google SecOps (ad esempio us-central1)
   Classe di archiviazione	Standard (consigliato per i log ad accesso frequente)
   Controllo degli accessi	Uniforme (consigliato)
   Strumenti di protezione	(Facoltativo) Abilita il controllo delle versioni degli oggetti o la policy di conservazione

6. Fai clic su Crea.

   Nota: registra il nome del bucket. Lo utilizzerai durante la configurazione della funzione Cloud Run e del feed Google SecOps.

Raccogli le credenziali dell'API HCL BigFix

Per consentire alla funzione Cloud Run di recuperare i dati degli endpoint, devi creare un operatore della console BigFix dedicato con autorizzazioni di accesso all'API REST.

Crea un operatore dedicato per l'accesso API

1. Accedi alla console BigFix.
2. Vai a Strumenti > Crea operatore.

3. Nella finestra di dialogo Aggiungi utente, fornisci le seguenti informazioni:

   • Nome utente: inserisci un nome utente descrittivo (ad esempio chronicle_api).
   • Password: inserisci una password efficace.
   • Conferma password: inserisci di nuovo la password.

4. Fai clic su OK.

Configura le autorizzazioni dell'operatore

Dopo aver creato l'operatore, si apre automaticamente la finestra Console Operator (Operatore console).

1. Fai clic sulla scheda Dettagli.
2. Nella sezione Panoramica:
   • Seleziona Consenti sempre a questo utente di accedere.
3. Nella sezione Autorizzazioni, imposta le seguenti autorizzazioni:
   • Can Submit Queries: imposta su Yes.
   • Contenuti personalizzati: imposta su Sì.
4. Nella sezione Interface Login Privileges (Privilegi di accesso all'interfaccia):
   • Può utilizzare l'API REST: imposta su Sì.
5. Fai clic sulla scheda Computer amministrati.
6. Seleziona Tutti i computer per consentire all'operatore di eseguire query su tutti gli endpoint oppure seleziona gruppi di computer specifici in base ai tuoi requisiti.
7. Fai clic su Salva modifiche.

Registra le credenziali API

Salva le seguenti informazioni per configurare le variabili di ambiente della funzione Cloud Run:

• Nome utente: il nome utente dell'operatore che hai creato (ad esempio chronicle_api)
• Password: la password dell'operatore
• Nome host del server BigFix: il nome di dominio completo o l'indirizzo IP del server BigFix (ad esempio, bigfix.company.com)

• Porta API: la porta dell'API REST (il valore predefinito è 52311)

  Importante: memorizza queste credenziali in modo sicuro. La password non può essere recuperata dopo questo passaggio.

Esegui il test della connettività API

• Verifica che le credenziali API funzionino eseguendo la seguente query di test da una macchina con accesso di rete al server BigFix:

  curl -k -u "chronicle_api:YOUR_PASSWORD" \
    -X POST \
    -H "Content-Type: application/x-www-form-urlencoded" \
    -d "relevance=(name of it, id of it) of bes computers" \
    "https://bigfix-server:52311/api/query"
  

  Nota: sostituisci bigfix-server con il nome host del server BigFix, chronicle_api con il nome utente dell'operatore e YOUR_PASSWORD con la password dell'operatore. Il flag -k ignora la verifica del certificato TLS solo a scopo di test.

  Una risposta positiva restituisce un documento XML contenente i nomi e gli ID dei computer. Se ricevi un errore di autenticazione, verifica le autorizzazioni dell'operatore configurate nei passaggi precedenti.

Crea un account di servizio per la funzione Cloud Run

1. Nella console Google Cloud, vai a IAM e amministrazione > Account di servizio.
2. Fai clic su Crea account di servizio.

3. Fornisci i seguenti dettagli di configurazione:

   • Nome account di servizio: inserisci bigfix-cloud-run (o un nome descrittivo).
   • Descrizione service account: inserisci Service account for HCL BigFix Cloud Run function to write logs to Cloud Storage.

4. Fai clic su Crea e continua.

5. Nella sezione Concedi a questo account di servizio l'accesso al progetto, aggiungi i seguenti ruoli:

   • Storage Object Admin (per leggere/scrivere oggetti nel bucket Cloud Storage).
   • Cloud Run Invoker (per consentire a Cloud Scheduler di richiamare la funzione).

6. Fai clic su Continua.

7. Fai clic su Fine.

   Nota: prendi nota dell'indirizzo email del account di servizio (ad esempio, bigfix-cloud-run@PROJECT_ID.iam.gserviceaccount.com). Lo utilizzerai durante la creazione del job Cloud Scheduler.

Crea un argomento Pub/Sub

Cloud Scheduler attiva la funzione Cloud Run tramite un argomento Pub/Sub.

1. Nella console Google Cloud, vai a Pub/Sub > Argomenti.
2. Fai clic su Crea argomento.
3. Nel campo ID argomento, inserisci bigfix-ingestion-trigger.
4. Lascia le impostazioni predefinite.

5. Fai clic su Crea.

   Nota: registra il nome completo dell'argomento (ad esempio, projects/PROJECT_ID/topics/bigfix-ingestion-trigger).

Crea la funzione Cloud Run

Prepara i file di origine della funzione

Crea i seguenti due file per il deployment della funzione Cloud Run.

• requirements.txt

  functions-framework==3.*
  google-cloud-storage==2.*
  requests>=2.31.0
  

• main.py

  import base64
  import json
  import logging
  import os
  import xml.etree.ElementTree as ET
  from datetime import datetime, timezone, timedelta
  
  import requests
  import urllib3
  from google.cloud import storage
  
  urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
  
  logging.basicConfig(level=logging.INFO)
  logger = logging.getLogger(__name__)
  
  GCS_BUCKET = os.environ["GCS_BUCKET"]
  GCS_PREFIX = os.environ.get("GCS_PREFIX", "bigfix_logs")
  STATE_KEY = os.environ.get("STATE_KEY", "bigfix_state.json")
  BIGFIX_SERVER = os.environ["BIGFIX_SERVER"]
  BIGFIX_PORT = os.environ.get("BIGFIX_PORT", "52311")
  BIGFIX_USERNAME = os.environ["BIGFIX_USERNAME"]
  BIGFIX_PASSWORD = os.environ["BIGFIX_PASSWORD"]
  MAX_RECORDS = int(os.environ.get("MAX_RECORDS", "10000"))
  LOOKBACK_HOURS = int(os.environ.get("LOOKBACK_HOURS", "24"))
  
  BIGFIX_BASE_URL = f"https://{BIGFIX_SERVER}:{BIGFIX_PORT}/api"
  
  RELEVANCE_QUERIES = {
      "computers": (
          '(id of it, name of it, operating system of it, '
          'last report time of it as string, ip address of it as string, '
          'root server of it) of bes computers'
      ),
      "actions": (
          '(id of it, name of it, name of issuer of it, '
          'time issued of it as string, state of it) of bes actions'
      ),
      "fixlets": (
          '(id of it, name of it, source severity of it, '
          'source of it, category of it) of bes fixlets '
          'whose (source severity of it is not "")'
      ),
  }
  
  def _get_auth_header():
      """Build Basic Auth header from credentials."""
      credentials = f"{BIGFIX_USERNAME}:{BIGFIX_PASSWORD}"
      encoded = base64.b64encode(credentials.encode("utf-8")).decode("utf-8")
      return {"Authorization": f"Basic {encoded}"}
  
  def _query_bigfix(relevance, session=None):
      """Execute a session relevance query against the BigFix REST API."""
      url = f"{BIGFIX_BASE_URL}/query"
      headers = _get_auth_header()
      headers["Content-Type"] = "application/x-www-form-urlencoded"
  
      data = {"relevance": relevance}
      http = session or requests
      resp = http.post(url, headers=headers, data=data, verify=False, timeout=120)
      resp.raise_for_status()
      return resp.text
  
  def _parse_query_xml(xml_text, query_name):
      """Parse BigFix XML response into a list of dicts."""
      root = ET.fromstring(xml_text)
      results = []
  
      for result_elem in root.iter("Result"):
          tuples = result_elem.findall("Tuple")
          if tuples:
              for tup in tuples:
                  answers = [a.text or "" for a in tup.findall("Answer")]
                  record = _map_answers(query_name, answers)
                  record["_query"] = query_name
                  record["_collection_time"] = (
                      datetime.now(timezone.utc).isoformat()
                  )
                  results.append(record)
          else:
              answers_elems = result_elem.findall("Answer")
              if answers_elems:
                  answers = [a.text or "" for a in answers_elems]
                  record = _map_answers(query_name, answers)
                  record["_query"] = query_name
                  record["_collection_time"] = (
                      datetime.now(timezone.utc).isoformat()
                  )
                  results.append(record)
  
      logger.info(
          "Query '%s' returned %d results.", query_name, len(results)
      )
      return results
  
  def _map_answers(query_name, answers):
      """Map positional answers to named fields based on query type."""
      field_maps = {
          "computers": [
              "client_id", "client_name", "OS",
              "last_report_time", "ip_address", "root_server",
          ],
          "actions": [
              "action_id", "action_name", "issuer",
              "time_issued", "state",
          ],
          "fixlets": [
              "fixlet_id", "fixlet_name", "source_severity",
              "source", "category",
          ],
      }
      fields = field_maps.get(query_name, [])
      record = {}
      for i, value in enumerate(answers):
          key = fields[i] if i < len(fields) else f"field_{i}"
          record[key] = value
      return record
  
  def _load_state(gcs_client):
      """Load the last run state from GCS."""
      bucket = gcs_client.bucket(GCS_BUCKET)
      blob = bucket.blob(f"{GCS_PREFIX}/{STATE_KEY}")
      if blob.exists():
          data = blob.download_as_text()
          return json.loads(data)
      return {}
  
  def _save_state(gcs_client, state):
      """Persist run state to GCS."""
      bucket = gcs_client.bucket(GCS_BUCKET)
      blob = bucket.blob(f"{GCS_PREFIX}/{STATE_KEY}")
      blob.upload_from_string(
          json.dumps(state), content_type="application/json"
      )
      logger.info("State saved to gs://%s/%s/%s", GCS_BUCKET, GCS_PREFIX, STATE_KEY)
  
  def _write_ndjson(gcs_client, records, timestamp_str):
      """Write records as NDJSON to GCS."""
      if not records:
          logger.info("No records to write.")
          return
  
      records = records[:MAX_RECORDS]
      ndjson_lines = "\n".join(json.dumps(r) for r in records)
      blob_name = f"{GCS_PREFIX}/{timestamp_str}/bigfix_events.ndjson"
      bucket = gcs_client.bucket(GCS_BUCKET)
      blob = bucket.blob(blob_name)
      blob.upload_from_string(ndjson_lines, content_type="application/x-ndjson")
      logger.info(
          "Wrote %d records to gs://%s/%s",
          len(records), GCS_BUCKET, blob_name,
      )
  
  def main(event, context):
      """Cloud Run function entry point triggered by Pub/Sub."""
      logger.info("HCL BigFix ingestion function started.")
  
      gcs_client = storage.Client()
      state = _load_state(gcs_client)
  
      last_run = state.get("last_run")
      if last_run:
          logger.info("Last successful run: %s", last_run)
  
      now = datetime.now(timezone.utc)
      timestamp_str = now.strftime("%Y/%m/%d/%H%M%S")
  
      all_records = []
      session = requests.Session()
  
      for query_name, relevance in RELEVANCE_QUERIES.items():
          try:
              xml_text = _query_bigfix(relevance, session=session)
              records = _parse_query_xml(xml_text, query_name)
              all_records.extend(records)
          except requests.exceptions.RequestException as e:
              logger.error(
                  "Failed to execute query '%s': %s", query_name, e
              )
          except ET.ParseError as e:
              logger.error(
                  "Failed to parse XML for query '%s': %s", query_name, e
              )
  
      session.close()
  
      _write_ndjson(gcs_client, all_records, timestamp_str)
  
      state["last_run"] = now.isoformat()
      state["records_written"] = len(all_records)
      _save_state(gcs_client, state)
  
      logger.info(
          "HCL BigFix ingestion complete. %d total records written.",
          len(all_records),
      )
      return "OK"
  

Esegui il deployment della funzione Cloud Run

1. Salva entrambi i file (main.py e requirements.txt) in una directory locale (ad esempio bigfix-function/).
2. Apri Cloud Shell o un terminale con l'interfaccia a riga di comando gcloud installata.

3. Esegui questo comando per eseguire il deployment della funzione:

   gcloud functions deploy bigfix-to-gcs \
     --gen2 \
     --region=us-central1 \
     --runtime=python312 \
     --trigger-topic=bigfix-ingestion-trigger \
     --entry-point=main \
     --memory=512MB \
     --timeout=540s \
     --service-account=bigfix-cloud-run@PROJECT_ID.iam.gserviceaccount.com \
     --set-env-vars="GCS_BUCKET=hcl-bigfix-logs,GCS_PREFIX=bigfix_logs,STATE_KEY=bigfix_state.json,BIGFIX_SERVER=bigfix.company.com,BIGFIX_PORT=52311,BIGFIX_USERNAME=chronicle_api,BIGFIX_PASSWORD=YOUR_PASSWORD,MAX_RECORDS=10000,LOOKBACK_HOURS=24"
   

4. Sostituisci i seguenti valori segnaposto:

   • PROJECT_ID: l'ID progetto Google Cloud .
   • hcl-bigfix-logs: il nome del tuo bucket Cloud Storage.
   • bigfix.company.com: il nome host o l'indirizzo IP del server BigFix.
   • chronicle_api: il tuo nome utente operatore BigFix.
   • YOUR_PASSWORD: la password dell'operatore BigFix.
   Nota: per i deployment di produzione, utilizza Secret Manager di Google Cloud per archiviare BIGFIX_PASSWORD anziché passarlo come variabile di ambiente in testo non crittografato. Fai riferimento al secret utilizzando --set-secrets="BIGFIX_PASSWORD=bigfix-password:latest".

5. Verifica il deployment controllando lo stato della funzione:

   gcloud functions describe bigfix-to-gcs --region=us-central1 --gen2
   

Riferimento alle variabili di ambiente

Variabile	Obbligatorio	Predefinito	Descrizione
GCS_BUCKET	Sì		Nome del bucket Cloud Storage per l'archiviazione dell'output NDJSON
GCS_PREFIX	No	bigfix_logs	Prefisso dell'oggetto (percorso della cartella) all'interno del bucket
STATE_KEY	No	bigfix_state.json	Nome del blob per il file di stato all'interno del prefisso
BIGFIX_SERVER	Sì		Nome host o indirizzo IP del server BigFix
BIGFIX_PORT	No	52311	Porta API REST BigFix
BIGFIX_USERNAME	Sì		Nome utente operatore BigFix
BIGFIX_PASSWORD	Sì		Password dell'operatore BigFix
MAX_RECORDS	No	10000	Numero massimo di record da scrivere per esecuzione
LOOKBACK_HOURS	No	24	Numero di ore da considerare per la raccolta iniziale

Crea un job Cloud Scheduler

Cloud Scheduler attiva la funzione Cloud Run a intervalli regolari tramite l'argomento Pub/Sub.

1. Nella console Google Cloud, vai a Cloud Scheduler.
2. Fai clic su Crea job.
3. Fornisci i seguenti dettagli di configurazione:
   • Nome: inserisci bigfix-ingestion-schedule.
   • Regione: seleziona la stessa regione della funzione Cloud Run (ad esempio us-central1).
   • Frequenza: inserisci 0 */4 * * * (esecuzione ogni 4 ore). Modifica la pianificazione in base al volume di dati e ai requisiti di aggiornamento.
   • Fuso orario: seleziona il fuso orario che preferisci.
4. Fai clic su Continua.
5. Nella sezione Configura l'esecuzione:
   • Tipo di destinazione: seleziona Pub/Sub.
   • Argomento: seleziona bigfix-ingestion-trigger.
   • Corpo del messaggio: inserisci {"run": true}.
6. Fai clic su Continua.
7. Nella sezione Configura le impostazioni facoltative:
   • Numero massimo di nuovi tentativi: inserisci 3.
   • Durata minima di backoff: inserisci 5s.
   • Durata massima backoff: inserisci 60s.
8. Fai clic su Crea.

9. Per eseguire un test immediato, fai clic sui tre puntini (…) accanto al nome del job e seleziona Forza esecuzione.

   Nota: monitora i log di esecuzione della funzione in Cloud Logging filtrando in base al nome della funzione bigfix-to-gcs. Verifica che i file NDJSON vengano visualizzati nel bucket Cloud Storage con il prefisso configurato.

Recupera il account di servizio Google SecOps e configura il feed

Google SecOps utilizza un account di servizio univoco per leggere i dati dal bucket Cloud Storage. Devi concedere a questo account di servizio l'accesso al tuo bucket.

Recupera l'indirizzo email del account di servizio

1. Vai a Impostazioni > Gestione dei dati > Feed. Se utilizzi una piattaforma solo SIEM, vai a Impostazioni > Feed.
2. Fai clic su Aggiungi nuovo feed.
3. Fai clic su Configura un singolo feed.
4. Nel campo Nome feed, inserisci un nome per il feed (ad es. HCL BigFix Logs).
5. Seleziona Google Cloud Storage V2 come Tipo di origine.
6. Seleziona HCL BigFix come Tipo di log.

7. Fai clic su Ottieni service account.

   Verrà visualizzato un indirizzo email univoco del account di servizio, ad esempio:

   [e-mail]
   

8. Copia questo indirizzo email per utilizzarlo nel passaggio successivo.

   Nota: ogni istanza Google SecOps ha un account di servizio univoco. Non utilizzare i service account di altri esempi o documentazione.

9. Fai clic su Avanti.

10. Specifica i valori per i seguenti parametri di input:

    • URL bucket di archiviazione: inserisci l'URI del bucket Cloud Storage:

      gs://hcl-bigfix-logs/bigfix_logs/
      

      • Sostituisci hcl-bigfix-logs con il nome del tuo bucket Cloud Storage.
      • Sostituisci bigfix_logs con il valore GCS_PREFIX configurato.
      Nota: includi sempre la barra finale (/) alla fine dell'URI.

    • Opzione di eliminazione dell'origine: seleziona l'opzione di eliminazione in base alle tue preferenze:

      • Mai: non elimina mai i file dopo i trasferimenti (opzione consigliata per i test).
      • Elimina i file trasferiti: elimina i file dopo il trasferimento riuscito.

      • Elimina i file trasferiti e le directory vuote: elimina i file e le directory vuote dopo il trasferimento riuscito.

        Nota: se selezioni un'opzione di eliminazione, il account di servizio Google SecOps deve avere il ruolo Storage Object Admin anziché Storage Object Viewer. Aggiorna le autorizzazioni IAM di conseguenza.

    • Età massima del file: includi i file modificati negli ultimi giorni (il valore predefinito è 180 giorni).

    • Spazio dei nomi dell'asset: lo spazio dei nomi dell'asset.

    • Etichette di importazione: l'etichetta da applicare agli eventi di questo feed.

11. Fai clic su Avanti.

12. Controlla la nuova configurazione del feed di notizie nella schermata Finalizza e poi fai clic su Invia.

Concedi le autorizzazioni IAM al account di servizio Google SecOps

Il account di servizio Google SecOps deve avere il ruolo Storage Object Viewer nel bucket Cloud Storage.

1. Vai a Cloud Storage > Bucket.
2. Fai clic sul nome del bucket (ad esempio hcl-bigfix-logs).
3. Vai alla scheda Autorizzazioni.
4. Fai clic su Concedi l'accesso.
5. Fornisci i seguenti dettagli di configurazione:
   • Aggiungi entità: incolla l'indirizzo email del account di servizio Google SecOps (ad esempio, [e-mail]).
   • Assegna ruoli: seleziona Visualizzatore oggetti Storage.

6. Fai clic su Salva.

   Nota: se prevedi di utilizzare l'opzione di eliminazione "Elimina i file trasferiti" o "Elimina i file trasferiti e le directory vuote", concedi il ruolo Storage Object Admin anziché Storage Object Viewer.

Tabella di mappatura UDM

Campo log	Mappatura UDM	Logica
WMI_Asset_ID	entity.asset.asset_id	Impostato su "ID asset:" + valore se non vuoto e non "N/A"
Computer_Serial_Number, Computer_Manufacturer, Computer_Model	entity.asset.hardware	Oggetto con serial_number, manufacturer, model
client_name	entity.asset.hostname	Valore copiato direttamente
MAC_Addresses	entity.asset.mac	Unito dall'array dopo la formattazione
OperatingSystemType	entity.asset.platform_software.platform	Valore in maiuscolo se non è vuoto e non è "Unix"
Sistema operativo	entity.asset.platform_software.platform_version	Valore copiato direttamente
SerialNumber	entity.asset.product_object_id	Valore se non vuoto
WindowsOperatingSystem	entity.asset.software	Oggetto con nome impostato su value se non è vuoto
OperatingSystemRole	entity.asset.type	Valore in maiuscolo se non vuoto
DNS_Name	entity.domain.name	Valore se non vuoto
CloudInfo	entity.labels	Etichetta l'oggetto con la chiave "CloudInfo" e il valore (senza virgolette) se non è vuoto
Ruolo del dominio	entity.user.attribute.roles	Oggetto con nome impostato su value se non è vuoto
Active_Directory_Path	entity.user.group_identifiers	Valore unito se non è vuoto e non è ""
WMI_Service_Tag	entity.user.product_object_id	Valore se non vuoto
User_Name	entity.user.user_display_name	Valore se non vuoto e non ""
client_id	entity.user.userid	Valore copiato direttamente
	metadata.entity_type	Impostato su "ASSET" se SerialNumber non è vuoto, altrimenti su "USER"
	metadata.product_name	Imposta su "HCL Big Fix"
	metadata.vendor_name	Impostalo su "HCL Software"
CloudInfo	event.idm.entity.entity.labels	Mappato dal log delle modifiche

Log delle modifiche

Visualizza il log delle modifiche per questo parser

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