Oracle NetSuite-Logs erfassen

Unterstützt in:

In diesem Dokument wird beschrieben, wie Sie Oracle NetSuite-Logs mit Cloud Storage V2 in Google Security Operations aufnehmen.

Oracle NetSuite ist eine cloudbasierte ERP-Plattform (Enterprise Resource Planning), die umfassende Funktionen für die Unternehmensverwaltung bietet, darunter Finanzverwaltung, CRM, E-Commerce und Inventarverwaltung. NetSuite generiert Audit-Logs für Nutzeranmeldungen, Systemänderungen und API-Aktivitäten, die über die SuiteQL REST API abgefragt und zur Sicherheitsüberwachung und Compliance in Google SecOps aufgenommen werden können.

Hinweis

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

  • Eine Google SecOps-Instanz
  • Ein Google Cloud -Projekt mit aktivierten Cloud Storage-, Cloud Run-, Pub/Sub- und Cloud Scheduler-APIs
  • Berechtigungen zum Erstellen und Verwalten von Cloud Storage-Buckets
  • Berechtigungen zum Verwalten von IAM-Richtlinien (Identity and Access Management) für Cloud Storage-Buckets
  • Berechtigungen zum Erstellen von Cloud Run-Diensten, Pub/Sub-Themen und Cloud Scheduler-Jobs
  • Administratorzugriff auf Ihr Oracle NetSuite-Konto
  • In Ihrem NetSuite-Konto aktivierte REST-Webdienste und OAuth 2.0-Funktionen
  • Ihre NetSuite-Konto-ID (unter Einrichtung > Unternehmen > Unternehmensinformationen)

Erforderliche NetSuite-Funktionen aktivieren

  1. Melden Sie sich als Administrator in Ihrem NetSuite-Konto an.
  2. Gehen Sie zu Einrichtung> Unternehmen > Funktionen aktivieren.
  3. Klicken Sie auf den Untertab SuiteCloud.
  4. Wählen Sie im Bereich SuiteTalk (Web Services) das folgende Kästchen aus:
    • REST-Webdienste
  5. Wählen Sie im Bereich Authentifizierung verwalten das folgende Kästchen aus:
    • OAuth 2.0
  6. Klicken Sie auf Speichern.

OAuth 2.0-Anmeldedaten für Oracle NetSuite konfigurieren

Damit die Cloud Run-Funktion Audit-Logs aus NetSuite abrufen kann, müssen Sie einen Integrationsdatensatz mit OAuth 2.0-Clientanmeldedaten erstellen.

Integrationsdatensatz erstellen

  1. Melden Sie sich als Administrator in Ihrem NetSuite-Konto an.
  2. Gehen Sie zu Einrichten> Integration > Integrationen verwalten > Neu.
  3. Geben Sie im Feld Name Google SecOps Integration ein.
  4. Geben Sie im Feld Beschreibung den Text Integration with Google SecOps for audit log ingestion ein.
  5. Wählen Sie im Feld Status die Option Aktiviert aus.
  6. Klicken Sie auf den Untertab Authentifizierung.
  7. Konfigurieren Sie im Bereich OAuth 2.0 Folgendes:
    • Wählen Sie Clientanmeldedaten (Maschine zu Maschine) aus.
    • Wählen Sie REST Web Services aus.
  8. Entfernen Sie alle Häkchen im Bereich Token-based Authentication (Tokenbasierte Authentifizierung).
  9. Klicken Sie auf Speichern.

OAuth 2.0-Anmeldedaten aufzeichnen

Nachdem Sie den Integrationseintrag gespeichert haben, werden in NetSuite Ihre OAuth 2.0-Anmeldedaten angezeigt:

  • Client-ID: Ihre eindeutige Client-Kennung

  • Clientschlüssel: Ihr API-Schlüssel

Zertifikat für OAuth 2.0-Clientanmeldedaten generieren

Für den OAuth 2.0-Vorgang für Clientanmeldedaten ist ein Zertifikat für die JWT-Signierung erforderlich. ECDSA-Zertifikat mit OpenSSL generieren:

  1. Führen Sie auf einem Computer mit installiertem OpenSSL den folgenden Befehl aus, um einen privaten Schlüssel und ein Zertifikat zu generieren:

    openssl req -new -x509 -newkey ec \
    -pkeyopt ec_paramgen_curve:prime256v1 \
    -pkeyopt ec_param_enc:named_curve \
    -nodes -days 730 \
    -out public.pem -keyout private.pem

  2. Geben Sie bei entsprechender Aufforderung die Zertifikat-Betrefffelder ein (Land, Bundesland, Organisation usw.). Geben Sie für Allgemeiner Name Google SecOps Integration ein.

  3. Mit dem Befehl werden zwei Dateien generiert:

    • private.pem: Privater Schlüssel (sicher aufbewahren; wird für die JWT-Signierung in der Cloud Run-Funktion verwendet)
    • public.pem: Öffentliches Zertifikat (in NetSuite hochladen)

