Oracle NetSuite - NetSuite Applications Suite のログを収集する

以下でサポートされています。

このドキュメントでは、Google Cloud Storage V2 を使用して Oracle NetSuite - NetSuite Applications Suite ログを Google Security Operations に取り込む方法について説明します。

Oracle NetSuite は、財務管理、顧客管理（CRM）、e コマース、広告在庫管理などの包括的なビジネス管理機能を提供するクラウドベースの ERP（エンタープライズ リソース プランニング）プラットフォームです。NetSuite は、ユーザー ログイン トレイル、レコードの変更を追跡するシステム ノート、セキュリティ モニタリングとコンプライアンスを目的として SuiteQL REST API 経由で取得できる API アクセスログなど、広範な監査ログを生成します。

始める前に

次の前提条件を満たしていることを確認します。

  • Google SecOps インスタンス
  • Cloud Storage、Cloud Run、Pub/Sub、Cloud Scheduler API が有効になっている GCP プロジェクト
  • GCS バケットを作成および管理する権限
  • GCS バケットの IAM ポリシーを管理する権限
  • Cloud Run サービス、Pub/Sub トピック、Cloud Scheduler ジョブを作成する権限
  • 管理者権限を持つ NetSuite アカウント
  • NetSuite REST ウェブサービスが有効になっている
  • NetSuite でインテグレーション レコードとアクセス トークンを作成する権限

NetSuite API アクセスを構成する

SuiteQL REST API 経由でログ収集を有効にするには、NetSuite で REST ウェブサービスを有効にし、統合レコードを作成して、トークンベース認証（TBA）の認証情報を生成する必要があります。

必要な NetSuite 機能を有効にする

  1. 管理者として NetSuite アカウントにログインします。
  2. [Setup] > [Company] > [Enable Features] に移動します。
  3. [SuiteCloud] タブをクリックします。
  4. [SuiteScript] セクションで、次のチェックボックスをオンにします。
    • クライアント SuiteScript
    • Server SuiteScript
  5. [SuiteTalk（ウェブサービス）] セクションで、次のチェックボックスをオンにします。
    • REST ウェブサービス
  6. [Manage Authentication] セクションで、[Token-Based Authentication] チェックボックスをオンにします。
  7. SuiteCloud 利用規約のページで [I Agree] をクリックします。
  8. [保存] をクリックします。

統合レコードを作成する

  1. NetSuite で、[Setup] > [Integration] > [Manage Integrations] > [New] に移動します。
  2. [名前] フィールドに「Google SecOps Integration」と入力します。
  3. [説明] フィールドに「Integration for Google Security Operations log collection」と入力します。
  4. [状態] フィールドで、デフォルトの [有効] オプションを選択したままにします。
  5. [Authentication] タブをクリックします。
  6. [Token-based Authentication] セクションで、[Token-based Authentication] チェックボックスをオンにします。
  7. [TBA: Authorization Flow] ボックスはオフのままにします。
  8. [TBA: ISSUETOKEN ENDPOINT] チェックボックスをオンにします。
  9. [保存] をクリックします。

  10. 確認ページで、コンシューマー キー/クライアント ID とコンシューマー シークレット/クライアント シークレットをコピーして保存します。

API アクセス用のカスタムロールを作成する

  1. [Setup] > [Users/Roles] > [Manage Roles] > [New] に移動します。
  2. [名前] フィールドに「Google SecOps API Role」と入力します。
  3. [ID] フィールドに「google_secops_api_role」と入力します。
  4. [権限] サブタブをクリックします。
  5. [設定] サブタブをクリックします。
  6. 完全アクセスレベルで次の権限を追加します。
    • アクセス トークンを使用してログインする
    • ユーザー アクセス トークン
    • REST ウェブサービス
  7. [レポート] サブタブをクリックします。
  8. 表示アクセスレベルで次の権限を追加します。
    • SuiteAnalytics ワークブック
  9. [リスト] サブタブをクリックします。
  10. 表示アクセスレベルで次の権限を追加します。
    • ログイン監査証跡
  11. [保存] をクリックします。

