收集 Netskope Alerts 記錄

支援的國家/地區：

本文說明如何使用 Google Cloud Storage V2，將 Netskope Alerts 記錄擷取至 Google Security Operations。Netskope 是一個雲端安全平台，可為雲端服務、網站和私人應用程式提供即時資料和威脅防護。Netskope 快訊會擷取安全性事件，包括 SaaS、IaaS 和網路流量的資料遺失防護違規、惡意軟體偵測、異常行為、遭盜用的憑證和政策違規。

詳情請參閱「收集 Netskope 快訊記錄」。

事前準備

請確認您已完成下列事前準備事項：

  • Google SecOps 執行個體。
  • 已啟用 Cloud Storage API 的 GCP 專案。
  • 具備建立及管理 GCS 值區和 IAM 政策的權限。
  • 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限。
  • Netskope 租戶管理控制台的特殊存取權。
  • 具備快訊讀取權限的 Netskope REST API v2 權杖。

收集 Netskope API v2 憑證

建立 REST API 第 2 版權杖

  1. 登入 Netskope 租戶管理控制台 (例如 https://your-tenant.goskope.com)。
  2. 依序點選「設定」>「工具」>「REST API v2」。
  3. 按一下「New Token」。
  4. 請提供下列設定詳細資料：
    • 權杖名稱：輸入描述性名稱 (例如 SecOps-Alert-Integration)。
    • 到期：根據安全性政策設定權杖到期日。
  5. 在「端點」下方，啟用下列項目：
    • /api/v2/events/data/alert - 讀取權限。
  6. 按一下 [儲存]。
  7. 複製並妥善儲存權杖值。權杖只會顯示一次。

驗證 API 存取權

  • 請先測試憑證，再繼續進行整合：

    NETSKOPE_TENANT="[https://your-tenant.goskope.com](https://your-tenant.goskope.com)"
API_TOKEN="your-api-v2-token"

curl -v -H "Netskope-Api-Token: ${API_TOKEN}" \
    "${NETSKOPE_TENANT}/api/v2/events/data/alert?limit=1"

建立 Google Cloud Storage bucket

  1. 前往 Google Cloud 控制台。
  2. 選取專案或建立新專案。
  3. 在導覽選單中，依序前往「Cloud Storage」>「bucket」。
  4. 按一下「建立值區」。

  5. 請提供下列設定詳細資料：

    設定 值
    為 bucket 命名 輸入全域不重複的名稱 (例如 netskope-alerts-logs)
    位置類型 根據需求選擇 (區域、雙區域、多區域)
    位置 選取最接近 Google SecOps 執行個體的位置
    儲存空間級別 標準 (建議用於經常存取的記錄)
    存取控管 統一 (建議)

  6. 點選「建立」。

為 Cloud Run 函式建立服務帳戶

Cloud Run 函式需要具備 GCS bucket 寫入權限，且可由 Pub/Sub 叫用的服務帳戶。

建立服務帳戶

  1. 在 GCP 主控台中，依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
  2. 按一下「Create Service Account」(建立服務帳戶)。
  3. 請提供下列設定詳細資料：
    • 服務帳戶名稱：netskope-alerts-collector-sa。
    • 服務帳戶說明：Service account for Cloud Run function to collect Netskope Alerts logs。
  4. 按一下「建立並繼續」。
  5. 在「將專案存取權授予這個服務帳戶」部分，新增下列角色：
    1. Storage 物件管理員
    2. Cloud Run 叫用者
    3. Cloud Functions 叫用者
  6. 依序點按「繼續」和「完成」。

授予 GCS 值區的 IAM 權限

  1. 依序前往「Cloud Storage」>「Buckets」。
  2. 點選 netskope-alerts-logs 值區。
  3. 前往「權限」分頁標籤。
  4. 按一下「授予存取權」。
  5. 請提供下列設定詳細資料：
    • 新增主體：輸入服務帳戶電子郵件地址 (例如 netskope-alerts-collector-sa@PROJECT_ID.iam.gserviceaccount.com)。
    • 指派角色：選取「Storage 物件管理員」。
  6. 按一下 [儲存]。

建立 Pub/Sub 主題

建立 Pub/Sub 主題，Cloud Scheduler 會將訊息發布至該主題，而 Cloud Run 函式會訂閱該主題。

  1. 在 GCP 控制台中，前往「Pub/Sub」>「主題」。
  2. 按一下「建立主題」。
  3. 請提供下列設定詳細資料：
    • 主題 ID：netskope-alerts-trigger。
    • 其他設定保留預設值。
  4. 點選「建立」。

建立 Cloud Run 函式來收集記錄

Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發，從 Netskope REST API 第 2 版擷取快訊，並寫入 GCS。

  1. 前往 GCP Console 的「Cloud Run」。
  2. 按一下「Create service」(建立服務)。
  3. 選取「函式」。

  4. 在「設定」部分，提供下列設定詳細資料：

    設定 值
    服務名稱 netskope-alerts-collector
    區域 選取與 GCS bucket 相符的區域 (例如 us-central1)
    執行階段 Python 3.12 以上版本

  5. 在「Trigger」(觸發條件) 專區：

    1. 按一下「+ 新增觸發條件」。
    2. 選取「Cloud Pub/Sub」。
    3. 在「Select a Cloud Pub/Sub topic」(選取 Cloud Pub/Sub 主題) 中，選擇 netskope-alerts-trigger。
    4. 按一下 [儲存]。

  6. 在「驗證」部分中，選取「需要驗證」，然後勾選「身分與存取權管理 (IAM)」。

  7. 向下捲動至「Containers, Networking, Security」。

  8. 在「安全性」分頁中，選取服務帳戶 netskope-alerts-collector-sa。

  9. 在「Containers」分頁中，按一下「Variables & Secrets」，然後新增下列項目：

    變數名稱 範例值 說明
    GCS_BUCKET netskope-alerts-logs GCS bucket 名稱
    GCS_PREFIX netskope/alerts 記錄檔的前置字串
    STATE_KEY netskope/alerts/state.json 狀態檔案路徑
    NETSKOPE_TENANT https://your-tenant.goskope.com Netskope 租戶網址
    NETSKOPE_API_TOKEN your-api-v2-token REST API v2 權杖
    MAX_RECORDS 5000 每次執行的記錄數上限
    PAGE_SIZE 1000 每頁記錄數
    LOOKBACK_HOURS 24 初始回溯期

  10. 在「Requests」(要求) 部分，將「Request timeout」(要求逾時) 設為 600 秒。

  11. 在「設定」分頁 (「資源」)，選取 512 MiB 記憶體和 1 個 CPU。

  12. 點按「Create」(建立)。服務建立完成後，系統會自動開啟內嵌程式碼編輯器。

新增函式程式碼

  1. 在「進入點」欄位中輸入「main」。

  2. 在內嵌程式碼編輯器中，建立兩個檔案：

    • main.py:
    import functions_framework
from google.cloud import storage
import json
import os
import urllib3
from datetime import datetime, timezone, timedelta
import time

http = urllib3.PoolManager(
    timeout=urllib3.Timeout(connect=5.0, read=30.0),
    retries=False,
)

storage_client = storage.Client()

GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'netskope/alerts')
STATE_KEY = os.environ.get('STATE_KEY', 'netskope/alerts/state.json')
NETSKOPE_TENANT = os.environ.get('NETSKOPE_TENANT')
NETSKOPE_API_TOKEN = os.environ.get('NETSKOPE_API_TOKEN')
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '5000'))
PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '1000'))
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))