OAuth 2.0-Clientanmeldedatenzuordnung erstellen

  1. Gehen Sie in NetSuite zu Setup > Integration > Manage Authentication > OAuth 2.0 Client Credentials (M2M) Setup.
  2. Klicken Sie auf Neu erstellen.
  3. Konfigurieren Sie im Dialogfeld Folgendes:
    • Entität: Wählen Sie den Mitarbeiter oder Nutzer aus, der mit API-Anfragen verknüpft werden soll.
    • Rolle: Wählen Sie eine Rolle mit der Berechtigung Mit OAuth 2.0-Zugriffstokens anmelden aus (siehe Erforderliche Berechtigungen). Ohne diese Berechtigung wird die Rolle nicht in dieser Liste angezeigt.
    • Anwendung: Wählen Sie Google SecOps Integration aus.
  4. Klicken Sie auf Datei auswählen und laden Sie die Zertifikatsdatei public.pem hoch.
  5. Klicken Sie auf Speichern.
  6. Notieren Sie sich nach dem Speichern den Wert der Zertifikats-ID, der in der Zuordnungsliste angezeigt wird. Dieser Wert wird als kid (Schlüssel-ID) im JWT-Header verwendet.

Erforderliche Berechtigungen

Die der OAuth 2.0-Zuordnung zugewiesene Rolle muss die folgenden Berechtigungen haben:

Berechtigung Zugriffsebene Zweck
Mit OAuth 2.0-Zugriffstokens anmelden Vollständig Rolle die Verwendung der OAuth 2.0-Tokenanmeldung erlauben
REST-Webdienste Vollständig REST API-Anfragen ausführen
Audit-Trail für Anmeldungen Vollständig Auf Daten des Anmelde-Audit-Trails zugreifen
SuiteAnalytics Workbook Ansehen Über SuiteQL auf Analysedaten zugreifen
Systemnotizen für Analytics und REST Ansehen SystemNote-Datensätze über SuiteQL abfragen

Benutzerdefinierte Rolle für die Integration erstellen

  1. Klicken Sie auf Einrichtung> Nutzer/Rollen> Rollen verwalten> Neu.
  2. Geben Sie im Feld Name Google SecOps Integration Role ein.
  3. Klicken Sie auf den Untertab Berechtigungen.
  4. Klicken Sie auf den Unteruntertab Einrichtung.
  5. Fügen Sie die folgenden Berechtigungen hinzu:
    • Mit OAuth 2.0-Zugriffstokens anmelden: Auf Vollständig festlegen
    • REST-Webdienste: Auf Vollständig festlegen
    • Audit-Trail für die Anmeldung: Auf Vollständig festgelegt
  6. Klicken Sie auf den Unter-Untertab Berichte.
  7. Fügen Sie die folgende Berechtigung hinzu:
    • SuiteAnalytics Workbook: Auf View festlegen
  8. Klicken Sie auf den Unter-Untertab Listen.
  9. Fügen Sie die folgende Berechtigung hinzu:
    • System Notes for Analytics and REST: Auf View festlegen
  10. Klicken Sie auf Speichern.

Berechtigungen prüfen

So prüfen Sie, ob die Rolle die erforderlichen Berechtigungen hat:

  1. Melden Sie sich in NetSuite an.
  2. Gehen Sie zu Setup > Users/Roles > Manage Roles.
  3. Suchen Sie nach dem Google SecOps Integration Role und klicken Sie auf Bearbeiten.
  4. Prüfen Sie, ob die unter Erforderliche Berechtigungen aufgeführten Berechtigungen auf dem Untertab Berechtigungen vorhanden sind.

API-Zugriff testen

  • Testen Sie Ihre Anmeldedaten, bevor Sie mit der Integration fortfahren:

    # Replace with your actual values
ACCOUNT_ID="your-account-id"
CLIENT_ID="your-client-id"
CERTIFICATE_ID="your-certificate-id"
PRIVATE_KEY_FILE="private.pem"

# Generate JWT (requires Python 3 with PyJWT and cryptography)
pip install PyJWT cryptography