ユーザーにロールを割り当てる

  1. [リスト> 社員> 社員] に移動します。
  2. 統合に使用するユーザー アカウントを選択します（または、新しい専用の統合ユーザーを作成します）。
  3. [編集] をクリックします。
  4. [アクセス] タブをクリックします。
  5. [ロール] セクションで、[追加] をクリックします。
  6. プルダウン リストから [Google SecOps API ロール] を選択します。
  7. [保存] をクリックします。

アクセス トークンを作成する

  1. Google SecOps API ロールが割り当てられているユーザーとして NetSuite にログインします。
  2. [Setup] > [Users/Roles] > [Access Tokens] > [New] に移動します。
  3. [アクセス トークン] ページで、次の操作を行います。
    • [Application Name] プルダウン リストで、[Google SecOps Integration] を選択します。
    • [ユーザー] プルダウン リストで、Google SecOps API ロールが割り当てられているユーザー アカウントを選択します。
    • [ロール] プルダウン リストで、[Google SecOps API ロール] を選択します。
    • [Token Name] フィールドは自動的に入力されます。必要に応じて変更できます。
  4. [保存] をクリックします。

  5. 確認ページで、[トークン ID] と [トークン シークレット] をコピーして保存します。

NetSuite アカウント ID を確認する

  1. NetSuite で、[Setup] > [Company] > [Company Information] に移動します。
  2. [アカウント ID] フィールド（[タイムゾーン] 設定の下に表示されます）を探します。
  3. アカウント ID（サンドボックス アカウントの場合は 1234567 や 1234567_SB1 など）をコピーして保存します。

権限を確認する

アカウントに必要な権限があることを確認するには:

  1. Google SecOps API ロールが割り当てられているユーザーで NetSuite にログインします。
  2. [設定> ユーザー/ロール> ログイン監査証跡を表示] に移動します。
  3. [ログイン監査証跡] ページが表示された場合は、アカウントに必要な権限が付与されています。
  4. このオプションが表示されない場合は、NetSuite 管理者に連絡して、[Lists] サブタブで [Login Audit Trail] 権限を付与してもらってください。

テスト 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"}

リクエストが成功すると、LoginAudit レコードを含む JSON オブジェクトが返されます。

Google Cloud Storage バケットを作成する

  1. Google Cloud コンソールに移動します。
  2. プロジェクトを選択するか、新しいプロジェクトを作成します。
  3. ナビゲーション メニューで、[Cloud Storage > バケット] に移動します。
  4. [バケットを作成] をクリックします。

  5. 次の構成情報を提供してください。

    設定 値
    バケットに名前を付ける グローバルに一意の名前（netsuite-audit-logs など）を入力します。
    ロケーション タイプ ニーズに応じて選択します（リージョン、デュアルリージョン、マルチリージョン）。
    ロケーション 場所を選択します（例: us-central1）。
    ストレージ クラス Standard（頻繁にアクセスするログにおすすめ）
    アクセス制御 均一（推奨）
    保護ツール 省略可: オブジェクトのバージョニングまたは保持ポリシーを有効にする

  6. [作成] をクリックします。

Cloud Run functions のサービス アカウントを作成する

Cloud Run functions の関数には、GCS バケットへの書き込み権限と Pub/Sub による呼び出し権限を持つサービス アカウントが必要です。

サービス アカウントの作成

  1. GCP コンソールで、[IAM と管理>サービス アカウント] に移動します。
  2. [サービス アカウントを作成] をクリックします。
  3. 次の構成情報を指定します。
    • サービス アカウント名: 「netsuite-audit-collector-sa」と入力します。
    • サービス アカウントの説明: 「Service account for Cloud Run function to collect NetSuite audit logs」と入力します。
  4. [作成して続行] をクリックします。
  5. [このサービス アカウントにプロジェクトへのアクセスを許可する] セクションで、次のロールを追加します。
    1. [ロールを選択] をクリックします。
    2. [Storage オブジェクト管理者] を検索して選択します。
    3. [+ 別のロールを追加] をクリックします。
    4. [Cloud Run 起動元] を検索して選択します。
    5. [+ 別のロールを追加] をクリックします。
    6. [Cloud Functions 起動元] を検索して選択します。
  6. [続行] をクリックします。
  7. [完了] をクリックします。