def to_epoch_seconds(dt):
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return int(dt.timestamp())

def parse_datetime(value):
    if isinstance(value, (int, float)):
        return datetime.fromtimestamp(value, tz=timezone.utc)
    if isinstance(value, str):
        if value.endswith("Z"):
            value = value[:-1] + "+00:00"
        return datetime.fromisoformat(value)
    return None

@functions_framework.cloud_event
def main(cloud_event):
    if not all([GCS_BUCKET, NETSKOPE_TENANT, NETSKOPE_API_TOKEN]):
        print('Error: Missing required environment variables')
        return

    try:
        bucket = storage_client.bucket(GCS_BUCKET)
        state = load_state(bucket, STATE_KEY)

        now = datetime.now(timezone.utc)
        last_time = None

        if isinstance(state, dict) and state.get("last_event_time"):
            try:
                last_time = parse_datetime(state["last_event_time"])
                last_time = last_time - timedelta(minutes=2)
            except Exception as e:
                print(f"Warning: Could not parse last_event_time: {e}")

        if last_time is None:
            last_time = now - timedelta(hours=LOOKBACK_HOURS)

        print(f"Fetching alerts from {last_time.isoformat()} to {now.isoformat()}")

        records, newest_event_time = fetch_alerts(
            tenant=NETSKOPE_TENANT,
            token=NETSKOPE_API_TOKEN,
            start_time=last_time,
            end_time=now,
            page_size=PAGE_SIZE,
            max_records=MAX_RECORDS,
        )

        if not records:
            print("No new alert records found.")
            save_state(bucket, STATE_KEY, to_epoch_seconds(now))
            return

        timestamp = now.strftime('%Y%m%d_%H%M%S')
        object_key = f"{GCS_PREFIX}/alerts_{timestamp}.ndjson"
        blob = bucket.blob(object_key)

        ndjson = '\n'.join([json.dumps(r, ensure_ascii=False) 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_event_time:
            save_state(bucket, STATE_KEY, newest_event_time)
        else:
            save_state(bucket, STATE_KEY, to_epoch_seconds(now))

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

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

def load_state(bucket, key):
    try:
        blob = bucket.blob(key)
        if blob.exists():
            state_data = blob.download_as_text()
            return json.loads(state_data)
    except Exception as e:
        print(f"Warning: Could not load state: {e}")
    return {}

def save_state(bucket, key, last_event_time):
    try:
        state = {'last_event_time': last_event_time}
        blob = bucket.blob(key)
        blob.upload_from_string(
            json.dumps(state, indent=2),
            content_type='application/json'
        )
        print(f"Saved state: last_event_time={last_event_time}")
    except Exception as e:
        print(f"Warning: Could not save state: {e}")

def fetch_alerts(tenant, token, start_time, end_time, page_size, max_records):
    base_url = tenant.rstrip('/')
    endpoint = f"{base_url}/api/v2/events/data/alert"

    headers = {
        'Netskope-Api-Token': token,
        'Accept': 'application/json',
        'User-Agent': 'GoogleSecOps-NetskopeAlertCollector/1.0'
    }

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

    start_epoch = to_epoch_seconds(start_time)
    end_epoch = to_epoch_seconds(end_time)
    operation_id = None

    url = f"{endpoint}?limit={page_size}&starttime={start_epoch}&endtime={end_epoch}"

    while True:
        page_num += 1

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

        if operation_id:
            url = f"{endpoint}?limit={page_size}&operation={operation_id}"

        try:
            response = http.request('GET', url, 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, 30.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 [], None

            data = json.loads(response.data.decode('utf-8'))
            page_results = data.get('result', [])

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

            print(f"Page {page_num}: Retrieved {len(page_results)} alerts")
            records.extend(page_results)

            for event in page_results:
                try:
                    event_time = event.get('timestamp')
                    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}")

            wait_time = data.get('wait_time')
            operation_id = data.get('operation')

            if not operation_id or len(page_results) < page_size:
                break

            if wait_time and wait_time > 0:
                time.sleep(wait_time)

        except Exception as e:
            print(f"Error fetching alerts: {e}")
            return [], None

    print(f"Retrieved {len(records)} total records from {page_num} pages")
    return records, newest_time
    • requirements.txt:
    functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0

  3. 點選「部署」即可儲存並部署函式。

建立 Cloud Scheduler 工作

  1. 前往 GCP 主控台的「Cloud Scheduler」。
  2. 點選「建立工作」。

  3. 請提供下列設定詳細資料：

    設定 值
    名稱 netskope-alerts-collector-hourly
    區域 選取與 Cloud Run 函式相同的區域
    頻率 0 * * * * (每小時整點)
    時區 選取時區 (建議使用世界標準時間)
    目標類型 Pub/Sub
    主題 選取「netskope-alerts-trigger」
    郵件內文 {}

  4. 點選「建立」。

在 Google SecOps 中設定動態饋給

  1. 依序前往「SIEM 設定」>「動態消息」。
  2. 依序點選「新增動態消息」>「設定單一動態消息」。
  3. 動態饋給名稱：Netskope Alerts。
  4. 來源類型：Google Cloud Storage V2。
  5. 記錄類型：Netskope 警示。
  6. 按一下「取得服務帳戶」，然後複製系統提供的電子郵件地址。
  7. 點選「下一步」。
  8. 儲存空間值區網址：gs://netskope-alerts-logs/netskope/alerts/ (請加上結尾斜線)。
  9. 來源刪除選項：根據偏好設定選取 (例如測試時選取「永不」)。
  10. 點選「下一步」，檢查內容，然後點選「提交」。

將 IAM 權限授予 Google SecOps 服務帳戶

  1. 依序前往「Cloud Storage」>「Buckets」。
  2. 按一下 netskope-alerts-logs。
  3. 前往「權限」分頁標籤。
  4. 按一下「授予存取權」。
  5. 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址。
  6. 指派角色：選取「Storage 物件檢視者」。(如果使用動態消息中的刪除選項，請使用「Storage 物件管理員」)。
  7. 按一下 [儲存]。

UDM 對應表

記錄欄位 UDM 對應 邏輯
app_activity_label additional.fields 已合併
count_label additional.fields 已合併
object_field additional.fields 已合併
object_id_field additional.fields 已合併
object_type_field additional.fields 已合併
publisher_cn_field additional.fields 已合併
publisher_name_field additional.fields 已合併
site_label additional.fields 已合併
tunnel_id_field additional.fields 已合併
tunnel_type_field additional.fields 已合併
access_method extensions.auth.auth_details 直接對應
accessmethod extensions.auth.auth_details 當 result != `` 時對應
activity extensions.auth.type 對應：Login Failed → MACHINE、Login Successful → MACHINE、Login Attempt → MACHINE
published metadata.event_timestamp 已剖析為 RFC 3339
timestamp metadata.event_timestamp 已剖析為 RFC 3339
Protocol metadata.event_type 已對應：HTTP → NETWORK_HTTP
activity metadata.event_type 已對應：Introspection Scan → EMAIL_UNCATEGORIZED、Login Failed → USER_LOGIN、「Login ...
has_principal metadata.event_type 已對應：true → NETWORK_HTTP、true → NETWORK_CONNECTION
Id metadata.product_log_id 當 result != `` 時對應
_id metadata.product_log_id 直接對應
protocol network.application_protocol 已重新命名/對應
from_user network.email.from 直接對應
useridemail network.email.to 已對應：@ → useridemail
user_agent network.http.parsed_user_agent 已重新命名/對應
useragent network.http.parsed_user_agent 已重新命名/對應
browser_version network.http.parsed_user_agent.browser_version 直接對應
browserversion network.http.parsed_user_agent.browser_version 當 result != `` 時對應
page network.http.referral_url 直接對應
referer network.http.referral_url 已重新命名/對應
Browser network.http.user_agent 當 result != `` 時對應
browser network.http.user_agent 直接對應
user_agent network.http.user_agent 直接對應
useragent network.http.user_agent 直接對應
ip_protocol_out network.ip_protocol 直接對應
server_bytes network.received_bytes 已重新命名/對應
client_bytes network.sent_bytes 已重新命名/對應
browserSessionid network.session_id 當 result != `` 時對應
browser_sess network.session_id 已重新命名/對應
network_session_id network.session_id 已重新命名/對應
ja3 network.tls.client.ja3 直接對應
ja3s network.tls.server.ja3s 直接對應
netskope_pop observer.hostname 直接對應
organization_unit principal.administrative_domain 直接對應
Host principal.asset.hostname 當 result != `` 時對應
hostname principal.asset.hostname 直接對應
instance_id principal.asset.hostname 直接對應
srcIP principal.asset.ip 已合併
srcip principal.asset.ip 已合併
Host principal.hostname 當 result != `` 時對應
hostname principal.hostname 直接對應
instance_id principal.hostname 直接對應
srcIP principal.ip 已合併
srcip principal.ip 已合併
src_location principal.location.city 已重新命名/對應
src_country principal.location.country_or_region 已重新命名/對應
srccountry principal.location.country_or_region 當 result != `` 時對應
src_region principal.location.name 已重新命名/對應
srcregion principal.location.name 當 result != `` 時對應
srclatitude principal.location.region_coordinates.latitude 已重新命名/對應
srclongitude principal.location.region_coordinates.longitude 已重新命名/對應
src_latitude principal.location.region_latitude 已重新命名/對應
src_longitude principal.location.region_longitude 已重新命名/對應
os principal.platform 對應：(?i)Windows → WINDOWS、(?i)MAC → MAC、(?i)LINUX → LINUX
os_version principal.platform_version 已重新命名/對應
osversion principal.platform_version 當 result != `` 時對應
srcport principal.port 已重新命名/對應
from_user_category_label principal.resource.attribute.labels 已合併
Device principal.resource.resource_subtype 當 result != `` 時對應
device principal.resource.resource_subtype 直接對應
User principal.user.email_addresses 已合併
matched_username principal.user.email_addresses 已合併
user principal.user.email_addresses 已合併
user principal.user.userid 已重新命名/對應
_security_result security_result 已合併
security_result1 security_result 已合併
app target.application 已重新命名/對應
appName target.application 當 result != `` 時對應
domain target.asset.hostname 直接對應
dstIP target.asset.ip 已合併
dst_ip target.asset.ip 已合併
dstip target.asset.ip 已合併
dlp_file target.file.full_path 直接對應
file_path target.file.full_path 直接對應
md5 target.file.md5 已重新命名/對應
file_type target.file.mime_type 直接對應
sha256 target.file.sha256 已重新命名/對應
file_size target.file.size 已重新命名/對應
domain target.hostname 直接對應
dstIP target.ip 已合併
dst_ip target.ip 已合併
dstip target.ip 已合併
dst_location target.location.city 已重新命名/對應
dst_country target.location.country_or_region 已重新命名/對應
dstcountry target.location.country_or_region 當 result != `` 時對應
dst_region target.location.name 已重新命名/對應
dstregion target.location.name 當 result != `` 時對應
dstlatitude target.location.region_coordinates.latitude 已重新命名/對應
dstlongitude target.location.region_coordinates.longitude 已重新命名/對應
dstport target.port 已重新命名/對應
app_session_label target.resource.attribute.labels 已合併
to_user_category_label target.resource.attribute.labels 已合併
Url target.url 當 result != `` 時對應
url target.url 已重新命名/對應
to_user target.user.email_addresses 已對應：^.+@.+$ → touser
touser target.user.email_addresses 已對應：^.+@.+$ → touser
不適用 extensions.auth.type 常數：MACHINE
不適用 metadata.event_type 常數：GENERIC_EVENT
不適用 metadata.product_name 常數：Netskope Alert
不適用 metadata.vendor_name 常數：Netskope
不適用 network.http.parsed_user_agent.family 常數：USER_DEFINED
不適用 principal.platform 常數：WINDOWS
不適用 principal.resource.type 常數：DEVICE
不適用 target.asset.hostname 常數：dsthost
不適用 target.hostname 常數：dsthost
action security_result.action_details 從變更記錄對應
dlp_rule_severity security_result.severity 從變更記錄對應
useragent" and "user_agent network.http.user_agent 從變更記錄對應
traffic_type security_result.detection_fields 從變更記錄對應
device principal.resource.resource_sub_type 從變更記錄對應
type security_result.detection_fields 從變更記錄對應
app_session_id target.resource.attribute.labels 從變更記錄對應
malware_severity security_result.severity 從變更記錄對應
malware_type security_result.detection_fields 從變更記錄對應
threat_match_field security_result.detection_fields 從變更記錄對應
cci", "ccl additional.fields 從變更記錄對應
dlp_profile security_result.rule_type 從變更記錄對應
dlp_rule security_result.rule_name 從變更記錄對應
security_result.alert_state ALERTING 從變更記錄對應
security_result.alert_state NOT_ALERTING 從變更記錄對應
security_result.alert_state UNSPECIFIED 從變更記錄對應

變更記錄

查看這個剖析器的變更記錄

還有其他問題嗎？向社群成員和 Google SecOps 專業人員尋求答案。