‫Google משתמשת בטכנולוגיית AI כדי לתרגם תוכן לשפה המועדפת עליך. בתרגומים כאלו עשויות להיות שגיאות.

איסוף יומנים של Oracle NetSuite

נתמך ב:

Google SecOps SIEM

במסמך הזה מוסבר איך להטמיע יומנים של Oracle NetSuite ב-Google Security Operations באמצעות Cloud Storage V2.

‫Oracle NetSuite היא פלטפורמה מבוססת-ענן לתכנון משאבים ארגוניים (ERP) שמספקת יכולות מקיפות לניהול עסקי, כולל ניהול פיננסי, ניהול קשרי לקוחות (CRM), מסחר אלקטרוני וניהול מלאי. מערכת NetSuite יוצרת יומני ביקורת של התחברויות משתמשים, שינויים במערכת ופעילות API. אפשר לשלוח שאילתות ליומנים האלה דרך SuiteQL API בארכיטקטורת REST ולהטמיע אותם ב-Google SecOps לצורכי מעקב אחרי אבטחה ותאימות.

לפני שמתחילים

חשוב לוודא שיש לכם את הדברים הבאים:

מכונה של Google SecOps
פרויקט ב- Google Cloud עם ממשקי ה-API של Cloud Storage,‏ Cloud Run,‏ Pub/Sub ו-Cloud Scheduler מופעלים.
הרשאות ליצירה ולניהול של קטגוריות של Cloud Storage
הרשאות לניהול כללי מדיניות של ניהול זהויות והרשאות גישה (IAM) בקטגוריות של Cloud Storage
הרשאות ליצירת שירותי Cloud Run, נושאי Pub/Sub ומשימות ב-Cloud Scheduler
הרשאת אדמין בחשבון Oracle NetSuite
התכונות REST Web Services ו-OAuth 2.0 מופעלות בחשבון NetSuite שלכם
מזהה החשבון שלכם ב-NetSuite (נמצא ב-Setup > Company > Company Information)

הפעלת התכונות הנדרשות של NetSuite

נכנסים לחשבון NetSuite כאדמין.
עוברים אל Setup > Company > Enable Features (הגדרה > חברה > הפעלת תכונות).
לוחצים על כרטיסיית המשנה SuiteCloud.
בקטע SuiteTalk (Web Services)‎, מסמנים את התיבה הבאה:
- שירותי אינטרנט של REST
בקטע ניהול אימות, מסמנים את התיבה הבאה:
- OAuth 2.0
לוחצים על Save.

הגדרת פרטי הכניסה של Oracle NetSuite OAuth 2.0

כדי לאפשר לפונקציית Cloud Run לאחזר יומני ביקורת מ-NetSuite, צריך ליצור רשומת שילוב עם אימות של פרטי כניסה של לקוח OAuth 2.0.

יצירת רשומת שילוב

נכנסים לחשבון NetSuite כאדמין.
עוברים אל Setup (הגדרה) > Integration (שילוב) > Manage Integrations (ניהול שילובים) > New (חדש).
בשדה Name, מזינים Google SecOps Integration.
בשדה Description כותבים Integration with Google SecOps for audit log ingestion.
בשדה State, בוחרים באפשרות Enabled.
לוחצים על כרטיסיית המשנה אימות.
בקטע OAuth 2.0, מגדירים את הפרטים הבאים:
- בוחרים באפשרות Client Credentials (Machine to Machine) Grant (הרשאה של פרטי כניסה של לקוח (בין מחשבים)).
- בוחרים באפשרות REST Web Services (שירותי אינטרנט של REST).
מבטלים את הסימון של כל התיבות בקטע Token-based Authentication.
לוחצים על Save.

תיעוד של פרטי הכניסה ל-OAuth 2.0

אחרי ששומרים את רשומת השילוב, פרטי הכניסה ל-OAuth 2.0 מוצגים ב-NetSuite:

מזהה לקוח: מזהה הלקוח הייחודי שלכם
‫Client Secret: מפתח ה-API הסודי שלכם

הערה: צריך להעתיק ולשמור מיד את מזהה הלקוח ואת סוד הלקוח. מערכת NetSuite מציגה את הערכים האלה רק פעם אחת. אם מאבדים אותם, צריך לאפס את פרטי הכניסה ברשומת השילוב.