これらのロールは、次の操作に必要です。

  • Storage オブジェクト管理者: ログを GCS バケットに書き込み、状態ファイルを管理する
  • Cloud Run 起動元: Pub/Sub が関数を呼び出すことを許可します。
  • Cloud Functions 起動元: 関数の呼び出しを許可します。

GCS バケットに対する IAM 権限を付与する

  1. [Cloud Storage] > [バケット] に移動します。
  2. バケット名（netsuite-audit-logs）をクリックします。
  3. [権限] タブに移動します。
  4. [アクセス権を付与] をクリックします。
  5. 次の構成情報を指定します。
    • プリンシパルを追加: サービス アカウントのメールアドレス（netsuite-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com）を入力します。
    • ロールを割り当てる: [Storage オブジェクト管理者] を選択します。
  6. [保存] をクリックします。

Pub/Sub トピックの作成

  1. GCP Console で、[Pub/Sub] > [トピック] に移動します。
  2. [トピックを作成] をクリックします。
  3. 次の構成情報を指定します。
    • トピック ID: 「netsuite-audit-trigger」と入力します。
    • その他の設定はデフォルトのままにします。
  4. [作成] をクリックします。

ログを収集する Cloud Run 関数を作成する

Cloud Run 関数は、Cloud Scheduler からの Pub/Sub メッセージによってトリガーされ、NetSuite SuiteQL REST API からログを取得して GCS に書き込みます。

  1. GCP Console で、[Cloud Run] に移動します。
  2. [サービスを作成] をクリックします。
  3. [関数] を選択します（インライン エディタを使用して関数を作成します）。

  4. [構成] セクションで、次の構成の詳細を指定します。

    設定 値
    サービス名 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）を確認します。

  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 1234567 NetSuite アカウント ID
    NS_CONSUMER_KEY your-consumer-key 統合レコードのコンシューマ キー
    NS_CONSUMER_SECRET your-consumer-secret 統合レコードのコンシューマー シークレット
    NS_TOKEN_ID your-token-id アクセス トークンのトークン ID
    NS_TOKEN_SECRET your-token-secret アクセス トークンのトークン シークレット
    MAX_RECORDS 10000 1 回の実行あたりの最大レコード数
    PAGE_SIZE 1000 SuiteQL ページあたりのレコード数
    LOOKBACK_HOURS 24 最初のルックバック期間

  10. [変数とシークレット] セクションで、[リクエスト] まで下にスクロールします。

    • リクエストのタイムアウト: 「600」秒（10 分）と入力します。

  11. [設定] タブに移動します。

    • [リソース] セクションで、次の操作を行います。
      • メモリ: 512 MiB 以上を選択します。
      • CPU: 1 を選択します。

  12. [リビジョン スケーリング] セクションで、次の操作を行います。

    • [インスタンスの最小数]: 「0」と入力します。
    • [インスタンスの最大数] に「100」と入力します。

  13. [作成] をクリックします。

  14. サービスが作成されるまで待ちます（1 ～ 2 分）。

  15. サービスを作成すると、インライン コードエディタが自動的に開きます。

関数コードを追加する

  1. [エントリ ポイント] フィールドに「main」と入力します。

  2. インライン コードエディタで、次の 2 つのファイルを作成します。

    • 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

  3. [デプロイ] をクリックして、関数を保存してデプロイします。

  4. デプロイが完了するまで待ちます（2 ～ 3 分）。

Cloud Scheduler ジョブの作成

Cloud Scheduler は、定期的に Pub/Sub トピックにメッセージをパブリッシュし、Cloud Run functions の関数をトリガーします。

  1. GCP Console で、[Cloud Scheduler] に移動します。
  2. [ジョブを作成] をクリックします。

  3. 次の構成情報を提供してください。

    設定 値
    名前 netsuite-audit-collector-hourly
    リージョン Cloud Run functions と同じリージョンを選択する
    周波数 0 * * * *（1 時間ごと、毎正時）
    タイムゾーン タイムゾーンを選択（UTC を推奨）
    ターゲット タイプ Pub/Sub
    トピック netsuite-audit-trigger を選択
    メッセージ本文 {}（空の JSON オブジェクト）

  4. [作成] をクリックします。

スケジュール頻度のオプション

ログの量とレイテンシの要件に基づいて頻度を選択します。

