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

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

נתמך ב:

Google SecOps SIEM

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

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

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

ודאו שאתם עומדים בדרישות המוקדמות הבאות:

מכונה של Google SecOps
פרויקט GCP עם ממשקי Cloud Storage,‏ Cloud Run,‏ Pub/Sub ו-Cloud Scheduler API מופעלים
הרשאות ליצירה ולניהול של מאגרי GCS
הרשאות לניהול מדיניות IAM בקטגוריות של GCS
הרשאות ליצירת שירותי Cloud Run, נושאי Pub/Sub ומשימות Cloud Scheduler
חשבון NetSuite עם הרשאת אדמין
שירותי האינטרנט של NetSuite REST מופעלים
הרשאות ליצירת רשומות שילוב וטוקני גישה ב-NetSuite

הגדרת גישה ל-API של NetSuite

כדי להפעיל איסוף יומנים באמצעות SuiteQL API בארכיטקטורת REST, צריך להפעיל את שירותי האינטרנט של REST, ליצור רשומת שילוב וליצור פרטי כניסה לאימות מבוסס-טוקן (TBA) ב-NetSuite.

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

נכנסים לחשבון NetSuite כאדמין.
עוברים אל הגדרה > חברה > הפעלת תכונות.
לוחצים על הכרטיסייה SuiteCloud.
בקטע SuiteScript, מסמנים את התיבות הבאות:
- Client SuiteScript
- Server SuiteScript
בקטע SuiteTalk (Web Services) (שירותי אינטרנט), מסמנים את התיבה הבאה:
- שירותי אינטרנט של REST
בקטע Manage Authentication (ניהול אימות), מסמנים את התיבה Token-Based Authentication (אימות מבוסס-אסימון).
לוחצים על I Agree (אני מסכים/ה) בדף התנאים וההגבלות של SuiteCloud.
לוחצים על Save.

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

ב-NetSuite, עוברים אל Setup > Integration > Manage Integrations > New (הגדרה > שילוב > ניהול שילובים > חדש).
בשדה Name, מזינים Google SecOps Integration.
בשדה Description (תיאור) מזינים Integration for Google Security Operations log collection.
בשדה State, משאירים את האפשרות Enabled שנבחרה כברירת מחדל.
לוחצים על הכרטיסייה אימות.
בקטע Token-based Authentication (אימות מבוסס-טוקן), מסמנים את התיבה Token-based Authentication.
משאירים את התיבה TBA: Authorization Flow לא מסומנת.
מסמנים את התיבה TBA: ISSUETOKEN ENDPOINT.
לוחצים על Save.
בדף האישור, מעתיקים ושומרים את Consumer Key/Client ID (טוקן צרכן/מזהה לקוח) ואת Consumer Secret/Client Secret (סוד לשימוש עם טוקן צרכן/סוד לקוח).

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

יצירת תפקיד בהתאמה אישית לגישה ל-API

עוברים אל Setup > Users/Roles > Manage Roles > New.
בשדה Name, מזינים Google SecOps API Role.
בשדה ID, מזינים את הערך google_secops_api_role.
לוחצים על הכרטיסייה המשנית Permissions (הרשאות).
לוחצים על כרטיסיית המשנה הגדרה.
מוסיפים את ההרשאות הבאות עם רמת גישה מלאה:
- כניסה באמצעות טוקני גישה
- טוקני גישה של משתמשים
- REST Web Services
לוחצים על כרטיסיית המשנה דוחות.
מוסיפים את ההרשאות הבאות עם רמת גישה של צפייה:
- חוברת עבודה של SuiteAnalytics
לוחצים על כרטיסיית המשנה רשימות.
מוסיפים את ההרשאות הבאות עם רמת גישה של צפייה:
- נתיב ביקורת של התחברות
לוחצים על Save.

הקצאת התפקיד למשתמש

עוברים אל Lists > Employees > Employees.
בוחרים את חשבון המשתמש שישמש לשילוב (או יוצרים משתמש חדש שמוקדש לשילוב).
לוחצים על Edit.
לוחצים על הכרטיסייה גישה.
בקטע תפקידים, לוחצים על הוספה.
בוחרים באפשרות Google SecOps API Role מהרשימה הנפתחת.
לוחצים על Save.

יצירת אסימון גישה