יצירת אישור לפרטי כניסה של לקוח ב-OAuth 2.0

תהליך פרטי הכניסה של הלקוח ב-OAuth 2.0 דורש אישור לחתימת JWT. יוצרים אישור ECDSA באמצעות OpenSSL:

במכונה שמותקן בה OpenSSL, מריצים את הפקודה הבאה כדי ליצור מפתח פרטי ואישור:

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

כשמוצגת בקשה, מזינים את שדות הנושא של האישור (מדינה, מדינה בארה"ב, ארגון וכו'). בשדה Common Name, מזינים Google SecOps Integration.
הפקודה יוצרת שני קבצים:
- ‫private.pem: מפתח פרטי (צריך לשמור עליו מאובטח, הוא משמש לחתימה על JWT בפונקציית Cloud Run)
- ‫public.pem: אישור ציבורי (צריך להעלות ל-NetSuite)
הערה: מערכת NetSuite תומכת באישורי ECDSA ‏(ES256, ‏ ES512) ו-RSA-PSS ‏(PS256). המסמך הזה משתמש ב-ECDSA עם העקומה prime256v1‏ (ES256). התוקף המקסימלי של אישורים הוא שנתיים.

יצירת מיפוי של פרטי כניסה של לקוח ב-OAuth 2.0

ב-NetSuite, עוברים אל Setup > Integration > Manage Authentication > OAuth 2.0 Client Credentials (M2M) Setup.
לוחצים על חדש.
בתיבת הדו-שיח, קובעים את ההגדרות הבאות:
- ישות: בוחרים את העובד או המשתמש שישויכו לבקשות ה-API.
- תפקיד: בוחרים תפקיד עם ההרשאה התחברות באמצעות אסימוני גישה של OAuth 2.0 (ראו הרשאות נדרשות). בלי ההרשאה הזו, התפקיד לא יופיע ברשימה.
- אפליקציה: בוחרים באפשרות Google SecOps Integration.
לוחצים על בחירת קובץ ומעלים את קובץ האישור של public.pem.
לוחצים על Save.
אחרי השמירה, שימו לב לערך של מזהה האישור שמוצג ברשימת המיפוי. הערך הזה משמש כ-kid (מזהה מפתח) בכותרת ה-JWT.

ההרשאות הנדרשות

לתפקיד שמוקצה למיפוי OAuth 2.0 צריכות להיות ההרשאות הבאות:

הרשאה	רמת גישה	מטרה
התחברות באמצעות אסימוני גישה מסוג OAuth 2.0	מלא	מתן הרשאה לתפקיד להשתמש בהתחברות באמצעות אסימון OAuth 2.0
שירותי אינטרנט של REST	מלא	ביצוע בקשות API בארכיטקטורת REST
נתיב ביקורת של התחברות	מלא	גישה לנתונים של נתיב הביקורת להתחברות
SuiteAnalytics Workbook	הצגה	גישה לנתוני ניתוח באמצעות SuiteQL
הערות מערכת ל-Analytics ול-REST	הצגה	שליחת שאילתות על רשומות SystemNote באמצעות SuiteQL

יצירת תפקיד בהתאמה אישית לשילוב

עוברים אל Setup > Users/Roles > Manage Roles > New.
בשדה Name, מזינים Google SecOps Integration Role.
לוחצים על כרטיסיית המשנה Permissions.
לוחצים על כרטיסיית המשנה הגדרה.
מוסיפים את ההרשאות הבאות:
- התחברות באמצעות טוקני גישה מסוג OAuth 2.0: צריך להגדיר את האפשרות מלא
- REST Web Services: מגדירים את האפשרות Full
- Login Audit Trail: הגדרה ל-Full
לוחצים על כרטיסיית המשנה דוחות.
מוסיפים את ההרשאה הבאה:
- SuiteAnalytics Workbook: הגדרה ל-View
לוחצים על כרטיסיית המשנה רשימות.
מוסיפים את ההרשאה הבאה:
- System Notes for Analytics and REST (הערות מערכת ל-Analytics ול-REST): מגדירים את ההרשאה View (הצגה)
לוחצים על Save.

אימות ההרשאות

כדי לוודא שלתפקיד יש את ההרשאות הנדרשות:

נכנסים ל-NetSuite.
עוברים אל Setup > Users/Roles > Manage Roles.
מוצאים את Google SecOps Integration Role ולוחצים על עריכה.
מוודאים שההרשאות שמפורטות בקטע הרשאות נדרשות מופיעות בכרטיסיית המשנה Permissions.

בדיקת הגישה ל-API

לפני שממשיכים בתהליך השילוב, בודקים את פרטי הכניסה:

# 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

תגובה מוצלחת מחזירה אובייקט JSON עם מערך items שמכיל רשומות של ביקורת כניסות.

השאילתה השנייה צריכה להחזיר הערות מערכת מהזמן האחרון, כולל שינויים שבוצעו על ידי משתמשים אחרים, לא רק המשתמש שביצע את האינטגרציה. מערך items ריק, או מערך שמופיע בו רק משתמש האינטגרציה ב-EmployeeName, מציין שבתפקיד חסרה ההרשאה System Notes for Analytics and REST.

יצירת קטגוריה של Google Cloud Storage

נכנסים לGoogle Cloud מסוף.
בוחרים את הפרויקט או יוצרים פרויקט חדש.
בתפריט הניווט, נכנסים ל-Cloud Storage > Buckets.
לוחצים על Create bucket.

ממלאים את פרטי ההגדרה הבאים:

הגדרה	ערך
Name your bucket	מזינים שם ייחודי ברמה גלובלית (לדוגמה, `netsuite-audit-logs`)
סוג המיקום	בוחרים לפי הצרכים (אזור, שני אזורים, מספר אזורים)
Location	בוחרים את המיקום (לדוגמה, `us-central1`)
סוג אחסון (storage class)	רגיל (מומלץ ליומנים שהגישה אליהם תכופה)
בקרת גישה	אחידה (מומלץ)
כלי הגנה	אופציונלי: הפעלת ניהול גרסאות של אובייקטים או מדיניות שמירת נתונים

לוחצים על יצירה.

יצירת חשבון שירות לפונקציית Cloud Run

לפונקציית Cloud Run נדרש חשבון שירות עם הרשאות כתיבה בקטגוריה של Cloud Storage, וצריך להפעיל אותה באמצעות Pub/Sub.

יצירת חשבון שירות

במסוף Google Cloud , נכנסים לדף IAM & Admin > Service Accounts.
לוחצים על Create Service Account.
מזינים את פרטי ההגדרה הבאים:
- Service account name: מזינים netsuite-audit-collector-sa
- Service account description: מזינים Service account for Cloud Run function to collect Oracle NetSuite audit logs
לוחצים על יצירה והמשך.
בקטע Grant this service account access to project, מוסיפים את התפקידים הבאים:
1. לוחצים על בחירת תפקיד.
2. מחפשים את התפקיד Storage Object Admin ובוחרים אותו.
3. לוחצים על + הוספת תפקיד.
4. מחפשים את התפקיד Cloud Run Invoker ובוחרים בו.
5. לוחצים על + הוספת תפקיד.
6. מחפשים את התפקיד Cloud Functions Invoker ובוחרים אותו.
לוחצים על Continue.
לוחצים על סיום.

התפקידים האלה נדרשים כדי:

אדמין של אובייקטים באחסון: כתיבת יומנים בקטגוריה של Cloud Storage וניהול קובצי תמונת מצב
Cloud Run Invoker: מאפשר ל-Pub/Sub להפעיל את הפונקציה
מפעיל פונקציות של Cloud Functions: מאפשר להפעיל פונקציות

הקצאת הרשאות IAM בקטגוריה של Cloud Storage

נכנסים לדף Cloud Storage > Buckets.
לוחצים על שם הקטגוריה (netsuite-audit-logs).
עוברים לכרטיסייה Permissions.
לוחצים על הענקת גישה.
מזינים את פרטי ההגדרה הבאים:
- הוספת חשבונות משתמשים: מזינים את כתובת האימייל של חשבון השירות (netsuite-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com)
- הקצאת תפקידים: בוחרים באפשרות אדמין של אובייקטים באחסון.
לוחצים על Save.

יוצרים נושא Pub/Sub

יוצרים נושא Pub/Sub ש-Cloud Scheduler יפרסם בו ופונקציית Cloud Run תירשם אליו.

במסוף GCP, נכנסים לדף Pub/Sub > Topics.
לוחצים על Create topic.
מזינים את פרטי ההגדרה הבאים:
- מזהה הנושא: מזינים netsuite-audit-trigger
- משאירים את שאר ההגדרות כברירת מחדל.
לוחצים על יצירה.

יצירת פונקציית Cloud Run לאיסוף יומנים

הפונקציה של Cloud Run תופעל על ידי הודעות Pub/Sub מ-Cloud Scheduler כדי לאחזר יומני ביקורת מ-Oracle NetSuite SuiteQL REST API ולכתוב אותם ב-Cloud Storage.

במסוף GCP, נכנסים ל-Cloud Run.
לוחצים על Create service.
בוחרים באפשרות Function (שימוש בעורך ישיר כדי ליצור פונקציה).

בקטע Configure, מזינים את פרטי ההגדרה הבאים:

הגדרה	ערך
שם השירות	`netsuite-audit-collector`
אזור	בוחרים את האזור שתואם לקטגוריה של Cloud Storage (לדוגמה, `us-central1`)
זמן ריצה	בוחרים באפשרות Python 3.12 או גרסה עדכנית יותר.

בקטע טריגר (אופציונלי):
1. לוחצים על + הוספת טריגר.
2. בוחרים באפשרות Cloud Pub/Sub.
3. בקטע Select a Cloud Pub/Sub topic, בוחרים את netsuite-audit-trigger.
4. לוחצים על Save.
בקטע אימות:
1. לוחצים על דרישת אימות.
2. בוחרים באפשרות ניהול זהויות והרשאות גישה (IAM).
הערה: Pub/Sub יטפל באופן אוטומטי באימות כשהפונקציה תופעל.
גוללים למטה ומרחיבים את הקטע Containers, Networking, Security.
עוברים לכרטיסייה אבטחה:
- חשבון שירות: בוחרים באפשרות netsuite-audit-collector-sa

עוברים לכרטיסייה מאגרי תגים:

לוחצים על Variables & Secrets.
לוחצים על ‎+ Add variable לכל משתנה סביבה:

שם משתנה	ערך לדוגמה	תיאור
`GCS_BUCKET`	`netsuite-audit-logs`	שם הקטגוריה של Cloud Storage
`GCS_PREFIX`	`netsuite-audit`	קידומת לקובצי יומן
`STATE_KEY`	`netsuite-audit-state.json`	נתיב המצב, מחוץ לקידומת היומן
`NS_ACCOUNT_ID`	`123456`	מזהה החשבון ב-NetSuite
`NS_CLIENT_ID`	`your-client-id`	מזהה לקוח OAuth 2.0
`NS_CERTIFICATE_ID`	`your-certificate-id`	מזהה האישור (kid) ממיפוי M2M
`NS_PRIVATE_KEY`	`-----BEGIN EC PRIVATE KEY-----...`	תוכן המפתח הפרטי של ECDSA
`LOOKBACK_HOURS`	`24`	תקופת המבט לאחור הראשונית
`PAGE_SIZE`	`1000`	רשומות לכל דף SuiteQL (מקסימום 1,000)
`MAX_RECORDS`	`50000`	מספר הרשומות המקסימלי לכל הרצה

בקטע Variables & Secrets, גוללים למטה אל Requests:
- ‫Request timeout: מזינים 600 שניות (10 דקות)
עוברים לכרטיסייה הגדרות:
- בקטע Resources:
  - זיכרון: בוחרים באפשרות ‎512 MiB או יותר
  - מעבד (CPU): בוחרים באפשרות 1
בקטע Revision scaling:
- Minimum number of instances: מזינים 0
- Maximum number of instances: מזינים 100
לוחצים על יצירה.
ממתינים ליצירת השירות (דקה או שתיים).
אחרי שיוצרים את השירות, עורך הקוד המוטבע נפתח באופן אוטומטי.

הוספת קוד הפונקציה

בשדה Entry point, מזינים main.

בעורך הקוד, יוצרים שני קבצים:

הקובץ הראשון – 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

הקובץ השני – requirements.txt:

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

לוחצים על Deploy כדי לשמור ולפרוס את הפונקציה.
מחכים שהפריסה תסתיים (2-3 דקות).

הערה: הגדרת הטריגר של Pub/Sub יוצרת באופן אוטומטי את המינויים וההרשאות הנדרשים.

יצירת משימה ב-Cloud Scheduler

‫Cloud Scheduler יפרסם הודעות בנושא Pub/Sub במרווחי זמן קבועים, ויפעיל את פונקציית Cloud Run.

במסוף GCP, נכנסים ל-Cloud Scheduler.
לוחצים על Create Job.

ממלאים את פרטי ההגדרה הבאים:

הגדרה	ערך
שם	`netsuite-audit-collector-hourly`
אזור	בוחרים את אותו אזור כמו של פונקציית Cloud Run.
תדירות	‫`0 * * * *` (כל שעה, בשעה עגולה)
אזור זמן	בחירת אזור זמן (מומלץ לבחור ב-UTC)
סוג היעד	Pub/Sub
נושא	בחירת `netsuite-audit-trigger`
גוף ההודעה	‫`{}` (אובייקט JSON ריק)

לוחצים על יצירה.

אפשרויות לתדירות התזמון

בוחרים את התדירות בהתאם לנפח היומן ולדרישות לגבי זמן האחזור:

תדירות	ביטוי Cron	תרחיש לדוגמה
כל 15 דקות	`/15 * * *`	נפח גבוה, זמן אחזור נמוך
כל שעה	`0 * * * *`	רגיל (מומלץ)
כל 6 שעות	`0 /6 * *`	נפח נמוך, עיבוד ברצף (batch processing)
יומי	`0 0 * * *`	איסוף נתונים היסטוריים

בדיקת השילוב

במסוף Cloud Scheduler, מוצאים את המשימה (netsuite-audit-collector-hourly).
לוחצים על Force run כדי להפעיל את המשימה באופן ידני.
מחכים כמה שניות.
נכנסים לדף Cloud Run > Services.
לוחצים על netsuite-audit-collector.
לוחצים על הכרטיסייה Logs.

מוודאים שהפונקציה בוצעה בהצלחה. חפשו את:

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)

