收集 Netskope 提醒日志

支持的平台：

本文档介绍了如何使用 Google Cloud Storage V2 将 Netskope Alerts 日志注入到 Google Security Operations。Netskope 是一个云安全平台，可为云服务、网站和私有应用提供实时数据和威胁防护。Netskope 提醒会捕获安全事件，包括 SaaS、IaaS 和 Web 流量中的 DLP 违规行为、恶意软件检测、异常行为、泄露的凭据和政策违规行为。

如需了解详情，请参阅收集 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 v2 令牌

  1. 登录 Netskope 租户管理控制台（例如 https://your-tenant.goskope.com）。
  2. 依次前往设置 > 工具 > REST API v2。
  3. 点击新建令牌。
  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 存储桶

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

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

    设置 值
    为存储桶命名 输入一个全局唯一的名称（例如 netskope-alerts-logs）
    位置类型 根据您的需求进行选择（区域级、双区域、多区域）
    位置 选择离您的 Google SecOps 实例最近的位置
    存储类别 标准（建议用于经常访问的日志）
    访问权限控制 均匀（推荐）

  6. 点击创建。

为 Cloud Run 函数创建服务账号

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

创建服务账号

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

授予对 GCS 存储桶的 IAM 权限

  1. 依次前往 Cloud Storage > 存储分区。
  2. 点击存储桶 netskope-alerts-logs。
  3. 前往权限标签页。
  4. 点击授予访问权限。
  5. 提供以下配置详细信息：
    • 添加主账号：输入服务账号电子邮件地址（例如 netskope-alerts-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：netskope-alerts-trigger。
    • 将其他设置保留为默认值。
  4. 点击创建。

创建 Cloud Run 函数以收集日志

Cloud Run 函数将由来自 Cloud Scheduler 的 Pub/Sub 消息触发，以从 Netskope REST API v2 中提取提醒并将其写入 GCS。

  1. 在 GCP 控制台中，前往 Cloud Run。
  2. 点击创建服务。
  3. 选择函数。

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

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

  5. 在触发器部分中：

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

  6. 在身份验证部分中，选择需要身份验证，然后选中 Identity and Access Management (IAM)。

  7. 向下滚动到容器、网络、安全性。

  8. 在安全性标签页中，选择服务账号 netskope-alerts-collector-sa。

  9. 在容器标签页中，点击变量和密钥，然后添加以下内容：

    变量名称 示例值 说明
    GCS_BUCKET netskope-alerts-logs GCS 存储桶名称
    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. 在请求部分中，将请求超时设置为 600 秒。

  11. 在设置标签页（资源）中，选择 512 MiB 内存和 1 个 CPU。

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

添加函数代码

  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 Console 中，前往 Cloud Scheduler。
  2. 点击创建作业。

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

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

  4. 点击创建。

在 Google SecOps 中配置 Feed

  1. 依次前往 SIEM 设置 > Feed。
  2. 依次点击添加新 Feed > 配置单个 Feed。
  3. Feed 名称：Netskope Alerts。
  4. 来源类型：Google Cloud Storage V2。
  5. 日志类型：Netskope 提醒。
  6. 点击 Get Service Account，然后复制提供的电子邮件地址。
  7. 点击下一步。
  8. 存储桶网址：gs://netskope-alerts-logs/netskope/alerts/（包含结尾斜杠）。
  9. 来源删除选项：根据您的偏好进行选择（例如，选择从不以进行测试）。
  10. 点击下一步，查看，然后点击提交。

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

  1. 依次前往 Cloud Storage > 存储分区。
  2. 点击 netskope-alerts-logs。
  3. 前往权限标签页。
  4. 点击授予访问权限。
  5. 添加主账号：粘贴 Google SecOps 服务账号电子邮件地址。
  6. 分配角色：选择 Storage Object Viewer。（如果使用 Feed 中的删除选项，请使用 Storage Object Admin）。
  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、“登录…”
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 专业人士的解答。