TOKEN=$(python3 -c "
import jwt, time, json
with open('${PRIVATE_KEY_FILE}', 'r') as f:
    key = f.read()
now = int(time.time())
payload = {
    'iss': '${CLIENT_ID}',
    'scope': ['restlets', 'rest_webservices'],
    'aud': 'https://${ACCOUNT_ID}.suitetalk.api.netsuite.com/services/rest/auth/oauth2/v1/token',
    'iat': now,
    'exp': now + 3600
}
print(jwt.encode(payload, key, algorithm='ES256', headers={'kid': '${CERTIFICATE_ID}', 'typ': 'JWT'}))
")

# Obtain access token
ACCESS_TOKEN=$(curl -s -X POST \
    "https://${ACCOUNT_ID}.suitetalk.api.netsuite.com/services/rest/auth/oauth2/v1/token" \
    -H "Content-Type: application/x-www-form-urlencoded" \
    -d "grant_type=client_credentials&client_assertion_type=urn%3Aietf%3Aparams%3Aoauth%3Aclient-assertion-type%3Ajwt-bearer&client_assertion=${TOKEN}" \
    | python3 -c "import sys,json; print(json.load(sys.stdin)['access_token'])")

# Test SuiteQL query against the login audit trail
curl -s -X POST \
    "https://${ACCOUNT_ID}.suitetalk.api.netsuite.com/services/rest/query/v1/suiteql?limit=5" \
    -H "Authorization: Bearer ${ACCESS_TOKEN}" \
    -H "Content-Type: application/json" \
    -H "Prefer: transient" \
    -d '{"q": "SELECT Date, Status, User, IPAddress FROM LoginAudit ORDER BY Date DESC"}' \
    | python3 -m json.tool

# Test SuiteQL query against system notes
curl -s -X POST \
    "https://${ACCOUNT_ID}.suitetalk.api.netsuite.com/services/rest/query/v1/suiteql?limit=5" \
    -H "Authorization: Bearer ${ACCESS_TOKEN}" \
    -H "Content-Type: application/json" \
    -H "Prefer: transient" \
    -d '{"q": "SELECT Date, RecordID, Field, BUILTIN.DF(SystemNote.Name) AS EmployeeName FROM SystemNote ORDER BY Date DESC"}' \
    | python3 -m json.tool

Eine erfolgreiche Antwort gibt ein JSON-Objekt mit einem items-Array zurück, das Audit-Datensätze für die Anmeldung enthält.

Die zweite Anfrage muss aktuelle Systemanmerkungen zurückgeben, einschließlich Änderungen, die von anderen Nutzern als dem Integrationsnutzer vorgenommen wurden. Ein leeres items-Array oder eines, in dem nur der Integrationsnutzer in EmployeeName aufgeführt ist, bedeutet, dass der Rolle die Berechtigung System Notes for Analytics and REST fehlt.

Google Cloud Storage-Bucket erstellen

  1. Rufen Sie die Google Cloud Console auf.
  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. netsuite-audit-logs.
    Standorttyp Wählen Sie die Option aus, die am besten zu Ihren Anforderungen passt (Region, Dual-Region, Multi-Region).
    Standort Wählen Sie den Speicherort aus, z. B. us-central1.
    Speicherklasse Standard (empfohlen für Logs, auf die häufig zugegriffen wird)
    Zugriffskontrolle Einheitlich (empfohlen)
    Schutztools Optional: Objektversionsverwaltung oder Aufbewahrungsrichtlinie aktivieren

  6. Klicken Sie auf Erstellen.

Dienstkonto für die Cloud Run-Funktion erstellen

Die Cloud Run-Funktion benötigt ein Dienstkonto mit Berechtigungen zum Schreiben in einen Cloud Storage-Bucket und zum Aufrufen durch Pub/Sub.

Dienstkonto 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 netsuite-audit-collector-sa ein.
    • Beschreibung des Dienstkontos: Geben Sie Service account for Cloud Run function to collect Oracle NetSuite audit logs 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:
    1. Klicken Sie auf Rolle auswählen.
    2. Suchen Sie nach Storage Object Admin und wählen Sie die Rolle aus.
    3. Klicken Sie auf + Weitere Rolle hinzufügen.
    4. Suchen Sie nach Cloud Run Invoker und wählen Sie die Rolle aus.
    5. Klicken Sie auf + Weitere Rolle hinzufügen.
    6. Suchen Sie nach Cloud Functions Invoker und wählen Sie die Rolle aus.
  6. Klicken Sie auf Weiter.
  7. Klicken Sie auf Fertig.

Diese Rollen sind erforderlich für:

  • Storage-Objekt-Administrator: Protokolle in den Cloud Storage-Bucket schreiben und Statusdateien verwalten
  • Cloud Run Invoker: Pub/Sub darf die Funktion aufrufen.
  • Cloud Functions-Invoker: Funktionsaufruf zulassen

IAM-Berechtigungen für den Cloud Storage-Bucket gewähren

  1. Rufen Sie Cloud Storage > Buckets auf.
  2. Klicken Sie auf den Namen Ihres Buckets (netsuite-audit-logs).
  3. Wechseln Sie zum Tab Berechtigungen.
  4. Klicken Sie auf Zugriff erlauben.
  5. Geben Sie die folgenden Konfigurationsdetails an:
    • Hauptkonten hinzufügen: Geben Sie die E-Mail-Adresse des Dienstkontos (netsuite-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com) ein.
    • Rollen zuweisen: Wählen Sie Storage Object Admin aus.
  6. Klicken Sie auf Speichern.

Pub/Sub-Thema erstellen

Erstellen Sie ein Pub/Sub-Thema, in dem Cloud Scheduler veröffentlicht und das von der Cloud Run-Funktion abonniert wird.

  1. Rufen Sie in der GCP Console Pub/Sub > Themen auf.
  2. Klicken Sie auf Thema erstellen.
  3. Geben Sie die folgenden Konfigurationsdetails an:
    • Themen-ID: Geben Sie netsuite-audit-trigger ein.
    • Übernehmen Sie die anderen Einstellungen.
  4. Klicken Sie auf Erstellen.

Cloud Run-Funktion zum Erfassen von Logs erstellen

Die Cloud Run-Funktion wird durch Pub/Sub-Nachrichten von Cloud Scheduler ausgelöst, um Audit-Logs aus der Oracle NetSuite SuiteQL REST API abzurufen und in Cloud Storage zu schreiben.

  1. Rufen Sie in der GCP Console Cloud Run auf.
  2. Klicken Sie auf Dienst erstellen.
  3. Wählen Sie Funktion aus, um eine Funktion mit einem Inline-Editor zu erstellen.

  4. Geben Sie im Abschnitt Konfigurieren die folgenden Konfigurationsdetails an:

    Einstellung Wert
    Dienstname netsuite-audit-collector
    Region Wählen Sie die Region aus, die Ihrem Cloud Storage-Bucket entspricht (z. B. us-central1).
    Laufzeit Wählen Sie Python 3.12 oder höher aus.

  5. Im Abschnitt Trigger (optional):

    1. Klicken Sie auf + Trigger hinzufügen.
    2. Wählen Sie Cloud Pub/Sub aus.
    3. Wählen Sie unter Cloud Pub/Sub-Thema auswählen die Option netsuite-audit-trigger aus.
    4. Klicken Sie auf Speichern.

  6. Im Abschnitt Authentifizierung:

    1. Wählen Sie Authentifizierung erforderlich aus.
    2. Wählen Sie Identity and Access Management (IAM) aus.

  7. Scrollen Sie nach unten und maximieren Sie Container, Netzwerk, Sicherheit.

  8. Rufen Sie den Tab Sicherheit auf:

    • Dienstkonto: Wählen Sie netsuite-audit-collector-sa aus.

  9. Rufen Sie den Tab Container auf:

    1. Klicken Sie auf Variablen und Secrets.
    2. Klicken Sie für jede Umgebungsvariable auf + Variable hinzufügen:
    Variablenname Beispielwert Beschreibung
    GCS_BUCKET netsuite-audit-logs Name des Cloud Storage-Buckets
    GCS_PREFIX netsuite-audit Präfix für Logdateien
    STATE_KEY netsuite-audit-state.json Zustandspfad, außerhalb des Logpräfixes
    NS_ACCOUNT_ID 123456 NetSuite-Konto-ID
    NS_CLIENT_ID your-client-id OAuth 2.0-Client-ID
    NS_CERTIFICATE_ID your-certificate-id Zertifikats-ID (kid) aus der M2M-Zuordnung
    NS_PRIVATE_KEY -----BEGIN EC PRIVATE KEY-----... Inhalt des privaten ECDSA-Schlüssels
    LOOKBACK_HOURS 24 Anfänglicher Rückschauzeitraum
    PAGE_SIZE 1000 Einträge pro SuiteQL-Seite (max. 1.000)
    MAX_RECORDS 50000 Maximale Anzahl von Datensätzen pro Ausführung

  10. Scrollen Sie im Bereich Variablen und Secrets nach unten zu Anfragen:

    • Zeitlimit für die Anfrage: Geben Sie 600 Sekunden (10 Minuten) ein.

  11. Rufen Sie den Tab Einstellungen auf:

    • Im Abschnitt Ressourcen:
      • Arbeitsspeicher: Wählen Sie 512 MiB oder höher aus.
      • CPU: Wählen Sie 1 aus.

  12. Im Abschnitt Versionsskalierung:

    • Mindestanzahl von Instanzen: Geben Sie 0 ein.
    • Maximale Anzahl von Instanzen: Geben Sie 100 ein.

  13. Klicken Sie auf Erstellen.

  14. Warten Sie, bis der Dienst erstellt wurde (1–2 Minuten).

  15. Nachdem der Dienst erstellt wurde, wird automatisch der Inline-Code-Editor geöffnet.

Funktionscode hinzufügen

  1. Geben Sie main in das Feld Einstiegspunkt ein.

  2. Erstellen Sie im Inline-Code-Editor zwei Dateien:

    • Erste Datei – main.py:

      import functions_framework
from google.cloud import storage
import json
import os
import urllib3
import urllib.parse
import jwt
import time
import hashlib
from datetime import datetime, timezone, timedelta

http = urllib3.PoolManager(
    timeout=urllib3.Timeout(connect=10.0, read=60.0),
    retries=False,
)

storage_client = storage.Client()

GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'netsuite-audit')
STATE_KEY = os.environ.get('STATE_KEY', 'netsuite-audit-state.json')
NS_ACCOUNT_ID = os.environ.get('NS_ACCOUNT_ID')
NS_CLIENT_ID = os.environ.get('NS_CLIENT_ID')
NS_CERTIFICATE_ID = os.environ.get('NS_CERTIFICATE_ID')
NS_PRIVATE_KEY = os.environ.get('NS_PRIVATE_KEY', '').replace('\\n', '\n')
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '1000'))
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '50000'))