נכנסים לדף Cloud Storage > Buckets.
לחץ על netsuite-audit-logs.
עוברים לתיקייה netsuite-audit/.
מוודאים שנוצר קובץ .ndjson חדש עם חותמת הזמן הנוכחית.

אם מופיעות שגיאות ביומנים, או אם חסר סוג רשומה בפלט:

‫HTTP 401: מוודאים שמשתני הסביבה NS_CLIENT_ID,‏ NS_CERTIFICATE_ID ו-NS_PRIVATE_KEY נכונים.
‫HTTP 403: מוודאים שלתפקיד ב-NetSuite יש הרשאות כניסה באמצעות טוקני גישה מסוג OAuth 2.0 (מלא), שירותי אינטרנט מסוג REST (מלא) ונתיב ביקורת כניסה (מלא).
אין רשומות של system_note בפלט: לתפקיד חסרה ההרשאה "הערות מערכת ל-Analytics ול-REST (צפייה)". השאילתה של SuiteQL עדיין מצליחה ומחזירה רק את ההערות לגבי השינויים שמשתמש האינטגרציה ביצע בעצמו, כך שלא מגיעה שגיאה ליומנים.
‫HTTP 429: הגבלת קצב של יצירת בקשות – הפונקציה תנסה שוב באופן אוטומטי עם השהיה מעריכית לפני ניסיון חוזר (exponential backoff).
משתני סביבה חסרים: מוודאים שכל המשתנים הנדרשים מוגדרים בהגדרות של פונקציית Cloud Run.

