收集 Oracle NetSuite 日志

支持的平台：

Google SecOps SIEM

本文档介绍了如何使用 Google Cloud Storage V2 将 Oracle NetSuite 日志注入到 Google Security Operations。

Oracle NetSuite 是一个基于云的企业资源规划 (ERP) 平台，提供全面的业务管理功能，包括财务管理、CRM、电子商务和产品目录管理系统。NetSuite 会针对用户登录、系统更改和 API 活动生成审核日志，这些日志可通过 SuiteQL REST API 进行查询，并可注入到 Google SecOps 中以进行安全监控和合规性检查。

准备工作

确保您满足以下前提条件：

• Google SecOps 实例
• 启用了 Cloud Storage、Cloud Run、Pub/Sub 和 Cloud Scheduler API 的 GCP 项目
• 创建和管理 GCS 存储分区的权限
• 管理 GCS 存储分区的 IAM 政策的权限
• 创建 Cloud Run 服务、Pub/Sub 主题和 Cloud Scheduler 作业的权限
• 对您的 Oracle NetSuite 账号的管理员访问权限
• 您的 NetSuite 账号中已启用 REST Web 服务和 OAuth 2.0 功能
• 您的 NetSuite 账号 ID（可在设置 > 公司 > 公司信息中找到）

启用必需的 NetSuite 功能

1. 以管理员身份登录您的 NetSuite 账号。
2. 依次前往设置 > 公司 > 启用功能。
3. 点击 SuiteCloud 子标签页。
4. 在 SuiteTalk（Web 服务）部分中，选中 REST Web 服务复选框。
5. 在管理身份验证部分中，选中 OAuth 2.0 框。
6. 点击保存。

配置 Oracle NetSuite OAuth 2.0 凭据

如需使 Cloud Run 函数能够从 NetSuite 检索审核日志，您需要创建一条采用 OAuth 2.0 客户端凭据身份验证的集成记录。

创建集成记录

1. 以管理员身份登录您的 NetSuite 账号。
2. 依次前往设置 > 集成 > 管理集成 > 新建。
3. 在名称字段中，输入 Google SecOps Integration。
4. 在说明字段中，输入 Integration with Google SecOps for audit log ingestion。
5. 在状态字段中，选择已启用。
6. 点击身份验证子标签页。
7. 在 OAuth 2.0 部分，配置以下内容：
   • 选择客户端凭据（机器对机器）授权。
   • 选择 REST Web 服务。
8. 清除基于令牌的身份验证部分中的所有复选框。
9. 点击保存。

记录 OAuth 2.0 凭证

保存集成记录后，NetSuite 会显示您的 OAuth 2.0 凭据：

• 客户端 ID：您的唯一客户端标识符

• 客户端密钥：您的 API 密钥

  注意： 请立即复制并保存客户端 ID 和客户端密钥。NetSuite 只会显示这些值一次。如果丢失这些凭据，您必须重置集成记录中的凭据。

为 OAuth 2.0 客户端凭据生成证书

OAuth 2.0 客户端凭据流需要用于 JWT 签名的证书。使用 OpenSSL 生成 ECDSA 证书：

1. 在安装了 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
   

2. 当系统提示时，输入证书主题字段（国家/地区、州/省/市、组织等）。对于通用名称，请输入 Google SecOps Integration。

3. 该命令会生成两个文件：

   • private.pem：私钥（请妥善保管；用于在 Cloud Run 函数中进行 JWT 签名）
   • public.pem：公共证书（上传到 NetSuite）
   注意： NetSuite 支持 ECDSA (ES256, ES512) 和 RSA-PSS (PS256) 证书。本指南使用 ECDSA 和 prime256v1 曲线 (ES256)。证书的有效期最长为 2 年。

创建 OAuth 2.0 客户端凭据映射

1. 在 NetSuite 中，依次前往设置 > 集成 > 管理身份验证 > OAuth 2.0 客户端凭据 (M2M) 设置。
2. 点击新建。
3. 在该对话框中，配置以下内容：
   • 实体：选择将与 API 请求关联的员工或用户。
   • 角色：选择具有使用 OAuth 2.0 访问令牌登录权限的角色（请参阅所需权限）。如果没有此权限，相应角色不会显示在此列表中。
   • 应用：选择 Google SecOps Integration。