נכנסים ל-NetSuite בתור המשתמש שהוקצה לו תפקיד Google SecOps API.
עוברים אל הגדרה > משתמשים/תפקידים > טוקנים של גישה > חדש.
בדף טוקן גישה:
- בתפריט הנפתח Application Name (שם האפליקציה), בוחרים באפשרות Google SecOps Integration (שילוב של Google SecOps).
- ברשימה הנפתחת User (משתמש), בוחרים את חשבון המשתמש שהוקצה לו תפקיד ב-Google SecOps API.
- ברשימה הנפתחת Role, בוחרים באפשרות Google SecOps API Role.
- השדה שם האסימון מתמלא אוטומטית. אפשר לשנות את ההגדרה הזו אם רוצים.
לוחצים על Save.
בדף האישור, מעתיקים ושומרים את מזהה האסימון ואת סוד האסימון.

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

איתור מזהה החשבון ב-NetSuite

ב-NetSuite, עוברים אל Setup > Company > Company Information (הגדרה > חברה > פרטי החברה).
מאתרים את השדה מזהה חשבון (מוצג מתחת להגדרה אזור זמן).
מעתיקים את מספר החשבון ושומרים אותו (לדוגמה, 1234567 או 1234567_SB1 לחשבונות בסביבת ארגז חול).

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

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

נכנסים ל-NetSuite עם המשתמש שהוקצה לו התפקיד Google SecOps API.
עוברים אל Setup > Users/Roles > View Login Audit Trail.
אם אתם יכולים לראות את הדף 'נתיב ביקורת של הכניסה', לחשבון יש את ההרשאות הנדרשות.
אם האפשרות הזו לא מופיעה, פנו לאדמין של NetSuite כדי לקבל את ההרשאה Login Audit Trail בכרטיסיית המשנה Lists.

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

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

# Replace with your actual values
ACCOUNT_ID="your-account-id"

# Replace underscores with hyphens in the account ID for the URL
# For example: 1234567_SB1 becomes 1234567-sb1
URL_ACCOUNT_ID=$(echo "$ACCOUNT_ID" | tr '_' '-' | tr '[:upper:]' '[:lower:]')

CONSUMER_KEY="your-consumer-key"
CONSUMER_SECRET="your-consumer-secret"
TOKEN_ID="your-token-id"
TOKEN_SECRET="your-token-secret"

# Test SuiteQL API access using OAuth 1.0 TBA
# Note: This example requires the oauth1 CLI tool or Postman.
# Use the following endpoint and headers to verify connectivity:
# POST https://${URL_ACCOUNT_ID}.suitetalk.api.netsuite.com/services/rest/query/v1/suiteql?limit=5
# Headers:
#   Authorization: OAuth (with consumer key, token, HMAC-SHA256 signature)
#   Content-Type: application/json
#   Prefer: transient
# Body: {"q": "SELECT id, date, status, user, emailaddress FROM LoginAudit ORDER BY date DESC"}

תגובה מוצלחת מחזירה אובייקט JSON שמכיל רשומות LoginAudit.

יצירת קטגוריה של 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 נדרש חשבון שירות עם הרשאות כתיבה בקטגוריית GCS, וצריך להפעיל אותה באמצעות Pub/Sub.

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

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

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

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

הענקת הרשאות IAM בקטגוריה של GCS

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

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

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

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

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

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

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

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

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

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

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

שם המשתנה	ערך לדוגמה	תיאור
`GCS_BUCKET`	`netsuite-audit-logs`	שם קטגוריית GCS
`GCS_PREFIX`	`netsuite-audit`	קידומת לקובצי יומן
`STATE_KEY`	`netsuite-audit/state.json`	נתיב קובץ תמונת המצב
`NS_ACCOUNT_ID`	`1234567`	מספר חשבון NetSuite
`NS_CONSUMER_KEY`	`your-consumer-key`	טוקן צרכן מרשומה של שילוב
`NS_CONSUMER_SECRET`	`your-consumer-secret`	סוד הצרכן מרשומת השילוב
`NS_TOKEN_ID`	`your-token-id`	מזהה הטוקן מטוקן הגישה
`NS_TOKEN_SECRET`	`your-token-secret`	סוד הטוקן מטוקן הגישה
`MAX_RECORDS`	`10000`	מספר הרשומות המקסימלי בכל הפעלה
`PAGE_SIZE`	`1000`	רשומות בכל דף של SuiteQL
`LOOKBACK_HOURS`	`24`	תקופת המבט לאחור הראשונית

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

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

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

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

main.py:

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

# Initialize HTTP client with timeouts
http = urllib3.PoolManager(
  timeout=urllib3.Timeout(connect=10.0, read=60.0),
  retries=False,
)

# Initialize Storage client
storage_client = storage.Client()