אחזור של חשבון השירות של Google SecOps

עוברים אל הגדרות > ניהול נתונים > פידים. אם אתם משתמשים בפלטפורמה של SIEM בלבד, עוברים אל הגדרות > פידים.
לוחצים על הוספת פיד חדש.
לוחצים על הגדרת פיד יחיד.
בשדה שם הפיד, מזינים שם לפיד (לדוגמה, NetSuite Audit Logs).
בוחרים באפשרות Google Cloud Storage V2 בתור סוג המקור.
בוחרים באפשרות Oracle NetSuite בתור סוג היומן.
לוחצים על Get Service Account (קבלת חשבון שירות).
תוצג כתובת אימייל ייחודית של חשבון שירות, לדוגמה:
```
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
```
מעתיקים את כתובת האימייל הזו כדי להשתמש בה בשלב הבא.

הערה: לכל מכונה של Google SecOps יש חשבון שירות ייחודי. אל תשתמשו בחשבונות שירות ממסמכים או מדוגמאות אחרים.

מתן הרשאות IAM לחשבון השירות של Google SecOps

לחשבון השירות של Google SecOps נדרשים שני תפקידים בקטגוריה של Cloud Storage: צפייה באובייקט אחסון כדי לקרוא את אובייקטי היומן, ותפקיד ברמת הקטגוריה כדי לקרוא את המטא-נתונים של הקטגוריה.