4. 点击选择文件，然后上传 public.pem 证书文件。
5. 点击保存。
6. 保存后，请记下映射列表中显示的证书 ID 值。此值用作 JWT 标头中的 kid（密钥 ID）。

所需权限

分配给 OAuth 2.0 映射的角色必须具有以下权限：

权限	访问权限级别	用途
使用 OAuth 2.0 访问令牌登录	完整	允许角色使用 OAuth 2.0 令牌登录
REST Web 服务	完整	执行 REST API 请求
登录审核跟踪记录	完整	访问登录审核轨迹数据
SuiteAnalytics 工作簿	查看	使用 SuiteQL 访问分析数据

为集成创建自定义角色

1. 依次前往设置 > 用户/角色 > 管理角色 > 新建。
2. 在名称字段中，输入 Google SecOps Integration Role。
3. 点击权限子标签页。
4. 点击设置子子标签页。
5. 添加以下权限：
   • 使用 OAuth 2.0 访问令牌登录：设置为完整。
   • REST Web 服务：设置为完整。
   • 登录审核跟踪：设置为完整。
6. 点击报告子子标签页。
7. 添加以下权限：
   • SuiteAnalytics 工作簿：设置为查看。
8. 点击保存。

验证权限

如需验证角色是否具有所需的权限，请执行以下操作：

1. 登录 NetSuite。
2. 依次前往设置 > 用户/角色 > 管理角色。
3. 找到 Google SecOps Integration Role，然后点击修改。
4. 验证所需权限中列出的权限是否显示在权限子标签页中。

测试 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
  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
  

如果响应成功，则会返回一个 JSON 对象，其中包含一个 items 数组，该数组包含登录审核记录。

创建 Google Cloud Storage 存储桶

1. 前往 Google Cloud 控制台。
2. 选择您的项目或创建新项目。
3. 在导航菜单中，依次前往 Cloud Storage > 存储分区。
4. 点击创建存储分区。

5. 提供以下配置详细信息：

   设置	值
   为存储桶命名	输入一个全局唯一的名称（例如 netsuite-audit-logs）
   位置类型	根据您的需求进行选择（区域级、双区域、多区域）
   位置	选择营业地点（例如 us-central1）
   存储类别	标准（建议用于经常访问的日志）
   访问权限控制	均匀（推荐）
   保护工具	可选：启用对象版本控制或保留政策

6. 点击创建。

为 Cloud Run 函数创建服务账号

Cloud Run 函数需要一个服务账号，该账号具有写入 GCS 存储桶和被 Pub/Sub 调用的权限。

创建服务账号

1. 在 GCP 控制台中，依次前往 IAM 和管理 > 服务账号。
2. 点击创建服务账号。
3. 提供以下配置详细信息：
   • 服务账号名称：输入 netsuite-audit-collector-sa。
   • 服务账号说明：输入 Service account for Cloud Run function to collect Oracle NetSuite audit logs。
4. 点击创建并继续。
5. 在向此服务账号授予对项目的访问权限部分中，添加以下角色：
   1. 点击选择角色。
   2. 搜索并选择 Storage Object Admin。
   3. 点击 + 添加其他角色。
   4. 搜索并选择 Cloud Run Invoker。
   5. 点击 + 添加其他角色。
   6. 搜索并选择 Cloud Functions Invoker。
6. 点击继续。
7. 点击完成。

必须拥有这些角色，才能：

• Storage Object Admin：将日志写入 GCS 存储桶并管理状态文件。
• Cloud Run Invoker：允许 Pub/Sub 调用该函数。
• Cloud Functions Invoker：允许调用函数。

授予对 GCS 存储桶的 IAM 权限

1. 前往 Cloud Storage > 存储分区。
2. 点击您的存储桶名称 (netsuite-audit-logs)。
3. 前往权限标签页。
4. 点击授予访问权限。
5. 提供以下配置详细信息：
   • 添加主账号：输入服务账号电子邮件地址 (netsuite-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com)。
   • 分配角色：选择 Storage Object Admin。
6. 点击保存。

创建 Pub/Sub 主题

创建一个 Pub/Sub 主题，Cloud Scheduler 将向该主题发布消息，而 Cloud Run 函数将订阅该主题。

1. 在 GCP 控制台中，前往 Pub/Sub > 主题。
2. 点击创建主题。
3. 提供以下配置详细信息：
   • 主题 ID：输入 netsuite-audit-trigger。
   • 将其他设置保留为默认值。
4. 点击创建。