# Neither SuiteQL query selects a unique row identifier, so the content hash
# is the only identity available.
ID_FIELDS = ()

def event_keys(event):
    """Return every identity this event can be recognised by."""
    keys = {'sha256:' + hashlib.sha256(
        json.dumps(event, sort_keys=True, ensure_ascii=False).encode('utf-8')
    ).hexdigest()}
    for field in ID_FIELDS:
        value = event.get(field)
        if value:
            keys.add(f'{field}:{value}')
    return keys

@functions_framework.cloud_event
def main(cloud_event):
    if not all([GCS_BUCKET, NS_ACCOUNT_ID, NS_CLIENT_ID, NS_CERTIFICATE_ID, NS_PRIVATE_KEY]):
        print('Error: Missing required environment variables')
        return

    try:
        bucket = storage_client.bucket(GCS_BUCKET)
        state = load_state(bucket)
        seen_keys = load_seen(bucket)
        now = datetime.now(timezone.utc)

        if isinstance(state, dict) and state.get('last_event_time'):
            try:
                last_val = state['last_event_time']
                if last_val.endswith('Z'):
                    last_val = last_val[:-1] + '+00:00'
                last_time = datetime.fromisoformat(last_val)
                last_time = last_time - timedelta(minutes=2)
            except Exception as e:
                print(f"Warning: Could not parse last_event_time: {e}")
                last_time = now - timedelta(hours=LOOKBACK_HOURS)
        else:
            last_time = now - timedelta(hours=LOOKBACK_HOURS)

        print(f"Fetching logs from {last_time.isoformat()} to {now.isoformat()}")

        access_token = get_access_token()

        login_audits = fetch_login_audit(access_token, last_time, now)
        system_notes = fetch_system_notes(access_token, last_time, now)

        all_records = []
        for record in login_audits:
            record['_netsuite_record_type'] = 'login_audit'
            all_records.append(record)
        for record in system_notes:
            record['_netsuite_record_type'] = 'system_note'
            all_records.append(record)

        if not all_records:
            print("No new records found.")
            save_state(bucket, now.isoformat())
            return

        # Drop the rows the overlap window re-read from the previous run
        fresh = [r for r in all_records if not (event_keys(r) & seen_keys)]
        all_keys = set()
        for record in all_records:
            all_keys |= event_keys(record)

        if not fresh:
            print("No new records found. Watermark left unchanged.")
            return

        timestamp = now.strftime('%Y%m%d_%H%M%S')
        object_key = f"{GCS_PREFIX}/netsuite_audit_{timestamp}.ndjson"
        blob = bucket.blob(object_key)

        ndjson = '\n'.join(
            [json.dumps(r, ensure_ascii=False, default=str) for r in fresh]
        ) + '\n'
        blob.upload_from_string(ndjson, content_type='application/x-ndjson')

        print(f"Wrote {len(fresh)} records to gs://{GCS_BUCKET}/{object_key}")

        # Remember every fetched row so the next overlap window skips it
        save_seen(bucket, all_keys)

        newest = find_newest_time(login_audits, system_notes)
        save_state(bucket, newest if newest else now.isoformat())

        print(f"Successfully processed {len(fresh)} records "
                f"(login_audit: {len(login_audits)}, system_note: {len(system_notes)})")

    except Exception as e:
        print(f'Error processing logs: {str(e)}')
        raise