נכנסים לדף Cloud Storage > Buckets.
לחץ על netsuite-audit-logs.
עוברים לכרטיסייה Permissions.
לוחצים על הענקת גישה.
מזינים את פרטי ההגדרה הבאים:
- הוספת חשבונות משתמשים: מדביקים את כתובת האימייל של חשבון השירות של Google SecOps.
- הקצאת תפקידים: בוחרים את שני התפקידים הבאים:
  - צפייה באובייקטים של אחסון: קריאת אובייקטים של יומנים.
  - קריאה בקטגוריה באחסון מדור קודם: קריאת המטא-נתונים של הקטגוריה. אם בחרתם באפשרות המחיקה Delete transferred files and empty directories, בחרו במקום זאת באפשרות Storage Legacy Bucket Writer, שגם מעניקה את הרשאת המחיקה.
לוחצים על Save.

הערה: צריך להקצות את שני התפקידים. התפקיד Storage Object Viewer לא מעניק את ההרשאה ברמת הקטגוריה שהפיד צריך, והתפקידים Storage Legacy Bucket לא מעניקים הרשאה לקרוא את תוכן האובייקט.

השלמת ההגדרה של הפיד ב-Google SecOps

חוזרים לאשף יצירת הפידים של Google SecOps.
לוחצים על הבא.
מציינים ערכים לפרמטרים הבאים של הקלט:
- כתובת ה-URL של קטגוריית האחסון: מזינים את ה-URI של קטגוריית Cloud Storage עם נתיב הקידומת:
```
gs://netsuite-audit-logs/netsuite-audit/
```
הערה: תמיד צריך לכלול את הלוכסן (/) בסוף ה-URI.
- אפשרות מחיקת המקור: בוחרים את אפשרות המחיקה המועדפת:
  - אף פעם לא למחוק קבצים: אף פעם לא למחוק קבצים מהמקור (מומלץ לבדיקה).
  - Delete transferred files and empty directories: מחיקת קבצים וספריות ריקות מהמקור אחרי שהאחזור הושלם בהצלחה.