创建 Cloud Run 函数以收集日志

Cloud Run 函数将由来自 Cloud Scheduler 的 Pub/Sub 消息触发，以从 Oracle NetSuite SuiteQL REST API 中提取审核日志并将其写入 GCS。

1. 在 GCP 控制台中，前往 Cloud Run。
2. 点击创建服务。
3. 选择函数（使用内嵌编辑器创建函数）。

4. 在配置部分中，提供以下配置详细信息：

   设置	值
   Service 名称	netsuite-audit-collector
   区域	选择与您的 GCS 存储桶匹配的区域（例如 us-central1）
   运行时	选择 Python 3.12 或更高版本

5. 在触发器（可选）部分中：

   1. 点击 + 添加触发器。
   2. 选择 Cloud Pub/Sub。
   3. 在选择 Cloud Pub/Sub 主题中，选择 netsuite-audit-trigger。
   4. 点击保存。

6. 在身份验证部分中：

   1. 选择需要进行身份验证。
   2. 选择 Identity and Access Management (IAM)。
   注意： 调用函数时，Pub/Sub 会自动处理身份验证。

7. 向下滚动并展开容器、网络、安全性。

8. 前往安全性标签页：

   • 服务账号：选择 netsuite-audit-collector-sa。

9. 前往容器标签页：

   1. 点击变量和密钥。
   2. 为每个环境变量点击+ 添加变量：

   变量名称	示例值	说明
   GCS_BUCKET	netsuite-audit-logs	GCS 存储桶名称
   GCS_PREFIX	netsuite-audit	日志文件的前缀
   STATE_KEY	netsuite-audit/state.json	状态文件路径
   NS_ACCOUNT_ID	123456	NetSuite 账号 ID
   NS_CLIENT_ID	your-client-id	OAuth 2.0 客户端 ID
   NS_CERTIFICATE_ID	your-certificate-id	来自 M2M 映射的证书 ID (kid)
   NS_PRIVATE_KEY	-----BEGIN EC PRIVATE KEY-----...	ECDSA 私钥内容
   LOOKBACK_HOURS	24	初始回溯期
   PAGE_SIZE	1000	每个 SuiteQL 页面的记录数（最多 1000）
   MAX_RECORDS	50000	每次运行的记录数上限

10. 在变量和 Secret 部分中，向下滚动到请求：

    • 请求超时：输入 600 秒（10 分钟）

11. 前往设置标签页：

    • 在资源部分中：
      • 内存：选择 512 MiB 或更高值。
      • CPU：选择 1。

12. 在修订版本伸缩部分中：

    • 最小实例数：输入 0。
    • 实例数上限：输入 100。

13. 点击创建。

14. 等待服务创建完成（1-2 分钟）。

15. 创建服务后，系统会自动打开内嵌代码编辑器。

添加函数代码

1. 在入口点字段中输入 main。

2. 在内嵌代码编辑器中，创建两个文件：

   • main.py:

     import functions_framework
     from google.cloud import storage
     import json
     import os
     import urllib3
     import urllib.parse
     import jwt
     import time
     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'))
     
     @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)
             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
     
             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 all_records]
             ) + '\n'
             blob.upload_from_string(ndjson, content_type='application/x-ndjson')
     
             print(f"Wrote {len(all_records)} records to gs://{GCS_BUCKET}/{object_key}")
     
             newest = find_newest_time(login_audits, system_notes)
             save_state(bucket, newest if newest else now.isoformat())
     
             print(f"Successfully processed {len(all_records)} 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}")
         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}")
     

   • requirements.txt:

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

3. 点击部署以保存并部署该函数。

4. 等待部署完成（2-3 分钟）。

   注意： Pub/Sub 触发器配置会自动创建必要的订阅和权限。

创建 Cloud Scheduler 作业

Cloud Scheduler 会定期向 Pub/Sub 主题发布消息，从而触发 Cloud Run 函数。

1. 在 GCP Console 中，前往 Cloud Scheduler。
2. 点击创建作业。

3. 提供以下配置详细信息：

   设置	值
   名称	netsuite-audit-collector-hourly
   区域	选择与 Cloud Run 函数相同的区域
   频率	0 * * * *（每小时一次，整点时）
   时区	选择时区（建议选择世界协调时间 [UTC]）
   目标类型	Pub/Sub
   主题	选择netsuite-audit-trigger
   消息正文	{}（空 JSON 对象）

