收集 SAP Cloud Identity Services 日志

支持的平台:

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

SAP Cloud Identity Services 是一种基于云的身份和访问管理 (IAM) 解决方案,可为 SAP 和第三方应用提供身份验证、单点登录 (SSO) 和用户配置功能。SAP BTP 中的审核日志服务会记录身份验证租户的身份验证事件、数据更改和配置更改。这些审核日志可以通过 SAP BTP 审核日志检索 API 进行检索,并写入 GCS 存储桶以供 Google SecOps 注入。

准备工作

请确保满足以下前提条件:

  • Google SecOps 实例
  • 已启用 Cloud Storage API 的 GCP 项目
  • 创建和管理 GCS 存储分区的权限
  • 创建 Cloud Run 服务、Pub/Sub 主题和 Cloud Scheduler 作业的权限
  • 具有 Cloud Foundry 子账号的 SAP BTP 全局账号
  • SAP Cloud Identity Services 租户 (Identity Authentication)
  • 在 SAP BTP Cloud Foundry 子账号中具有空间开发者角色的用户
  • 已添加到子账号的“审核日志管理”服务使用权

创建 auditlog-management 服务实例

在 SAP BTP Cloud Foundry 子账号中创建 auditlog-management 服务的服务实例,以获取访问 Audit Log Retrieval API 所需的凭据。

配置服务授权

  1. 登录 SAP BTP 主控室
  2. 前往您的子账号。
  3. 前往授权
  4. 依次点击修改> 添加服务方案
  5. 搜索审核日志管理,然后选中默认复选框。
  6. 点击添加 1 个服务方案
  7. 点击保存