- גיל הקובץ המקסימלי: כולל קבצים ששונו במספר הימים האחרונים (ברירת המחדל היא 180 ימים)
- מרחב השמות של הנכס: מרחב השמות של הנכס
- תוויות הטמעה: התווית שתתווסף לאירועים מהפיד הזה
לוחצים על הבא.
בודקים את הגדרות הפיד החדש במסך סיום ולוחצים על שליחה.

מגבלות קצב של יצירת בקשות ב-NetSuite API

ל-NetSuite API בארכיטקטורת REST יש את מגבלות הפיקוח הבאות:

סוג המגבלה	ערך
מגבלת הפעולות בו-זמנית	‫10 בקשות מקבילות לכל שילוב
תוצאות שאילתת SuiteQL לכל דף	עד 1,000 שורות
סה"כ תוצאות של SuiteQL	‫100,000 שורות לכל שאילתה

פונקציית Cloud Run מטפלת באופן אוטומטי בהגבלת קצב של יצירת בקשות באמצעות השהיה מעריכית לפני ניסיון חוזר (exponential backoff).

טבלת מיפוי UDM

שדה ביומן	מיפוי UDM	Logic (פונקציה לוגית)
`Login_Audit_Trail_Access_Token_Name_label`	`additional.fields`	ממוזג
`Login_Audit_Trail_Detail_label`	`additional.fields`	ממוזג
`additional_preview`	`additional.fields`	ממוזג
`detail_label`	`additional.fields`	ממוזג
`role_change_label`	`additional.fields`	ממוזג
`sec_challenge_label`	`additional.fields`	ממוזג
`oauthaccesstokenname`	`extensions.auth.auth_details`	מיפוי ישיר
`has_principal`	`extensions.auth.type`	מופה: `true` ← `AUTHTYPE_UNSPECIFIED`
`column7`	`metadata.description`	מיפוי ישיר
`msg`	`metadata.description`	מיפוי ישיר
`Role_Change_Date`	`metadata.event_timestamp`	נותח בתור `M/d/yyyy h:mm a`
`date`	`metadata.event_timestamp`	נותח בתור `M/d/yyyy h:mm:ss a`
`result.Login_Audit_Trail_Date`	`metadata.event_timestamp`	נותח בתור `LDAP`
`ts`	`metadata.event_timestamp`	נותח בתור `MMM d HH:mm:ss`
`has_principal`	`metadata.event_type`	מופה: `true` ← `USER_LOGIN`, `true` ← `STATUS_UPDATE`
`has_principal_user`	`metadata.event_type`	מופה: `true` ← `USER_UNCATEGORIZED`
`column1`	`metadata.product_event_type`	מיפוי ישיר
`log_type`	`metadata.product_event_type`	מיפוי ישיר
`useragent`	`network.http.parsed_user_agent`	מיפוי ישיר
`Login_Audit_Trail_User_Agent`	`network.http.user_agent`	מיפוי ישיר
`result.Login_Audit_Trail_User_Agent`	`network.http.user_agent`	מיפוי ישיר
`useragent`	`network.http.user_agent`	מיפוי ישיר
`Login_Audit_Trail_Application`	`principal.application`	מיפוי ישיר
`column10`	`principal.application`	מיפוי ישיר
`oauthappname`	`principal.application`	מיפוי ישיר
`result.host`	`principal.asset.hostname`	מיפוי ישיר
`column3`	`principal.asset.ip`	ממוזג
`ipaddress`	`principal.asset.ip`	ממוזג
`result.Login_Audit_Trail_IP_Address`	`principal.asset.ip`	מיפוי ישיר
`src_ip`	`principal.asset.ip`	ממוזג
`result.host`	`principal.hostname`	מיפוי ישיר
`column3`	`principal.ip`	ממוזג
`ipaddress`	`principal.ip`	ממוזג
`result.Login_Audit_Trail_IP_Address`	`principal.ip`	מיפוי ישיר
`src_ip`	`principal.ip`	ממוזג
`Login_Audit_Trail_Role_label`	`principal.user.attribute.labels`	ממוזג
`login_audits_role_label`	`principal.user.attribute.labels`	ממוזג
`supervisor_label`	`principal.user.attribute.labels`	ממוזג
`role`	`principal.user.attribute.roles`	ממוזג
`Email`	`principal.user.email_addresses`	מופה: `^.+@.+$` ← `Email`
`column2`	`principal.user.email_addresses`	ממוזג
`emailaddress`	`principal.user.email_addresses`	מופה: `^.+@.+$` ← `emailaddress`
`result.Login_Audit_Trail_Email_Address`	`principal.user.email_addresses`	ממוזג
`Job_Title`	`principal.user.title`	מיפוי ישיר
`Name`	`principal.user.user_display_name`	מיפוי ישיר
`Login_Audit_Trail_User`	`principal.user.userid`	מיפוי ישיר
`result.Login_Audit_Trail_User`	`principal.user.userid`	מיפוי ישיר
`user`	`principal.user.userid`	מיפוי ישיר
`Login_Audit_Trail_Security_Challenge_label`	`security_result.about.resource.attribute.labels`	ממוזג
`security_result_action`	`security_result.action`	ממוזג
`Login_Audit_Trail_Status`	`security_result.summary`	מיפוי ישיר
`result.Login_Audit_Trail_Status`	`security_result.summary`	מיפוי ישיר
`api_type_label`	`target.resource.attribute.labels`	ממוזג
`Login_Audit_Trail_Request_URI`	`target.url`	מיפוי ישיר
`result.Login_Audit_Trail_Request_URI`	`target.url`	מיפוי ישיר
לא רלוונטי	`extensions.auth.type`	קבוע: `AUTHTYPE_UNSPECIFIED`
לא רלוונטי	`metadata.event_type`	קבוע: `USER_LOGIN`
לא רלוונטי	`network.http.parsed_user_agent`	קבוע: `parseduseragent`

יומן שינויים

הצגת יומן השינויים של המנתח הזה

הבעיה עדיין לא נפתרה? קבלת תשובות מחברי הקהילה וממומחי Google SecOps.