def get_access_token():
    token_url = (
        f"https://{NS_ACCOUNT_ID}.suitetalk.api.netsuite.com"
        f"/services/rest/auth/oauth2/v1/token"
    )

    now = int(time.time())
    payload = {
        'iss': NS_CLIENT_ID,
        'scope': ['restlets', 'rest_webservices'],
        'aud': token_url,
        'iat': now,
        'exp': now + 3600,
    }

    client_assertion = jwt.encode(
        payload,
        NS_PRIVATE_KEY,
        algorithm='ES256',
        headers={'kid': NS_CERTIFICATE_ID, 'typ': 'JWT'},
    )

    body = urllib.parse.urlencode({
        'grant_type': 'client_credentials',
        'client_assertion_type': 'urn:ietf:params:oauth:client-assertion-type:jwt-bearer',
        'client_assertion': client_assertion,
    }).encode('utf-8')

    response = http.request(
        'POST', token_url,
        body=body,
        headers={'Content-Type': 'application/x-www-form-urlencoded'},
    )

    if response.status != 200:
        raise Exception(
            f"Token request failed: {response.status} - "
            f"{response.data.decode('utf-8')}"
        )

    data = json.loads(response.data.decode('utf-8'))
    access_token = data.get('access_token')
    if not access_token:
        raise Exception("No access_token in token response")

    print("Successfully obtained NetSuite OAuth 2.0 access token")
    return access_token