创建服务实例和密钥

  1. 安装 Cloud Foundry CLI(如果尚未安装)。
  2. 登录 Cloud Foundry 环境:

    cf login -a <CF_API_ENDPOINT> -o <ORG> -s <SPACE> -u <USER>
    
  3. 创建具有 mTLS 身份验证的 auditlog-management 服务实例(推荐):

    cf create-service auditlog-management default sap-ias-auditlog \
      -c '{"xs-security": {"xsappname": "sap-ias-auditlog", "oauth2-configuration": {"credential-types": ["x509"], "grant-types": ["client_credentials"]}}}'
    
  4. 创建具有 X.509 证书的服务钥匙:

    cf create-service-key sap-ias-auditlog sap-ias-auditlog-key \
      -c '{"xsuaa": {"credential-type": "x509", "x509": {"key-length": 2048, "validity": 365, "validity-type": "DAYS"}}}'
    
  5. 检索服务密钥:

    cf service-key sap-ias-auditlog sap-ias-auditlog-key
    
  6. 在服务钥匙输出中,记下以下值:

    • uaa.certurl:OAuth2 令牌端点网址(例如 https://yoursubaccount.authentication.eu10.hana.ondemand.com
    • uaa.clientid:OAuth2 客户端 ID
    • uaa.certificate:X.509 证书(PEM 格式)
    • uaa.key:RSA 私钥(PEM 格式)
    • url:审核日志检索 API 的基本网址(例如 https://auditlog-management.cfapps.eu10.hana.ondemand.com
  7. 将证书和私钥提取并保存到 PEM 文件中:

    cf service-key sap-ias-auditlog sap-ias-auditlog-key | sed '/Getting key/d' | jq --raw-output .credentials.uaa.certificate > certificate.pem
    cf service-key sap-ias-auditlog sap-ias-auditlog-key | sed '/Getting key/d' | jq --raw-output .credentials.uaa.key > private-key.pem
    

配置 Identity Authentication 审核日志传送

  1. 登录 https://<tenant-id>.accounts.ondemand.com/admin,进入 SAP Cloud Identity Services 的管理控制台
  2. 前往审核和更改日志板块。
  3. 选择审核日志标签页(如果您的租户位于 SAP 基础架构上,则选择 Cloud Foundry 标签页)。
  4. 点击 +Add(+添加)。
  5. 提供以下配置详细信息:

    • 租户 ID:输入 Cloud Foundry 子账号的租户 ID
    • 区域:选择与您的身份验证区域对应的 SAP BTP Cloud Foundry 区域
    • 子网域(可选):输入子账号子网域,以启用指向审核日志查看器的直接链接
  6. 点击保存

测试 API 访问权限

  • 在继续进行集成之前,请先测试您的凭据:

    # Extract values from the service key
    CERT_URL="<uaa.certurl value>"
    CLIENT_ID="<uaa.clientid value>"
    API_URL="<url value>"
    
    # Get OAuth access token using mTLS
    TOKEN=$(curl -s --cert certificate.pem --key private-key.pem \
      -X POST "${CERT_URL}/oauth/token?grant_type=client_credentials&client_id=${CLIENT_ID}" \
      | python3 -c "import sys,json; print(json.load(sys.stdin).get('access_token',''))")
    
    echo "Token: ${TOKEN:0:50}..."
    
    # Test audit log retrieval
    curl -v -H "Authorization: Bearer ${TOKEN}" \
      "${API_URL}/auditlog/v2/auditlogrecords?time_from=$(date -u -v-1d '+%Y-%m-%dT%H:%M:%S')&time_to=$(date -u '+%Y-%m-%dT%H:%M:%S')"
    

创建 Google Cloud Storage 存储桶

  1. 前往 Google Cloud 控制台
  2. 选择您的项目或创建新项目。
  3. 在导航菜单中,依次前往 Cloud Storage > 存储分区
  4. 点击创建存储分区
  5. 提供以下配置详细信息:

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

为 Cloud Run 函数创建服务账号

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

创建服务账号

  1. GCP 控制台中,依次前往 IAM 和管理 > 服务账号
  2. 点击创建服务账号
  3. 提供以下配置详细信息:
    • 服务账号名称:输入 sap-ias-logs-collector-sa
    • 服务账号说明:输入 Service account for Cloud Run function to collect SAP Cloud Identity Services 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 权限

向服务账号授予对 GCS 存储桶的写入权限:

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

创建 Cloud Run 函数以收集日志

Cloud Run 函数将由来自 Cloud Scheduler 的 Pub/Sub 消息触发,以从 SAP BTP 审核日志检索 API 中提取审核日志并将其写入 GCS。

  1. GCP 控制台中,前往 Cloud Run
  2. 点击创建服务
  3. 选择函数(使用内嵌编辑器创建函数)。
  4. 配置部分中,提供以下配置详细信息:

    设置
    Service 名称 sap-ias-logs-collector
    区域 选择与您的 GCS 存储桶匹配的区域(例如 us-central1
    运行时 选择 Python 3.12 或更高版本
  5. 触发器(可选)部分中:

    1. 点击 + 添加触发器
    2. 选择 Cloud Pub/Sub
    3. 选择 Cloud Pub/Sub 主题部分,选择主题 sap-ias-logs-trigger
    4. 点击保存
  6. 身份验证部分中:

    1. 选择需要进行身份验证
    2. 检查 Identity and Access Management (IAM)
  7. 向下滚动并展开容器、网络、安全性

  8. 前往安全性标签页:

    • 服务账号:选择服务账号 sap-ias-logs-collector-sa
  9. 前往容器标签页:

    1. 点击变量和密钥
    2. 为每个环境变量点击+ 添加变量
    变量名称 示例值 说明
    GCS_BUCKET sap-ias-audit-logs GCS 存储桶名称
    GCS_PREFIX sap-ias 日志文件的前缀
    STATE_KEY sap-ias/state.json 状态文件路径
    SAP_CERT_URL https://sub.authentication.eu10.hana.ondemand.com OAuth2 令牌端点 (uaa.certurl)
    SAP_CLIENT_ID sb-sap-ias-auditlog!t12345 OAuth2 客户端 ID (uaa.clientid)
    SAP_CERTIFICATE PEM 证书内容 X.509 证书 (uaa.certificate)
    SAP_PRIVATE_KEY PEM 私钥内容 RSA 私钥 (uaa.key)
    SAP_API_URL https://auditlog-management.cfapps.eu10.hana.ondemand.com 审核日志检索 API 基本网址
    MAX_RECORDS 10000 每次运行的记录数上限
    LOOKBACK_HOURS 2 初始回溯期
  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 tempfile
      from datetime import datetime, timezone, timedelta
      import time
      
      # Initialize Storage client
      storage_client = storage.Client()
      
      # Environment variables
      GCS_BUCKET = os.environ.get('GCS_BUCKET')
      GCS_PREFIX = os.environ.get('GCS_PREFIX', 'sap-ias').strip('/')
      STATE_KEY = os.environ.get('STATE_KEY') or f"{GCS_PREFIX}/state.json"
      SAP_CERT_URL = os.environ.get('SAP_CERT_URL', '').rstrip('/')
      SAP_CLIENT_ID = os.environ.get('SAP_CLIENT_ID')
      SAP_CERTIFICATE = os.environ.get('SAP_CERTIFICATE', '')
      SAP_PRIVATE_KEY = os.environ.get('SAP_PRIVATE_KEY', '')
      SAP_API_URL = os.environ.get('SAP_API_URL', '').rstrip('/')
      MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000'))
      LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '2'))
      
      def parse_datetime(value):
        """Parse ISO datetime string to datetime object."""
        if not value:
          return None
        if value.endswith("Z"):
          value = value[:-1] + "+00:00"
        try:
          return datetime.fromisoformat(value)
        except Exception:
          return None
      
      def write_pem_to_tempfile(pem_content, suffix='.pem'):
        """Write PEM content to a temporary file and return the path."""
        tmp = tempfile.NamedTemporaryFile(
          mode='w', suffix=suffix, delete=False
        )
        tmp.write(pem_content)
        tmp.flush()
        tmp.close()
        return tmp.name
      
      def get_oauth_token(cert_file, key_file):
        """Get OAuth2 access token using mTLS client credentials."""
        http = urllib3.PoolManager(
          cert_file=cert_file,
          key_file=key_file,
          timeout=urllib3.Timeout(connect=10.0, read=30.0),
          retries=False,
        )
      
        token_url = (
          f"{SAP_CERT_URL}/oauth/token"
          f"?grant_type=client_credentials"
          f"&client_id={SAP_CLIENT_ID}"
        )
      
        response = http.request('POST', token_url)
      
        if response.status != 200:
          raise Exception(
            f"Failed to get OAuth token: HTTP {response.status} - "
            f"{response.data.decode('utf-8')}"
          )
      
        token_data = json.loads(response.data.decode('utf-8'))
        return token_data['access_token']
      
      @functions_framework.cloud_event
      def main(cloud_event):
        """
        Cloud Run function triggered by Pub/Sub to fetch
        SAP Cloud Identity Services audit logs and write to GCS.
      
        Args:
          cloud_event: CloudEvent object containing Pub/Sub message
        """
      
        if not all([
          GCS_BUCKET, SAP_CERT_URL, SAP_CLIENT_ID,
          SAP_CERTIFICATE, SAP_PRIVATE_KEY, SAP_API_URL
        ]):
          print('Error: Missing required environment variables')
          return
      
        cert_file = None
        key_file = None
      
        try:
          bucket = storage_client.bucket(GCS_BUCKET)
      
          # Write certificate and key to temp files for mTLS
          cert_file = write_pem_to_tempfile(SAP_CERTIFICATE)
          key_file = write_pem_to_tempfile(SAP_PRIVATE_KEY)
      
          # Get OAuth2 access token
          access_token = get_oauth_token(cert_file, key_file)
          print("Successfully obtained OAuth2 access token")
      
          # Load state
          state = load_state(bucket, STATE_KEY)
      
          # Determine time window
          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"])
              # Overlap by 2 minutes to catch delayed events
              if last_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)
      
          time_from = last_time.strftime('%Y-%m-%dT%H:%M:%S')
          time_to = now.strftime('%Y-%m-%dT%H:%M:%S')
      
          print(f"Fetching audit logs from {time_from} to {time_to}")
      
          # Fetch audit logs
          records, newest_event_time = fetch_audit_logs(
            access_token=access_token,
            time_from=time_from,
            time_to=time_to,
          )
      
          if not records:
            print("No new audit log records found.")
            save_state(bucket, STATE_KEY, now.isoformat())
            return
      
          # Write to GCS as NDJSON
          timestamp = now.strftime('%Y%m%d_%H%M%S')
          object_key = f"{GCS_PREFIX}/logs_{timestamp}.ndjson"
          blob = bucket.blob(object_key)
      
          ndjson = '\n'.join(
            [json.dumps(record, ensure_ascii=False) for record in records]
          ) + '\n'
          blob.upload_from_string(
            ndjson, content_type='application/x-ndjson'
          )
      
          print(
            f"Wrote {len(records)} records to "
            f"gs://{GCS_BUCKET}/{object_key}"
          )
      
          # Update state
          if newest_event_time:
            save_state(bucket, STATE_KEY, newest_event_time)
          else:
            save_state(bucket, STATE_KEY, now.isoformat())
      
          print(f"Successfully processed {len(records)} records")
      
        except Exception as e:
          print(f'Error processing audit logs: {str(e)}')
          raise
        finally:
          # Clean up temp files
          for f in [cert_file, key_file]:
            if f:
              try:
                os.unlink(f)
              except OSError:
                pass
      
      def load_state(bucket, key):
        """Load state from GCS."""
        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_iso):
        """Save the last event timestamp to GCS state file."""
        try:
          state = {'last_event_time': last_event_time_iso}
          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_iso}")
        except Exception as e:
          print(f"Warning: Could not save state: {e}")
      
      def fetch_audit_logs(access_token, time_from, time_to):
        """
        Fetch audit logs from SAP BTP Audit Log Retrieval API
        with handle-based pagination.
      
        Args:
          access_token: OAuth2 bearer token
          time_from: Start time in YYYY-MM-DDTHH:MM:SS format (UTC)
          time_to: End time in YYYY-MM-DDTHH:MM:SS format (UTC)
      
        Returns:
          Tuple of (records list, newest_event_time ISO string)
        """
        http = urllib3.PoolManager(
          timeout=urllib3.Timeout(connect=5.0, read=30.0),
          retries=False,
        )
      
        headers = {
          'Authorization': f'Bearer {access_token}',
          'Accept': 'application/json',
        }
      
        endpoint = f"{SAP_API_URL}/auditlog/v2/auditlogrecords"
      
        records = []
        newest_time = None
        page_num = 0
        handle = None
        backoff = 1.0
      
        while True:
          if len(records) >= MAX_RECORDS:
            print(f"Reached max_records limit ({MAX_RECORDS})")
            break
      
          # Build URL with parameters
          if handle:
            url = f"{endpoint}?handle={handle}"
          else:
            url = (
              f"{endpoint}"
              f"?time_from={time_from}"
              f"&time_to={time_to}"
            )
      
          try:
            response = http.request('GET', url, headers=headers)
      
            # Handle rate limiting with exponential backoff
            if response.status == 429:
              retry_after = int(
                response.headers.get(
                  'Retry-After', str(int(backoff))
                )
              )
              print(
                f"Rate limited (429). "
                f"Retrying after {retry_after}s..."
              )
              time.sleep(retry_after)
              backoff = min(backoff * 2, 30.0)
              continue
      
            backoff = 1.0
      
            if response.status == 204:
              print("No more audit log records (HTTP 204)")
              break
      
            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
      
            page_results = json.loads(
              response.data.decode('utf-8')
            )
      
            if not page_results:
              print("No more results (empty response)")
              break
      
            # Handle single record vs list
            if isinstance(page_results, dict):
              page_results = [page_results]
      
            page_num += 1
            print(
              f"Page {page_num}: "
              f"Retrieved {len(page_results)} events"
            )
            records.extend(page_results)
      
            # Track newest event time
            for event in page_results:
              try:
                event_time = event.get('time')
                if event_time:
                  if newest_time is None:
                    newest_time = event_time
                  else:
                    current_dt = parse_datetime(event_time)
                    newest_dt = parse_datetime(newest_time)
                    if (current_dt and newest_dt
                        and current_dt > newest_dt):
                      newest_time = event_time
              except Exception as e:
                print(
                  f"Warning: Could not parse event time: {e}"
                )
      
            # Check for pagination handle in response header
            paging_header = response.headers.get('Paging', '')
            if paging_header and 'handle=' in paging_header:
              handle = paging_header.split('handle=')[1].strip()
              print(f"Next page handle: {handle[:50]}...")
            else:
              print("No more pages (no paging handle)")
              break
      
          except Exception as e:
            print(f"Error fetching audit logs: {e}")
            return [], None
      
        print(
          f"Retrieved {len(records)} total records "
          f"from {page_num} pages"
        )
        return records, newest_time
      
    • 第二个文件 - 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 函数。

  1. GCP Console 中,前往 Cloud Scheduler
  2. 点击创建作业
  3. 提供以下配置详细信息:

    设置
    名称 sap-ias-logs-collector-hourly
    区域 选择与 Cloud Run 函数相同的区域
    频率 0 * * * *(每小时一次,整点时)
    时区 选择时区(建议选择世界协调时间 [UTC])
    目标类型 Pub/Sub
    主题 选择主题 sap-ias-logs-trigger
    消息正文 {}(空 JSON 对象)
  4. 点击创建

