收集 Oracle NetSuite 日志

支持：

本文档介绍了如何使用 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 的 Google Cloud 项目
  • 创建和管理 Cloud Storage 存储桶的权限
  • 管理 Cloud Storage 存储桶的 Identity and Access Management (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 密钥

为 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）

创建 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 访问分析数据
Google Analytics 和 REST 的系统注释 查看 通过 SuiteQL 查询 SystemNote 记录

为集成创建自定义角色

  1. 依次前往设置 > 用户/角色 > 管理角色 > 新建。
  2. 在名称字段中，输入 Google SecOps Integration Role。
  3. 点击权限子标签页。
  4. 点击设置子子标签页。
  5. 添加以下权限：
    • 使用 OAuth 2.0 访问令牌登录：设置为 Full
    • REST Web Services（REST Web 服务）：设置为 Full（完整）
    • 登录审核跟踪记录：设置为完整
  6. 点击报告子子标签页。
  7. 添加以下权限：
    • SuiteAnalytics 工作簿：设置为“查看”
  8. 点击列表子子标签页。
  9. 添加以下权限：
    • System Notes for Analytics and REST（Analytics 和 REST 的系统备注）：设置为查看
  10. 点击保存。

验证权限

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

  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 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 中的集成用户，则表示相应角色缺少 Google Analytics 和 REST 的系统备注权限。

创建 Google Cloud Storage 存储桶

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

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

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

  6. 点击创建。

为 Cloud Run 函数创建服务账号

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

创建服务账号

  1. 在 Google Cloud 控制台中，依次前往 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：将日志写入 Cloud Storage 存储桶并管理状态文件
  • Cloud Run Invoker：允许 Pub/Sub 调用函数
  • Cloud Functions Invoker：允许调用函数

授予对 Cloud Storage 存储桶的 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 中提取审核日志并将其写入 Cloud Storage。

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

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

    设置 值
    Service 名称 netsuite-audit-collector
    区域 选择与您的 Cloud Storage 存储桶匹配的区域（例如 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)。

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

  8. 前往安全标签页：

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

  9. 前往容器标签页：

    1. 点击变量和密钥。
    2. 为每个环境变量点击 + 添加变量：
    变量名称 示例值 说明
    GCS_BUCKET netsuite-audit-logs Cloud Storage 存储桶名称
    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. 在变量和密钥部分中，向下滚动到请求：

    • 请求超时：输入 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
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

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

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

创建 Cloud Scheduler 作业

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

  1. 在 GCP 控制台中，前往 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 服务（完全）和登录审核轨迹（完全）。
  • 输出中没有 system_note 记录：相应角色缺少“System Notes for Analytics and REST (View)”权限。SuiteQL 查询仍会成功，并且仅返回集成用户自己所做更改的备注，因此日志中不会出现错误。
  • 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. 点击获取服务账号。

  8. 系统会显示一个唯一的服务账号电子邮件地址，例如：

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

  9. 复制此邮箱，以便在下一步中使用。

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

Google SecOps 服务账号需要对您的 Cloud Storage 存储桶拥有两个角色：Storage Object Viewer（用于读取日志对象）和存储桶级角色（用于读取存储桶元数据）。

  1. 前往 Cloud Storage > 存储桶。
  2. 点击 netsuite-audit-logs。
  3. 前往权限标签页。
  4. 点击授予访问权限。
  5. 提供以下配置详细信息：
    • 添加主账号：粘贴 Google SecOps 服务账号电子邮件地址。
    • 分配角色：选择以下两项：
      • Storage Object Viewer：读取日志对象。
      • Storage Legacy Bucket Reader：读取存储桶元数据。如果您选择了删除已转移的文件和空目录删除选项，请改为选择存储旧版存储桶写入者，该角色也会授予删除权限。

  6. 点击保存。

完成 Google SecOps Feed 配置

  1. 返回到 Google SecOps Feed 创建向导。
  2. 点击下一步。

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

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

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

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

      • 永不删除文件：永不从源中删除文件（建议用于测试）。
      • 删除已转移的文件和空目录：在成功完成提取后，从来源位置删除文件和空目录。

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

    • 资产命名空间：资产命名空间

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

  4. 点击下一步。

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

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 专业人士的解答。