HCL BigFix-Protokolle erfassen

Unterstützt in:

In diesem Dokument wird beschrieben, wie Sie HCL BigFix-Logs über eine Cloud Run-Funktion mit Google Cloud Storage V2 in Google Security Operations aufnehmen.

HCL BigFix ist eine Plattform für die Endpunktverwaltung, die einheitliche Funktionen für die Endpunktverwaltung, Sicherheit und Compliance bietet. Mit BigFix können Unternehmen Endpunkte in ihrer gesamten Infrastruktur erkennen, verwalten und beheben. Dazu gehören Patch-Management, Softwareverteilung, Konfigurationsverwaltung und Sicherheits-Compliance. Die Cloud Run-Funktion fragt die HCL BigFix REST API mithilfe von Sitzungsrelevanzabfragen ab, schreibt die Ergebnisse als NDJSON in einen Cloud Storage-Bucket und Google SecOps nimmt sie über einen Cloud Storage V2-Feed auf.

Hinweis

Prüfen Sie, ob folgende Voraussetzungen erfüllt sind:

  • Eine Google SecOps-Instanz
  • Ein Google Cloud -Projekt mit aktivierter Abrechnung
  • Zugriff auf die HCL BigFix-Konsole mit Master-Operator-Berechtigungen
  • HCL BigFix Server Version 9.5 oder höher
  • WebReports-Dienst, der auf dem BigFix-Server ausgeführt wird
  • Netzwerkverbindung von der Cloud Run-Funktion zum BigFix-Server über Port 52311 (Standard-REST API-Port)
  • Google Cloud Aktivierte APIs: Cloud Run-Funktionen, Cloud Storage, Cloud Scheduler, Pub/Sub und Identity and Access Management (IAM)

Google Cloud -Storage-Bucket erstellen

  1. Gehen Sie zur Google Cloud Console.
  2. Wählen Sie Ihr Projekt aus oder erstellen Sie ein neues.
  3. Rufen Sie im Navigationsmenü Cloud Storage > Buckets auf.
  4. Klicken Sie auf Bucket erstellen.

  5. Geben Sie die folgenden Konfigurationsdetails an:

    Einstellung Wert
    Bucket benennen Geben Sie einen global eindeutigen Namen ein (z. B. hcl-bigfix-logs).
    Standorttyp Wählen Sie die Option, die am besten zu Ihren Anforderungen passt (Region, Dual-Region, Multi-Region).
    Standort Wählen Sie den Standort aus, der Ihrer Google SecOps-Instanz am nächsten liegt (z. B. us-central1).
    Speicherklasse Standard (empfohlen für Logs, auf die häufig zugegriffen wird)
    Zugriffssteuerung Einheitlich (empfohlen)
    Schutztools Optional: Objektversionsverwaltung oder Aufbewahrungsrichtlinie aktivieren

  6. Klicken Sie auf Erstellen.

HCL BigFix API-Anmeldedaten erfassen

Damit die Cloud Run-Funktion Endpunktdaten abrufen kann, müssen Sie einen dedizierten BigFix Console-Operator mit REST API-Zugriffsberechtigungen erstellen.

Dedizierten Operator für den API-Zugriff erstellen

  1. Melden Sie sich in der BigFix Console an.
  2. Gehen Sie zu Tools > Operator erstellen.

  3. Geben Sie im Dialogfeld Nutzer hinzufügen die folgenden Informationen ein:

    • Nutzername: Geben Sie einen aussagekräftigen Nutzernamen ein (z. B. chronicle_api).
    • Passwort: Geben Sie ein starkes Passwort ein.
    • Passwort bestätigen: Geben Sie das Passwort noch einmal ein.

  4. Klicken Sie auf OK.

Operatorberechtigungen konfigurieren

Nachdem Sie den Operator erstellt haben, wird automatisch das Fenster Console Operator geöffnet.

  1. Klicken Sie auf den Tab Details.
  2. Im Abschnitt Übersicht:
    • Wählen Sie Diesem Nutzer immer die Anmeldung erlauben aus.
  3. Legen Sie im Abschnitt Berechtigungen die folgenden Berechtigungen fest:
    • Can Submit Queries (Kann Abfragen senden): Legen Sie diese Option auf Yes (Ja) fest.
    • Benutzerdefinierter Inhalt: Auf Ja festlegen.
  4. Im Abschnitt Anmeldeberechtigungen für die Benutzeroberfläche:
    • Can use REST API (Kann REST API verwenden): Auf Yes (Ja) festgelegt.
  5. Klicken Sie auf den Tab Verwaltete Computer.
  6. Wählen Sie Alle Computer aus, damit der Operator alle Endpunkte abfragen kann, oder wählen Sie bestimmte Computergruppen aus, die Ihren Anforderungen entsprechen.
  7. Klicken Sie auf Änderungen speichern.