时间表频率选项

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

频率 Cron 表达式 使用场景
每小时 0 * * * * 标准(推荐)
每 2 小时 0 */2 * * * 调低音量
每 6 小时 0 */6 * * * 低成交量、批处理

测试集成

  1. Cloud Scheduler 控制台中,找到您的作业。
  2. 点击强制运行以手动触发作业。
  3. 等待几秒钟。
  4. 前往 Cloud Run > 服务
  5. 点击函数名称 sap-ias-logs-collector
  6. 点击日志标签页。
  7. 验证函数是否已成功执行。查找:

    Successfully obtained OAuth2 access token
    Fetching audit logs from YYYY-MM-DDTHH:MM:SS to YYYY-MM-DDTHH:MM:SS
    Page 1: Retrieved X events
    Wrote X records to gs://sap-ias-audit-logs/sap-ias/logs_YYYYMMDD_HHMMSS.ndjson
    Successfully processed X records
    
  8. 前往 Cloud Storage > 存储分区

  9. 点击您的存储桶名称。

  10. 前往前缀文件夹 sap-ias/

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

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

  • 未能获取 OAuth 令牌:HTTP 401:检查环境变量中的证书和私钥。验证它们是否与服务钥匙一致。确保服务钥匙未过期。
  • 在检索审核日志时出现 HTTP 401 错误:OAuth 访问令牌可能已过期。该函数会在每次调用时获取新令牌。
  • HTTP 403:验证 auditlog-management 服务实例是否已正确创建,以及服务钥匙是否有效。
  • HTTP 429:速率限制 - 函数会自动重试,并采用指数退避算法。如果这种情况经常发生,请考虑延长调度间隔。
  • HTTP 204:在指定的时间范围内未找到任何审核日志记录。对于活动较少的租户,这是正常现象。
  • 缺少环境变量:检查是否已在 Cloud Run 函数配置中设置所有必需的变量。