頻度 CRON 式 ユースケース
5 分おき */5 * * * * 大容量、低レイテンシ
15 分ごと */15 * * * * 検索量が普通
1 時間ごと 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
Page 1: Retrieved X records
Wrote X records to gs://netsuite-audit-logs/netsuite-audit/netsuite_audit_YYYYMMDD_HHMMSS.ndjson
Successfully processed X records

  8. [Cloud Storage] > [バケット] に移動します。

  9. netsuite-audit-logs をクリックします。

  10. netsuite-audit/ フォルダに移動します。

  11. 現在のタイムスタンプで新しい .ndjson ファイルが作成されたことを確認します。

ログにエラーが表示された場合:

  • HTTP 401: OAuth 認証情報（NS_CONSUMER_KEY、NS_CONSUMER_SECRET、NS_TOKEN_ID、NS_TOKEN_SECRET）と NS_ACCOUNT_ID 環境変数が正しいことを確認します。
  • HTTP 403: NetSuite ロールに必要な権限（REST ウェブサービス、ログイン監査証跡、SuiteAnalytics ワークブック）があることを確認します。
  • HTTP 429: レート制限 - 関数は指数バックオフで自動的に再試行されます
  • 環境変数がありません: 必要な変数がすべて Cloud Run functions の構成で設定されていることを確認します。

Google SecOps サービス アカウントを取得する

  1. [SIEM 設定] > [フィード] に移動します。
  2. [Add New Feed] をクリックします。
  3. [単一フィードを設定] をクリックします。
  4. [フィード名] フィールドに、フィードの名前を入力します（例: NetSuite Audit Logs）。
  5. [ソースタイプ] として [Google Cloud Storage V2] を選択します。
  6. [ログタイプ] として [Oracle NetSuite - NetSuite Applications Suite] を選択します。

  7. [サービス アカウントを取得する] をクリックします。一意のサービス アカウントのメールアドレスが表示されます。次に例を示します。

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

  8. このメールアドレスをコピーして、次のステップで使用します。

  9. [次へ] をクリックします。

  10. 次の入力パラメータの値を指定します。

    • ストレージ バケットの URL: 接頭辞パスを含む GCS バケット URI を入力します。

      gs://netsuite-audit-logs/netsuite-audit/
    • Source deletion option: 必要に応じて削除オプションを選択します。
      • なし: 転送後にファイルを削除しません（テストにおすすめ）。
      • 転送されたファイルを削除する: 転送が完了した後にファイルを削除します。
      • 転送されたファイルと空のディレクトリを削除する: 転送が成功した後にファイルと空のディレクトリを削除します。
    • ファイルの最大経過日数: 過去の日数以内に変更されたファイルを含めます（デフォルトは 180 日）。
    • Asset namespace: アセットの名前空間。
    • Ingestion labels: このフィードのイベントに適用されるラベル。

  11. [次へ] をクリックします。

  12. [Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。

Google SecOps サービス アカウントに IAM 権限を付与する

Google SecOps サービス アカウントには、GCS バケットに対する Storage オブジェクト閲覧者のロールが必要です。

  1. [Cloud Storage] > [バケット] に移動します。
  2. netsuite-audit-logs をクリックします。
  3. [権限] タブに移動します。
  4. [アクセス権を付与] をクリックします。
  5. 次の構成情報を指定します。
    • プリンシパルを追加: Google SecOps サービス アカウントのメールアドレスを貼り付けます。
    • ロールを割り当てる: [Storage オブジェクト閲覧者] を選択します。

  6. [保存] をクリックします。

NetSuite API のレート制限

NetSuite API には、次の同時実行数とレート制限があります。

制限のタイプ デフォルト値 メモ
同時リクエスト 15 件のリクエスト アカウントごとの SOAP API リクエストと REST API リクエストの合計に適用されます
SuiteQL クエリの結果 最大 100,000 行 SuiteQL クエリごとに返される最大行数
リクエストあたりのオブジェクト数 最大 1,000 個のオブジェクト API リクエストごとに返されるオブジェクトの最大数

Cloud Run functions は、指数バックオフを使用してレート制限を自動的に処理します。レート制限エラー（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 セキュリティ結果に関与するリソースの属性を記述する Key-Value ラベル
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 のプロフェッショナルから回答を得ることができます。