API-Anmeldedaten aufzeichnen

Speichern Sie die folgenden Informationen, um die Umgebungsvariablen der Cloud Run-Funktion zu konfigurieren:

  • Nutzername: Der von Ihnen erstellte Nutzername des Operators (z. B. chronicle_api)
  • Password: Das Passwort des Operators
  • BigFix-Server-Hostname: Der voll qualifizierte Domainname oder die IP-Adresse Ihres BigFix-Servers (z. B. bigfix.company.com)

  • API-Port: Der REST API-Port (Standard ist 52311)

API-Konnektivität testen

  • Prüfen Sie, ob die API-Anmeldedaten funktionieren, indem Sie die folgende Testabfrage von einem Computer mit Netzwerkzugriff auf den BigFix-Server ausführen:

    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"

    Bei einer erfolgreichen Antwort wird ein XML-Dokument mit Computernamen und ‑IDs zurückgegeben. Wenn Sie einen Authentifizierungsfehler erhalten, prüfen Sie die in den vorherigen Schritten konfigurierten Operatorberechtigungen.

Dienstkonto für die Cloud Run-Funktion erstellen

  1. Rufen Sie in der Google Cloud Console IAM und Verwaltung > Dienstkonten auf.
  2. Klicken Sie auf Dienstkonto erstellen.

  3. Geben Sie die folgenden Konfigurationsdetails an:

    • Name des Dienstkontos: Geben Sie bigfix-cloud-run (oder einen aussagekräftigen Namen) ein.
    • Beschreibung des Dienstkontos: Geben Sie Service account for HCL BigFix Cloud Run function to write logs to Cloud Storage ein.

  4. Klicken Sie auf Erstellen und fortfahren.

  5. Fügen Sie im Abschnitt Diesem Dienstkonto Zugriff auf das Projekt erteilen die folgenden Rollen hinzu:

    • Storage-Objekt-Administrator (zum Lesen/Schreiben von Objekten im Cloud Storage-Bucket).
    • Cloud Run-Aufrufer (damit Cloud Scheduler die Funktion aufrufen kann).

  6. Klicken Sie auf Weiter.

  7. Klicken Sie auf Fertig.

Pub/Sub-Thema erstellen

Cloud Scheduler löst die Cloud Run-Funktion über ein Pub/Sub-Thema aus.

  1. Rufen Sie in der Google Cloud Console Pub/Sub > Themen auf.
  2. Klicken Sie auf Thema erstellen.
  3. Geben Sie im Feld Themen-ID den Wert bigfix-ingestion-trigger ein.
  4. Behalten Sie die Standardeinstellungen bei.

  5. Klicken Sie auf Erstellen.

Cloud Run-Funktion erstellen

Funktionsquelldateien vorbereiten

Erstellen Sie die folgenden beiden Dateien für die Bereitstellung der Cloud Run-Funktion.

  • 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"

Cloud Run-Funktion bereitstellen

  1. Speichern Sie beide Dateien (main.py und requirements.txt) in einem lokalen Verzeichnis (z. B. bigfix-function/).
  2. Öffnen Sie Cloud Shell oder ein Terminal mit installiertem gcloud-CLI.

  3. Führen Sie den folgenden Befehl aus, um die Funktion bereitzustellen:

    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. Ersetzen Sie die folgenden Platzhalterwerte:

    • PROJECT_ID: Ihre Google Cloud Projekt-ID
    • hcl-bigfix-logs: Name Ihres Cloud Storage-Bucket.
    • bigfix.company.com: Der Hostname oder die IP-Adresse Ihres BigFix-Servers.
    • chronicle_api: Ihr BigFix-Operator-Nutzername.
    • YOUR_PASSWORD: Ihr BigFix-Operator-Passwort.

  5. Prüfen Sie die Bereitstellung, indem Sie den Funktionsstatus abrufen:

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

Referenz zu Umgebungsvariablen