# Environment variables
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_CONSUMER_KEY = os.environ.get('NS_CONSUMER_KEY')
NS_CONSUMER_SECRET = os.environ.get('NS_CONSUMER_SECRET')
NS_TOKEN_ID = os.environ.get('NS_TOKEN_ID')
NS_TOKEN_SECRET = os.environ.get('NS_TOKEN_SECRET')
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000'))
PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '1000'))
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))

def get_netsuite_base_url(account_id):
  """Build the NetSuite REST API base URL from the account ID."""
  url_account = account_id.replace('_', '-').lower()
  return f"https://{url_account}.suitetalk.api.netsuite.com"

def generate_oauth_header(method, url, account_id, consumer_key,
  consumer_secret, token_id, token_secret):
  """Generate OAuth 1.0 Authorization header using HMAC-SHA256."""
  timestamp = str(int(time.time()))
  nonce = secrets.token_hex(16)

  # Parse URL to separate base URL and query parameters
  parsed = urllib.parse.urlparse(url)
  base_url = f"{parsed.scheme}://{parsed.netloc}{parsed.path}"
  query_params = urllib.parse.parse_qs(parsed.query, keep_blank_values=True)

  # Build parameter string (OAuth params + query params)
  params = {
    'oauth_consumer_key': consumer_key,
    'oauth_token': token_id,
    'oauth_nonce': nonce,
    'oauth_timestamp': timestamp,
    'oauth_signature_method': 'HMAC-SHA256',
    'oauth_version': '1.0',
  }
  for key, values in query_params.items():
    params[key] = values[0]

  # Sort parameters and create parameter string
  sorted_params = sorted(params.items())
  param_string = '&'.join(
    f"{urllib.parse.quote(k, safe='')}={urllib.parse.quote(v, safe='')}"
    for k, v in sorted_params
  )

  # Create signature base string
  signature_base = (
    f"{method.upper()}&"
    f"{urllib.parse.quote(base_url, safe='')}&"
    f"{urllib.parse.quote(param_string, safe='')}"
  )

  # Create signing key
  signing_key = (
    f"{urllib.parse.quote(consumer_secret, safe='')}&"
    f"{urllib.parse.quote(token_secret, safe='')}"
  )

  # Generate HMAC-SHA256 signature
  signature = base64.b64encode(
    hmac.new(
      signing_key.encode('utf-8'),
      signature_base.encode('utf-8'),
      hashlib.sha256
    ).digest()
  ).decode('utf-8')

  # Build realm from account ID
  realm = account_id.replace('-', '_').upper()

  # Build Authorization header
  auth_header = (
    f'OAuth realm="{realm}",'
    f'oauth_consumer_key="{consumer_key}",'
    f'oauth_token="{token_id}",'
    f'oauth_nonce="{nonce}",'
    f'oauth_timestamp="{timestamp}",'
    f'oauth_signature_method="HMAC-SHA256",'
    f'oauth_version="1.0",'
    f'oauth_signature="{urllib.parse.quote(signature, safe="")}"'
  )
  return auth_header

@functions_framework.cloud_event
def main(cloud_event):
  """
  Cloud Run function triggered by Pub/Sub to fetch NetSuite
  LoginAudit logs via SuiteQL and write to GCS.
  """
  if not all([GCS_BUCKET, NS_ACCOUNT_ID, NS_CONSUMER_KEY,
    NS_CONSUMER_SECRET, NS_TOKEN_ID, NS_TOKEN_SECRET]):
    print('Error: Missing required environment variables')
    return

  try:
    bucket = storage_client.bucket(GCS_BUCKET)
    state = load_state(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()}")

    records, newest_time = fetch_login_audit(last_time, now)

    if not records:
      print("No new log records found.")
      save_state(bucket, now.isoformat())
      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 records]
    ) + '\n'
    blob.upload_from_string(ndjson, content_type='application/x-ndjson')

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

    if newest_time:
      save_state(bucket, newest_time)
    else:
      save_state(bucket, now.isoformat())

    print(f"Successfully processed {len(records)} records")

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

