‫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 שאפשר לאחזר באמצעות SuiteQL API בארכיטקטורת REST למטרות ניטור לצורכי אבטחה ותאימות.

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

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

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

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

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

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

נכנסים לחשבון NetSuite כאדמין.
עוברים אל Setup > Company > Enable Features (הגדרה > חברה > הפעלת תכונות).
לוחצים על הכרטיסייה 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 שנבחרה כברירת מחדל.
לוחצים על הכרטיסייה אימות.
בקטע אימות מבוסס-אסימון, מסמנים את התיבה אימות מבוסס-אסימון.
משאירים את התיבה 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
לוחצים על כרטיסיית המשנה דוחות.
מוסיפים את ההרשאות הבאות עם רמת הגישה צפייה:
- SuiteAnalytics Workbook
לוחצים על כרטיסיית המשנה רשימות.
מוסיפים את ההרשאות הבאות עם רמת הגישה צפייה:
- נתיב ביקורת של התחברות
לוחצים על Save.

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

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

יצירת טוקן גישה

נכנסים ל-NetSuite בתור המשתמש שהוקצה לתפקיד Google SecOps API.
עוברים אל Setup (הגדרה) > Users/Roles (משתמשים/תפקידים) > Access Tokens (אסימוני גישה) > New (חדש).
בדף טוקן גישה:
- בתפריט הנפתח Application Name, בוחרים באפשרות Google SecOps Integration.
- בתפריט הנפתח 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.
לוחצים על Create Service Account.
מזינים את פרטי ההגדרה הבאים:
- Service account name: מזינים 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 ובוחרים בו.
5. לוחצים על + הוספת תפקיד.
6. מחפשים את התפקיד Cloud Functions Invoker ובוחרים אותו.
לוחצים על Continue.
לוחצים על סיום.

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

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

הענקת הרשאות 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.
לוחצים על Create topic.
מזינים את פרטי ההגדרה הבאים:
- מזהה הנושא: מזינים netsuite-audit-trigger
- משאירים את שאר ההגדרות כברירת מחדל.
לוחצים על יצירה.

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

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

במסוף GCP, נכנסים ל-Cloud Run.
לוחצים על Create service.
בוחרים באפשרות 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.
בקטע אימות:
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:
- ‫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 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 * *`	נפח נמוך, עיבוד ברצף (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
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: הגבלת קצב של יצירת בקשות – הפונקציה תנסה שוב באופן אוטומטי עם השהיה מעריכית לפני ניסיון חוזר (exponential backoff)
משתני סביבה חסרים: מוודאים שכל המשתנים הנדרשים מוגדרים בפונקציית Cloud Run.

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

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

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

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

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

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

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

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

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

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

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

טבלת מיפוי UDM

שדה ביומן	מיפוי UDM	Logic (פונקציה לוגית)
משתמש	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.