Variable Erforderlich Standard Beschreibung
GCS_BUCKET Ja Name des Cloud Storage-Bucket zum Speichern der NDJSON-Ausgabe
GCS_PREFIX Nein bigfix_logs Objektpräfix (Ordnerpfad) im Bucket
STATE_KEY Nein bigfix_state.json Blob-Name für die Statusdatei innerhalb des Präfixes
BIGFIX_SERVER Ja Hostname oder IP-Adresse des BigFix-Servers
BIGFIX_PORT Nein 52311 BigFix REST API-Port
BIGFIX_USERNAME Ja Nutzername des BigFix-Operators
BIGFIX_PASSWORD Ja BigFix-Operator-Passwort
MAX_RECORDS Nein 10000 Maximale Anzahl von Datensätzen, die pro Ausführung geschrieben werden sollen
LOOKBACK_HOURS Nein 24 Anzahl der Stunden, die für die erste Erfassung zurückgeblickt werden soll

Cloud Scheduler-Job erstellen

Cloud Scheduler löst die Cloud Run-Funktion in regelmäßigen Abständen über das Pub/Sub-Thema aus.

  1. Rufen Sie in der Google Cloud Console Cloud Scheduler auf.
  2. Klicken Sie auf Job erstellen.
  3. Geben Sie die folgenden Konfigurationsdetails an:
    • Name: Geben Sie bigfix-ingestion-schedule ein.
    • Region: Wählen Sie dieselbe Region wie für Ihre Cloud Run-Funktion aus (z. B. us-central1).
    • Häufigkeit: Geben Sie 0 */4 * * * ein (wird alle 4 Stunden ausgeführt). Passen Sie den Zeitplan an Ihr Datenvolumen und Ihre Anforderungen an die Aktualität an.
    • Zeitzone: Wählen Sie Ihre bevorzugte Zeitzone aus.
  4. Klicken Sie auf Weiter.
  5. Im Abschnitt Ausführung konfigurieren:
    • Zieltyp: Wählen Sie Pub/Sub aus.
    • Thema: Wählen Sie bigfix-ingestion-trigger aus.
    • Inhalt der Nachricht: Geben Sie {"run": true} ein.
  6. Klicken Sie auf Weiter.
  7. Im Abschnitt Optionale Einstellungen konfigurieren:
    • Maximale Wiederholungsversuche: Geben Sie 3 ein.
    • Minimale Backoff-Dauer: Geben Sie 5s ein.
    • Maximale Backoff-Dauer: Geben Sie 60s ein.
  8. Klicken Sie auf Erstellen.

  9. Wenn Sie einen sofortigen Test ausführen möchten, klicken Sie auf das Dreipunkt-Menü (…) neben dem Jobnamen und wählen Sie Ausführung erzwingen aus.

Google SecOps-Dienstkonto abrufen und Feed konfigurieren

Google SecOps verwendet ein eindeutiges Dienstkonto, um Daten aus Ihrem Cloud Storage-Bucket zu lesen. Sie müssen diesem Dienstkonto Zugriff auf Ihren Bucket gewähren.

E‑Mail-Adresse des Dienstkontos abrufen

  1. Rufen Sie die Einstellungen > Datenverwaltung > Feeds auf. Wenn Sie eine reine SIEM-Plattform verwenden, rufen Sie die Einstellungen > Feeds auf.
  2. Klicken Sie auf Neuen Feed hinzufügen.
  3. Klicken Sie auf Einzelnen Feed konfigurieren.
  4. Geben Sie im Feld Feedname einen Namen für den Feed ein, z. B. HCL BigFix Logs.
  5. Wählen Sie Google Cloud Storage V2 als Quelltyp aus.
  6. Wählen Sie HCL BigFix als Protokolltyp aus.

  7. Klicken Sie auf Dienstkonto abrufen.

    Eine eindeutige E-Mail-Adresse für das Dienstkonto wird angezeigt, z. B.:

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com

  8. Kopieren Sie diese E-Mail-Adresse, um sie im nächsten Schritt zu verwenden.

  9. Klicken Sie auf Weiter.

  10. Geben Sie Werte für die folgenden Eingabeparameter an:

    • Storage-Bucket-URL: Geben Sie den Cloud Storage-Bucket-URI ein:

      gs://hcl-bigfix-logs/bigfix_logs/
      • Ersetzen Sie hcl-bigfix-logs durch den Namen des Cloud Storage-Buckets.
      • Ersetzen Sie bigfix_logs durch den konfigurierten GCS_PREFIX-Wert.

    • Option zum Löschen der Quelle: Wählen Sie die gewünschte Löschoption aus:

      • Nie: Nach Übertragungen werden keine Dateien gelöscht (empfohlen für Tests).
      • Übertragene Dateien löschen: Dateien werden nach erfolgreicher Übertragung gelöscht.

      • Übertragene Dateien und leere Verzeichnisse löschen: Dateien und leere Verzeichnisse werden nach erfolgreicher Übertragung gelöscht.

    • Maximales Dateialter: Dateien einschließen, die in den letzten Tagen geändert wurden (Standardwert: 180 Tage).

    • Asset-Namespace: Der Asset-Namespace.

    • Aufnahmelabels: Das Label, das auf die Ereignisse aus diesem Feed angewendet werden soll.

  11. Klicken Sie auf Weiter.

  12. Prüfen Sie die neue Feedkonfiguration auf dem Bildschirm Abschließen und klicken Sie dann auf Senden.