4. 点击创建。

时间表频率选项

根据日志量和延迟时间要求选择频次：

频率	Cron 表达式	使用场景
每 15 分钟	*/15 * * * *	大批量、低延迟
每小时	0 * * * *	标准（推荐）
每 6 小时	0 */6 * * *	低成交量、批处理
每天	0 0 * * *	历史数据收集

测试集成

1. 在 Cloud Scheduler 控制台中，找到您的作业 (netsuite-audit-collector-hourly)。
2. 点击强制运行以手动触发作业。
3. 等待几秒钟。
4. 前往 Cloud Run > 服务。
5. 点击 netsuite-audit-collector。
6. 点击日志标签页。

7. 验证函数是否已成功执行。请查找以下内容：

   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. 前往 Cloud Storage > 存储分区。

9. 点击 netsuite-audit-logs。

10. 转到 netsuite-audit/ 文件夹。

11. 验证是否已创建具有当前时间戳的新 .ndjson 文件。

如果您在日志中看到错误，请执行以下操作：

• HTTP 401：验证 NS_CLIENT_ID、NS_CERTIFICATE_ID 和 NS_PRIVATE_KEY 环境变量是否正确。
• HTTP 403：验证 NetSuite 角色是否具有使用 OAuth 2.0 访问令牌登录（完全）、REST Web 服务（完全）和登录审核轨迹（完全）权限。
• HTTP 429：速率限制 - 函数将自动重试，并采用指数退避算法。
• 缺少环境变量：验证是否已在 Cloud Run function 配置中设置所有必需的变量。

检索 Google SecOps 服务账号

1. 依次前往 SIEM 设置 > Feed。
2. 点击添加新 Feed。
3. 点击配置单个 Feed。
4. 在 Feed 名称字段中，输入 Feed 的名称（例如 NetSuite Audit Logs）。
5. 选择 Google Cloud Storage V2 作为来源类型。
6. 选择 Oracle NetSuite 作为日志类型。

7. 点击获取服务账号。系统会显示一个唯一的服务账号电子邮件地址，例如：

   [e-mail]
   

8. 复制此电子邮件地址，以便在下一步中使用。

   注意： 每个 Google SecOps 实例都有一个唯一的服务账号。请勿使用其他文档或示例中的服务账号。

9. 点击下一步。

10. 为以下输入参数指定值：

    • 存储桶网址：输入带有前缀路径的 GCS 存储桶 URI：

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

    注意： 请务必在 URI 末尾添加尾随斜杠 (/)。

    • 来源删除选项：根据您的偏好选择删除选项：

      • 永不：转移后永不删除任何文件（建议用于测试）。
      • 删除已转移的文件：在成功转移后删除文件。

      • 删除已转移的文件和空目录：在成功转移后删除文件和空目录。

        注意 ：如果您选择删除选项，服务账号必须具有 Storage Object Admin 角色，而不是 Storage Object Viewer 角色。相应地更新 IAM 权限。

    • 文件存在时间上限：包含在过去指定天数内修改的文件（默认值为 180 天）。

    • 资源命名空间：资源命名空间。

    • 注入标签：要应用于此 Feed 中事件的标签。

11. 点击下一步。

12. 在最终确定界面中查看新的 Feed 配置，然后点击提交。

向 Google SecOps 服务账号授予 IAM 权限

Google SecOps 服务账号需要对您的 GCS 存储桶具有 Storage Object Viewer 角色。

1. 前往 Cloud Storage > 存储分区。
2. 点击 netsuite-audit-logs。
3. 前往权限标签页。
4. 点击授予访问权限。
5. 提供以下配置详细信息：
   • 添加主账号：粘贴 Google SecOps 服务账号电子邮件地址。
   • 分配角色：选择 Storage Object Viewer。

6. 点击保存。

   注意： 如果您打算使用“删除已转移的文件”或“删除已转移的文件和空目录”删除选项，请授予存储对象管理员角色，而不是存储对象查看者角色。

NetSuite API 速率限制

NetSuite REST API 具有以下监管限制：

上限类型	值
并发限制	每个集成的并发请求数上限为 10
每页的 SuiteQL 查询结果数	最多 1,000 行
SuiteQL 总结果数	每个查询 10 万行

Cloud Run 函数会自动处理指数退避算法速率限制。

UDM 映射表

日志字段	UDM 映射	逻辑
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 专业人士的解答。