在 Google SecOps 中配置 Feed 以注入 SAP Cloud Identity Services 日志

  1. 依次前往 SIEM 设置 > Feed
  2. 点击添加新 Feed
  3. 点击配置单个 Feed
  4. Feed 名称字段中,输入 Feed 的名称(例如 SAP Cloud Identity Services Audit Logs)。
  5. 选择 Google Cloud Storage V2 作为来源类型
  6. 选择 SAP 身份和身份验证数据作为日志类型
  7. 点击获取服务账号。系统会显示一个唯一的服务账号电子邮件地址,例如:

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
    
  8. 复制此电子邮件地址,以便在下一步中使用。

  9. 点击下一步

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

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

      gs://sap-ias-audit-logs/sap-ias/
      
      • 替换:
        • sap-ias-audit-logs:您的 GCS 存储桶名称。
        • sap-ias:存储日志的可选前缀/文件夹路径(留空表示根目录)。
    • 来源删除选项:根据您的偏好选择删除选项:

      • 永不:转移后永不删除任何文件(建议用于测试)。
      • 删除已转移的文件:在成功转移后删除文件。
      • 删除已转移的文件和空目录:成功转移后删除文件和空目录。
    • 文件存在时间上限:包含在过去指定天数内修改的文件(默认值为 180 天)

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

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

  11. 点击下一步

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

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

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

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