def fetch_login_audit(start_time, end_time):
  """
  Fetch LoginAudit records from NetSuite SuiteQL API with pagination.

  Returns:
    Tuple of (records list, newest_event_time ISO string)
  """
  base_url = get_netsuite_base_url(NS_ACCOUNT_ID)
  suiteql_endpoint = f"{base_url}/services/rest/query/v1/suiteql"

  start_str = start_time.strftime('%Y-%m-%d %H:%M:%S')
  end_str = end_time.strftime('%Y-%m-%d %H:%M:%S')

  query = (
    "SELECT id, date, status, user, role, emailaddress, "
    "ipaddress, useragent, detail, requesturi, "
    "oauthaccesstokenname, oauthappname "
    "FROM LoginAudit "
    f"WHERE date BETWEEN TO_DATE('{start_str}', 'YYYY-MM-DD HH24:MI:SS') "
    f"AND TO_DATE('{end_str}', 'YYYY-MM-DD HH24:MI:SS') "
    "ORDER BY date ASC"
  )

  records = []
  newest_time = None
  offset = 0
  backoff = 1.0

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

    remaining = min(PAGE_SIZE, MAX_RECORDS - len(records))
    url = f"{suiteql_endpoint}?limit={remaining}&offset={offset}"

    auth_header = generate_oauth_header(
      'POST', url, NS_ACCOUNT_ID, NS_CONSUMER_KEY,
      NS_CONSUMER_SECRET, NS_TOKEN_ID, NS_TOKEN_SECRET
    )

    headers = {
      'Authorization': auth_header,
      'Content-Type': 'application/json',
      'Prefer': 'transient',
      'User-Agent': 'GoogleSecOps-NetSuiteCollector/1.0',
    }

    body = json.dumps({'q': query}).encode('utf-8')

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

      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"HTTP Error: {response.status}")
        response_text = response.data.decode('utf-8')
        print(f"Response body: {response_text}")
        return records, newest_time

      data = json.loads(response.data.decode('utf-8'))

      page_results = data.get('items', [])

      if not page_results:
        print("No more results (empty page)")
        break

      page_num = (offset // PAGE_SIZE) + 1
      print(f"Page {page_num}: Retrieved {len(page_results)} records")
      records.extend(page_results)

      for event in page_results:
        try:
          event_time = event.get('date')
          if event_time:
            if newest_time is None or event_time > newest_time:
              newest_time = event_time
        except Exception as e:
          print(f"Warning: Could not parse event time: {e}")

      has_more = data.get('hasMore', False)
      if not has_more:
        print("Reached last page (hasMore=false)")
        break

      offset += len(page_results)

    except Exception as e:
      print(f"Error fetching logs: {e}")
      return records, newest_time

  print(f"Retrieved {len(records)} total records")
  return records, newest_time

def load_state(bucket):
  """Load state from GCS."""
  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}")
  return {}

def save_state(bucket, last_event_time_iso):
  """Save the last event timestamp to GCS state file."""
  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}")

requirements.txt:

functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.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	תרחיש לדוגמה
כל 5 דקות	`/5 * * *`	נפח גבוה, זמן אחזור קצר
כל 15 דקות	`/15 * * *`	נפח חיפושים בינוני
כל שעה	`0 * * * *`	רגיל (מומלץ)
כל 6 שעות	`0 /6 * *`	נפח נמוך, עיבוד באצווה
יומי	`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
Page 1: Retrieved X records
Wrote X records to gs://netsuite-audit-logs/netsuite-audit/netsuite_audit_YYYYMMDD_HHMMSS.ndjson
Successfully processed X records

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

אם מופיעות שגיאות ביומנים:

‫HTTP 401: מוודאים שהפרטים של OAuth (NS_CONSUMER_KEY, ‏ NS_CONSUMER_SECRET, ‏ NS_TOKEN_ID, ‏ NS_TOKEN_SECRET) ומשתני הסביבה NS_ACCOUNT_ID נכונים
‫HTTP 403: מוודאים שלתפקיד ב-NetSuite יש את ההרשאות הנדרשות (REST Web Services, ‏ נתיב ביקורת כניסה, ‏ SuiteAnalytics Workbook)
HTTP 429: הגבלת קצב – הפונקציה תנסה שוב באופן אוטומטי עם השהיה מעריכית לפני ניסיון חוזר
משתני סביבה חסרים: מוודאים שכל המשתנים הנדרשים מוגדרים בהגדרות של פונקציית Cloud Run

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

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

הערה: לכל מכונה של Google SecOps יש חשבון שירות ייחודי. אל תשתמשו בחשבונות שירות ממסמכים או מדוגמאות אחרים.
לוחצים על הבא.
מציינים ערכים לפרמטרים הבאים של הקלט:
- כתובת ה-URL של קטגוריית האחסון: מזינים את ה-URI של קטגוריית GCS עם נתיב הקידומת:
```
gs://netsuite-audit-logs/netsuite-audit/
```
הערה: תמיד צריך לכלול את הלוכסן (/) בסוף ה-URI.
- אפשרות מחיקת המקור: בוחרים את אפשרות המחיקה הרצויה:
  - Never: אף פעם לא למחוק קבצים אחרי העברות (מומלץ לבדיקה).
  - מחיקת קבצים שהועברו: הקבצים נמחקים אחרי שההעברה מסתיימת.
  - מחיקת קבצים שהועברו וספריות ריקות: מחיקת קבצים וספריות ריקות אחרי שההעברה מסתיימת בהצלחה.
הערה: אם בוחרים באפשרות מחיקה, לחשבון השירות צריך להיות התפקיד Storage Object Admin ולא התפקיד Storage Object Viewer. מעדכנים את ההרשאות ב-IAM בהתאם.
- הגיל המקסימלי של הקובץ: כולל קבצים ששונו במספר הימים האחרון (ברירת המחדל היא 180 ימים)
- מרחב השמות של הנכס: מרחב השמות של הנכס
- תוויות הטמעה: התווית שתתווסף לאירועים מהפיד הזה
לוחצים על הבא.
בודקים את ההגדרות של הפיד החדש במסך סיום ולוחצים על שליחה.

הענקת הרשאות IAM לחשבון השירות של Google SecOps

חשבון השירות של Google SecOps צריך את התפקיד צפייה באובייקט אחסון בקטגוריית GCS.

נכנסים לדף Cloud Storage > Buckets.
לוחצים על netsuite-audit-logs.
עוברים לכרטיסייה Permissions.
לוחצים על הענקת גישה.
מזינים את פרטי ההגדרה הבאים:
- Add principals (הוספת ישויות מורשות): מדביקים את כתובת האימייל בחשבון של חשבון השירות של Google SecOps
- הקצאת תפקידים: בוחרים באפשרות צפייה באובייקטים באחסון.
לוחצים על Save.

הערה: אם אתם מתכננים להשתמש באפשרות המחיקה 'מחיקת קבצים שהועברו' או 'מחיקת קבצים שהועברו וספריות ריקות', צריך להעניק את התפקיד Storage Object Admin במקום Storage Object Viewer.

מגבלות קצב של NetSuite API

ל-NetSuite API יש את המגבלות הבאות על קצב יצירת הבקשות ועל מספר הבקשות בו-זמנית:

סוג המגבלה	ערך ברירת המחדל	הערות
בקשות מקבילות	‫15 בקשות	רלוונטי לבקשות משולבות של SOAP ו-REST API לכל חשבון
תוצאות של שאילתת SuiteQL	עד 100,000 שורות	מספר השורות המקסימלי שמוחזרות לכל שאילתת SuiteQL
אובייקטים לכל בקשה	עד 1,000 אובייקטים	המספר המקסימלי של אובייקטים שמוחזרים לכל בקשת API

הפונקציה של Cloud Run מטפלת אוטומטית בהגבלת קצב יצירת הבקשות באמצעות השהיה מעריכית לפני ניסיון חוזר (exponential backoff). אם נתקלתם בשגיאות חוזרות של הגבלת קצב של יצירת בקשות (HTTP 429), פנו לתמיכה של NetSuite כדי להגדיל את מכסת בקשות ה-API או לשנות את התדירות של Cloud Scheduler.

טבלת מיפוי UDM

שדה ביומן	מיפוי UDM	היגיון
משתמש	extensions.auth.type	סוג מנגנון האימות שבו נעשה שימוש
תאריך	metadata.event_timestamp	חותמת הזמן שבה האירוע התרחש
event_type	metadata.event_type	סוג האירוע (למשל, USER_LOGIN, ‏ NETWORK_CONNECTION)
security_result	security_result	פרטים על תוצאת האבטחה של האירוע
oauthaccesstokenname	security_result.about.resource.attribute.labels	תוויות של צמדי מפתח/ערך שמתארות מאפיינים של המשאב שקשור לתוצאת האבטחה
secchallenge	security_result.about.resource.attribute.labels
סטטוס	security_result.summary	תיאור סיכום של תוצאת האבטחה
oauthappname	target.application	שם אפליקציית היעד
useragent	target.application
ipaddress	target.ip	כתובת ה-IP של היעד
requesturi	target.url	כתובת ה-URL של משאב היעד
emailaddress	target.user.email_addresses	כתובות אימייל שמשויכות למשתמש היעד
תפקיד	target.user.role_name	שם התפקיד שהוקצה למשתמש היעד
משתמש	target.user.userid	מזהה ייחודי של משתמש היעד

יומן שינויים

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

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