Dem Google SecOps-Dienstkonto IAM-Berechtigungen gewähren

Das Google SecOps-Dienstkonto benötigt die Rolle Storage-Objekt-Betrachter für Ihren Cloud Storage-Bucket.

  1. Rufen Sie Cloud Storage > Buckets auf.
  2. Klicken Sie auf den Namen Ihres Buckets (z. B. hcl-bigfix-logs).
  3. Wechseln Sie zum Tab Berechtigungen.
  4. Klicken Sie auf Zugriff erlauben.
  5. Geben Sie die folgenden Konfigurationsdetails an:
    • Hauptkonten hinzufügen: Fügen Sie die E-Mail-Adresse des Google SecOps-Dienstkontos ein (z. B. chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com).
    • Rollen zuweisen: Wählen Sie Storage Object Viewer aus.

  6. Klicken Sie auf Speichern.

UDM-Zuordnungstabelle

Logfeld UDM-Zuordnung Logik
WMI_Asset_ID entity.asset.asset_id Set to "Asset Id:" + value if not empty and not "N/A"
Computer_Serial_Number, Computer_Manufacturer, Computer_Model entity.asset.hardware Objekt mit Seriennummer, Hersteller, Modell
client_name entity.asset.hostname Wert direkt kopiert
MAC_Addresses entity.asset.mac Nach der Formatierung aus Array zusammengeführt
OperatingSystemType entity.asset.platform_software.platform Wert in Großbuchstaben, wenn nicht leer und nicht „Unix“
Betriebssystem entity.asset.platform_software.platform_version Wert direkt kopiert
SerialNumber entity.asset.product_object_id Wert, wenn nicht leer
WindowsOperatingSystem entity.asset.software Objekt, dessen Name auf den Wert festgelegt ist, wenn er nicht leer ist
OperatingSystemRole entity.asset.type Wert in Großbuchstaben, wenn nicht leer
DNS_Name entity.domain.name Wert, wenn nicht leer
CloudInfo entity.labels Labelobjekt mit dem Schlüssel „CloudInfo“ und dem Wert (Anführungszeichen entfernt), falls nicht leer
Domainrolle entity.user.attribute.roles Objekt, dessen Name auf den Wert festgelegt ist, wenn er nicht leer ist
Active_Directory_Path entity.user.group_identifiers Zusammengeführter Wert, wenn nicht leer und nicht „“
WMI_Service_Tag entity.user.product_object_id Wert, wenn nicht leer
User_Name entity.user.user_display_name Wert, wenn nicht leer und nicht „“
client_id entity.user.userid Wert direkt kopiert
metadata.entity_type Auf „ASSET“ festgelegt, wenn „SerialNumber“ nicht leer ist, andernfalls auf „USER“
metadata.product_name Legen Sie diesen Wert auf „HCL Big Fix“ fest.
metadata.vendor_name Legen Sie diesen Wert auf „HCL Software“ fest.
CloudInfo event.idm.entity.entity.labels Aus Änderungsprotokoll zugeordnet

Änderungsprotokoll

Änderungsprotokoll für diesen Parser ansehen

Benötigen Sie weitere Hilfe? Antworten von Community-Mitgliedern und Google SecOps-Experten erhalten