UDM 映射表

日志字段 UDM 映射 逻辑
childAgr about.group.group_display_name 直接映射
field1 additional.fields 已合并
has_principal_hostname metadata.event_type 已映射:trueSTATUS_UPDATE
has_principal_user metadata.event_type 已映射:trueUSER_CHANGE_PASSWORDtrueGROUP_MODIFICATION
has_target_user metadata.event_type 已映射:trueUSER_UNCATEGORIZED
field1 principal.asset.attribute.labels 已合并
key principal.asset.attribute.labels 已映射:"systemid", "partype"field1
hostname principal.hostname 直接映射
repid principal.process.file.full_path 直接映射
aname principal.user.userid 直接映射
changeUsr principal.user.userid 直接映射
modbe principal.user.userid 直接映射
modifier principal.user.userid 直接映射
uname principal.user.userid 直接映射
field1 security_result.detection_fields 已合并
key security_result.detection_fields 已映射:"direct", "uflag","securityPolicy"field1
agrName target.group.group_display_name 直接映射
field1 target.resource.attribute.labels 已合并
key target.resource.attribute.labels 映射:`"field","struct","mandt","parentAgr","type","fromDat","toDat","ustyp","langu","iana...
tcode target.resource.name 直接映射
class target.user.group_identifiers 已合并
profile target.user.group_identifiers 已合并
bname target.user.userid 直接映射
username target.user.userid 直接映射
不适用 metadata.event_type 常量:USER_UNCATEGORIZED
不适用 metadata.product_name 常量:SAP_IDENTITY_AND_AUTH_DATA
不适用 metadata.vendor_name 常量:SAP
changeUsr event.idm.read_only_udm.principal.user.userid 从变更日志映射
uname event.idm.read_only_udm.principal.user.userid 从变更日志映射
aname event.idm.read_only_udm.principal.user.userid 从变更日志映射
modifier event.idm.read_only_udm.principal.user.userid 从变更日志映射
modbe event.idm.read_only_udm.principal.user.userid 从变更日志映射
hostname event.idm.read_only_udm.principal.hostname 从变更日志映射
bname event.idm.read_only_udm.target.user.userid 从变更日志映射
username event.idm.read_only_udm.target.user.userid 从变更日志映射
agrName event.idm.read_only_udm.target.group.group_display_name 从变更日志映射
childAgr event.idm.read_only_udm.about.group.group_display_name 从变更日志映射
tcode event.idm.read_only_udm.target.resource.name 从变更日志映射
class event.idm.read_only_udm.target.user.group_identifiers 从变更日志映射
profile event.idm.read_only_udm.target.user.group_identifiers 从变更日志映射
repid event.idm.read_only_udm.principal.process.file.full_path 从变更日志映射
field event.idm.read_only_udm.target.resource.attribute.labels 从变更日志映射
struct event.idm.read_only_udm.target.resource.attribute.labels 从变更日志映射
mandt event.idm.read_only_udm.target.resource.attribute.labels 从变更日志映射
parentAgr event.idm.read_only_udm.target.resource.attribute.labels 从变更日志映射
type event.idm.read_only_udm.target.resource.attribute.labels 从变更日志映射
fromDat event.idm.read_only_udm.target.resource.attribute.labels 从变更日志映射
toDat event.idm.read_only_udm.target.resource.attribute.labels 从变更日志映射
ustyp event.idm.read_only_udm.target.resource.attribute.labels 从变更日志映射
langu event.idm.read_only_udm.target.resource.attribute.labels 从变更日志映射
ianatzonecode event.idm.read_only_udm.target.resource.attribute.labels 从变更日志映射
trdat event.idm.read_only_udm.target.resource.attribute.labels 从变更日志映射
ltime event.idm.read_only_udm.target.resource.attribute.labels 从变更日志映射
pwdchgdate event.idm.read_only_udm.target.resource.attribute.labels 从变更日志映射
pwdlgndate event.idm.read_only_udm.target.resource.attribute.labels 从变更日志映射
pwdlockdate event.idm.read_only_udm.target.resource.attribute.labels 从变更日志映射
pwdsaltedhash event.idm.read_only_udm.target.resource.attribute.labels 从变更日志映射
systemid event.idm.read_only_udm.principal.asset.attribute.labels 从变更日志映射
partype event.idm.read_only_udm.principal.asset.attribute.labels 从变更日志映射
direct event.idm.read_only_udm.security_result.detection_fields 从变更日志映射
uflag event.idm.read_only_udm.security_result.detection_fields 从变更日志映射
securityPolicy event.idm.read_only_udm.security_result.detection_fields 从变更日志映射

更新日志

查看相应解析器的更改日志

需要更多帮助?获得社区成员和 Google SecOps 专业人士的解答。