def run_suiteql(access_token, query, page_size=1000, max_records=50000):
    base_url = (
        f"https://{NS_ACCOUNT_ID}.suitetalk.api.netsuite.com"
        f"/services/rest/query/v1/suiteql"
    )

    headers = {
        'Authorization': f'Bearer {access_token}',
        'Content-Type': 'application/json',
        'Prefer': 'transient',
    }

    all_items = []
    offset = 0
    backoff = 1.0

    while True:
        if len(all_items) >= max_records:
            print(f"Reached max_records limit ({max_records})")
            break

        url = f"{base_url}?limit={page_size}&offset={offset}"
        body = json.dumps({'q': query}).encode('utf-8')

        try:
            response = http.request('POST', url, body=body, headers=headers)

            if response.status == 429:
                retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
                print(f"Rate limited (429). Retrying after {retry_after}s...")
                time.sleep(retry_after)
                backoff = min(backoff * 2, 60.0)
                continue

            backoff = 1.0

            if response.status != 200:
                print(f"SuiteQL error: {response.status} - "
                        f"{response.data.decode('utf-8')}")
                break

            data = json.loads(response.data.decode('utf-8'))
            items = data.get('items', [])

            if not items:
                break

            all_items.extend(items)
            page_num = (offset // page_size) + 1
            print(f"Page {page_num}: Retrieved {len(items)} records "
                    f"(total: {len(all_items)})")

            has_more = data.get('hasMore', False)
            if not has_more:
                break

            offset += len(items)

        except Exception as e:
            print(f"Error executing SuiteQL query: {e}")
            break

    return all_items

def fetch_login_audit(access_token, start_time, end_time):
    start_str = start_time.strftime('%m/%d/%Y %H:%M:%S')
    end_str = end_time.strftime('%m/%d/%Y %H:%M:%S')

    query = (
        "SELECT "
        "LoginAudit.Date, "
        "LoginAudit.Status, "
        "LoginAudit.User, "
        "BUILTIN.DF(LoginAudit.User) AS Username, "
        "LoginAudit.EmailAddress, "
        "LoginAudit.Role, "
        "BUILTIN.DF(LoginAudit.Role) AS Rolename, "
        "LoginAudit.IPAddress, "
        "LoginAudit.UserAgent, "
        "LoginAudit.Detail, "
        "LoginAudit.oAuthAppName, "
        "LoginAudit.oAuthAccessTokenName, "
        "LoginAudit.requestUri "
        "FROM LoginAudit "
        f"WHERE LoginAudit.Date >= TO_DATE('{start_str}', 'MM/DD/YYYY HH24:MI:SS') "
        f"AND LoginAudit.Date <= TO_DATE('{end_str}', 'MM/DD/YYYY HH24:MI:SS') "
        "ORDER BY LoginAudit.Date ASC"
    )

    print("Fetching LoginAudit records...")
    return run_suiteql(access_token, query, PAGE_SIZE, MAX_RECORDS)

def fetch_system_notes(access_token, start_time, end_time):
    start_str = start_time.strftime('%m/%d/%Y %H:%M:%S')
    end_str = end_time.strftime('%m/%d/%Y %H:%M:%S')

    query = (
        "SELECT "
        "SystemNote.Date, "
        "SystemNote.RecordTypeID, "
        "BUILTIN.DF(SystemNote.RecordTypeID) AS RecordType, "
        "SystemNote.RecordID, "
        "SystemNote.Field, "
        "SystemNote.OldValue, "
        "SystemNote.NewValue, "
        "SystemNote.Name AS EmployeeID, "
        "BUILTIN.DF(SystemNote.Name) AS EmployeeName, "
        "SystemNote.Role, "
        "BUILTIN.DF(SystemNote.Role) AS RoleName, "
        "SystemNote.Context, "
        "BUILTIN.DF(SystemNote.Context) AS ContextName "
        "FROM SystemNote "
        f"WHERE SystemNote.Date >= TO_DATE('{start_str}', 'MM/DD/YYYY HH24:MI:SS') "
        f"AND SystemNote.Date <= TO_DATE('{end_str}', 'MM/DD/YYYY HH24:MI:SS') "
        "ORDER BY SystemNote.Date ASC"
    )

    print("Fetching SystemNote records...")
    return run_suiteql(access_token, query, PAGE_SIZE, MAX_RECORDS)

def find_newest_time(login_audits, system_notes):
    newest = None
    for r in login_audits:
        t = r.get('date')
        if t and (newest is None or t > newest):
            newest = t
    for r in system_notes:
        t = r.get('date')
        if t and (newest is None or t > newest):
            newest = t
    return newest

def load_state(bucket):
    try:
        blob = bucket.blob(STATE_KEY)
        if blob.exists():
            return json.loads(blob.download_as_text())
    except Exception as e:
        print(f"Warning: Could not load state: {e}")
        raise
    return {}

def save_state(bucket, last_event_time_iso):
    try:
        state = {
            'last_event_time': last_event_time_iso,
            'last_run': datetime.now(timezone.utc).isoformat()
        }
        blob = bucket.blob(STATE_KEY)
        blob.upload_from_string(
            json.dumps(state, indent=2),
            content_type='application/json'
        )
        print(f"Saved state: last_event_time={last_event_time_iso}")
    except Exception as e:
        print(f"Warning: Could not save state: {e}")
        raise

def load_seen(bucket):
    try:
        blob = bucket.blob(STATE_KEY + '.seen')
        if blob.exists():
            return set(json.loads(blob.download_as_text()))
    except Exception as e:
        print(f"Warning: Could not load seen keys: {e}")
        raise
    return set()

def save_seen(bucket, keys):
    try:
        blob = bucket.blob(STATE_KEY + '.seen')
        blob.upload_from_string(
            json.dumps(sorted(keys)),
            content_type='application/json'
        )
        print(f"Saved {len(keys)} event keys for deduplication")
    except Exception as e:
        print(f"Warning: Could not save seen keys: {e}")
        raise

    • Zweite Datei:requirements.txt:

      functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0
PyJWT[crypto]>=2.8.0

  3. Klicken Sie auf Bereitstellen, um die Funktion zu speichern und bereitzustellen.

  4. Warten Sie, bis die Bereitstellung abgeschlossen ist (2–3 Minuten).

Cloud Scheduler-Job erstellen

Cloud Scheduler veröffentlicht in regelmäßigen Abständen Nachrichten im Pub/Sub-Thema und löst so die Cloud Run-Funktion aus.

  1. Rufen Sie in der GCP Console Cloud Scheduler auf.
  2. Klicken Sie auf Job erstellen.

  3. Geben Sie die folgenden Konfigurationsdetails an:

    Einstellung Wert
    Name netsuite-audit-collector-hourly
    Region Dieselbe Region wie die Cloud Run-Funktion auswählen
    Frequenz 0 * * * * (stündlich zur vollen Stunde)
    Zeitzone Zeitzone auswählen (UTC empfohlen)
    Zieltyp Pub/Sub
    Thema netsuite-audit-trigger auswählen
    Inhalt der Nachricht {} (leeres JSON-Objekt)

  4. Klicken Sie auf Erstellen.

Optionen für die Häufigkeit des Zeitplans

Wählen Sie die Häufigkeit basierend auf dem Logvolumen und den Latenzanforderungen aus:

Häufigkeit Cron-Ausdruck Anwendungsfall
Alle 15 Minuten */15 * * * * Hohes Volumen, niedrige Latenz
Stündlich 0 * * * * Standard (empfohlen)
Alle 6 Stunden 0 */6 * * * Geringes Volumen, Batchverarbeitung
Täglich 0 0 * * * Erhebung von Verlaufsdaten

Integration testen

  1. Suchen Sie in der Cloud Scheduler-Konsole nach Ihrem Job (netsuite-audit-collector-hourly).
  2. Klicken Sie auf Ausführung erzwingen, um den Job manuell auszulösen.
  3. Warten Sie einige Sekunden.
  4. Rufen Sie Cloud Run > Dienste auf.
  5. Klicken Sie auf netsuite-audit-collector.
  6. Klicken Sie auf den Tab Logs.

  7. Prüfen Sie, ob die Funktion erfolgreich ausgeführt wurde. Darauf sollten Sie achten:

    Fetching logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
Successfully obtained NetSuite OAuth 2.0 access token
Fetching LoginAudit records...
Page 1: Retrieved X records (total: X)
Fetching SystemNote records...
Page 1: Retrieved X records (total: X)
Wrote X records to gs://netsuite-audit-logs/netsuite-audit/netsuite_audit_YYYYMMDD_HHMMSS.ndjson
Successfully processed X records (login_audit: X, system_note: X)

  8. Rufen Sie Cloud Storage > Buckets auf.

  9. Klicken Sie auf netsuite-audit-logs.

  10. Rufen Sie den Ordner netsuite-audit/ auf.

  11. Prüfen Sie, ob eine neue .ndjson-Datei mit dem aktuellen Zeitstempel erstellt wurde.

Wenn in den Logs Fehler angezeigt werden oder ein Datensatztyp in der Ausgabe fehlt:

  • HTTP 401: Prüfen Sie, ob die Umgebungsvariablen NS_CLIENT_ID, NS_CERTIFICATE_ID und NS_PRIVATE_KEY korrekt sind.
  • HTTP 403: Prüfen Sie, ob die NetSuite-Rolle Anmeldeberechtigungen mit den Berechtigungen „OAuth 2.0 Access Tokens (Full)“, „REST Web Services (Full)“ und „Login Audit Trail (Full)“ hat.
  • Keine system_note-Einträge in der Ausgabe: Der Rolle fehlt die Berechtigung „System Notes for Analytics and REST (View)“. Die SuiteQL-Abfrage wird weiterhin ausgeführt und gibt nur die Notizen für Änderungen zurück, die der Integrationsnutzer selbst vorgenommen hat. Es werden also keine Fehler in den Logs protokolliert.
  • HTTP 429: Ratenbegrenzung – die Funktion wird automatisch mit exponentiellem Backoff wiederholt.
  • Fehlende Umgebungsvariablen: Prüfen Sie, ob alle erforderlichen Variablen in der Konfiguration der Cloud Run-Funktion festgelegt sind.

Google SecOps-Dienstkonto abrufen

  1. Rufen Sie die SIEM-Einstellungen > Feeds auf.
  2. Klicken Sie auf Neuen Nachrichten-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. NetSuite Audit Logs.
  5. Wählen Sie Google Cloud Storage V2 als Quelltyp aus.
  6. Wählen Sie Oracle NetSuite als Logtyp aus.
  7. Klicken Sie auf Dienstkonto abrufen.

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

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

  9. Kopieren Sie diese E‑Mail-Adresse für den nächsten Schritt.

Dem Google SecOps-Dienstkonto IAM-Berechtigungen gewähren

Das Google SecOps-Dienstkonto benötigt zwei Rollen für Ihren Cloud Storage-Bucket: Storage-Objekt-Betrachter zum Lesen der Logobjekte und eine Rolle auf Bucket-Ebene zum Lesen der Bucket-Metadaten.

  1. Rufen Sie Cloud Storage > Buckets auf.
  2. Klicken Sie auf netsuite-audit-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.
    • Rollen zuweisen: Wählen Sie die beiden folgenden Rollen aus:
      • Storage Object Viewer: Liest die Logobjekte.
      • Leser alter Storage-Buckets: Berechtigt zum Lesen der Bucket-Metadaten. Wenn Sie die Löschoption Übertragene Dateien und leere Verzeichnisse löschen ausgewählt haben, wählen Sie stattdessen Storage Legacy Bucket Writer aus. Dadurch wird auch die Löschberechtigung erteilt.

  6. Klicken Sie auf Speichern.

Google SecOps-Feedkonfiguration abschließen

  1. Kehren Sie zum Assistenten zum Erstellen von Google SecOps-Feeds zurück.
  2. Klicken Sie auf Weiter.

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

    • Storage-Bucket-URL: Geben Sie den Cloud Storage-Bucket-URI mit dem Präfixpfad ein:

      gs://netsuite-audit-logs/netsuite-audit/

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

      • Dateien nie löschen: Dateien werden nie aus der Quelle gelöscht (empfohlen für Tests).
      • Übertragene Dateien und leere Verzeichnisse löschen: Dateien und leere Verzeichnisse werden nach einem erfolgreichen Abruf aus der Quelle gelöscht.

    • Höchstalter der Datei: Dateien einbeziehen, die in den letzten Tagen geändert wurden (Standardwert: 180 Tage)

    • Asset-Namespace: Der Asset-Namespace

    • Labels für Datenaufnahme: Das Label, das auf die Ereignisse aus diesem Feed angewendet werden soll.

  4. Klicken Sie auf Weiter.

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

NetSuite API-Ratenlimits

Für die NetSuite REST API gelten die folgenden Governance-Limits:

Limit-Typ Wert
Nebenläufigkeitslimit 10 gleichzeitige Anfragen pro Integration
SuiteQL-Abfrageergebnisse pro Seite Maximal 1.000 Zeilen
SuiteQL-Gesamtergebnisse 100.000 Zeilen pro Abfrage

Die Cloud Run-Funktion verarbeitet die Ratenbegrenzung automatisch mit exponentiellem Backoff.

UDM-Zuordnungstabelle

Logfeld UDM-Zuordnung Logik
Login_Audit_Trail_Access_Token_Name_label additional.fields Zusammengeführt
Login_Audit_Trail_Detail_label additional.fields Zusammengeführt
additional_preview additional.fields Zusammengeführt
detail_label additional.fields Zusammengeführt
role_change_label additional.fields Zusammengeführt
sec_challenge_label additional.fields Zusammengeführt
oauthaccesstokenname extensions.auth.auth_details Direkt zugeordnet
has_principal extensions.auth.type Zugeordnet: true → AUTHTYPE_UNSPECIFIED
column7 metadata.description Direkt zugeordnet
msg metadata.description Direkt zugeordnet
Role_Change_Date metadata.event_timestamp Geparst als M/d/yyyy h:mm a
date metadata.event_timestamp Geparst als M/d/yyyy h:mm:ss a
result.Login_Audit_Trail_Date metadata.event_timestamp Geparst als LDAP
ts metadata.event_timestamp Als MMM d HH:mm:ss geparst
has_principal metadata.event_type Zugeordnet: true → USER_LOGIN, true → STATUS_UPDATE
has_principal_user metadata.event_type Zugeordnet: true → USER_UNCATEGORIZED
column1 metadata.product_event_type Direkt zugeordnet
log_type metadata.product_event_type Direkt zugeordnet
useragent network.http.parsed_user_agent Direkt zugeordnet
Login_Audit_Trail_User_Agent network.http.user_agent Direkt zugeordnet
result.Login_Audit_Trail_User_Agent network.http.user_agent Direkt zugeordnet
useragent network.http.user_agent Direkt zugeordnet
Login_Audit_Trail_Application principal.application Direkt zugeordnet
column10 principal.application Direkt zugeordnet
oauthappname principal.application Direkt zugeordnet
result.host principal.asset.hostname Direkt zugeordnet
column3 principal.asset.ip Zusammengeführt
ipaddress principal.asset.ip Zusammengeführt
result.Login_Audit_Trail_IP_Address principal.asset.ip Direkt zugeordnet
src_ip principal.asset.ip Zusammengeführt
result.host principal.hostname Direkt zugeordnet
column3 principal.ip Zusammengeführt
ipaddress principal.ip Zusammengeführt
result.Login_Audit_Trail_IP_Address principal.ip Direkt zugeordnet
src_ip principal.ip Zusammengeführt
Login_Audit_Trail_Role_label principal.user.attribute.labels Zusammengeführt
login_audits_role_label principal.user.attribute.labels Zusammengeführt
supervisor_label principal.user.attribute.labels Zusammengeführt
role principal.user.attribute.roles Zusammengeführt
Email principal.user.email_addresses Zugeordnet: ^.+@.+$ → Email
column2 principal.user.email_addresses Zusammengeführt
emailaddress principal.user.email_addresses Zugeordnet: ^.+@.+$ → emailaddress
result.Login_Audit_Trail_Email_Address principal.user.email_addresses Zusammengeführt
Job_Title principal.user.title Direkt zugeordnet
Name principal.user.user_display_name Direkt zugeordnet
Login_Audit_Trail_User principal.user.userid Direkt zugeordnet
result.Login_Audit_Trail_User principal.user.userid Direkt zugeordnet
user principal.user.userid Direkt zugeordnet
Login_Audit_Trail_Security_Challenge_label security_result.about.resource.attribute.labels Zusammengeführt
security_result_action security_result.action Zusammengeführt
Login_Audit_Trail_Status security_result.summary Direkt zugeordnet
result.Login_Audit_Trail_Status security_result.summary Direkt zugeordnet
api_type_label target.resource.attribute.labels Zusammengeführt
Login_Audit_Trail_Request_URI target.url Direkt zugeordnet
result.Login_Audit_Trail_Request_URI target.url Direkt zugeordnet
– extensions.auth.type Konstante: AUTHTYPE_UNSPECIFIED
– metadata.event_type Konstante: USER_LOGIN
– network.http.parsed_user_agent Konstante: parseduseragent

Änderungsprotokoll

Änderungsprotokoll für diesen Parser ansehen

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