收集 BeyondTrust Endpoint Privilege Management (EPM) 記錄

支援語言：

本文說明如何使用 Cloud Storage，將 BeyondTrust Endpoint Privilege Management (EPM) 記錄檔擷取至 Google Security Operations。剖析器主要負責將 BeyondTrust Endpoint 的原始 JSON 記錄資料轉換為符合 Google SecOps UDM 的結構化格式。這個函式會先初始化各種欄位的預設值，然後剖析 JSON 酬載，接著將原始記錄中的特定欄位對應至 event.idm.read_only_udm 物件中的對應 UDM 欄位。

事前準備

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

  • Google SecOps 執行個體
  • 已啟用 Cloud Storage API 的 Google Cloud 專案
  • 建立及管理 Cloud Storage bucket 的權限
  • 管理 Cloud Storage 值區身分與存取權管理 (IAM) 政策的權限
  • 建立 Cloud Run 函式服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
  • BeyondTrust Endpoint Privilege Management 租戶或 API 的特殊存取權

建立 Cloud Storage bucket

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

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

    設定 值
    為 bucket 命名 輸入全域不重複的名稱 (例如 beyondtrust-epm-logs)
    位置類型 根據需求選擇 (區域、雙區域、多區域)
    位置 選取位置 (例如 us-central1)
    儲存空間級別 標準 (建議用於經常存取的記錄)
    存取控管 統一 (建議)
    保護工具 選用：啟用物件版本管理或保留政策

  6. 點選「建立」。

收集 BeyondTrust EPM API 憑證

  1. 以管理員身分登入 BeyondTrust Privilege Management 網頁控制台。
  2. 依序前往「設定」>「設定」>「API 設定」。
  3. 按一下「Create an API Account」(建立 API 帳戶)。
  4. 提供下列設定詳細資料：
    • 「Name」(名稱)：輸入 Google SecOps Collector。
    • API 存取：啟用「報表 (唯讀)」。這個收集器使用的端點屬於「回報」/management-api/v3/Events/FromStartDate權限類別。如要收集網路控制台活動稽核記錄，請一併啟用「稽核 (唯讀)」。
  5. 複製並儲存「用戶端 ID」和「用戶端密鑰」。

  6. 複製「API 設定」頁面頂端顯示的 API 基礎網址。這通常是 https://<your-tenant>-services.pm.beyondtrustcloud.com。您會將此項目當做 BPT_API_URL 使用。

為 Cloud Run 函式建立服務帳戶

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

建立服務帳戶

  1. 在 Google Cloud 控制台中，依序前往「IAM & Admin」(IAM 與管理)>「Service Accounts」(服務帳戶)。
  2. 按一下「Create Service Account」(建立服務帳戶)。
  3. 提供下列設定詳細資料：
    • 服務帳戶名稱：輸入 beyondtrust-epm-collector-sa。
    • 服務帳戶說明：輸入 Service account for Cloud Run function to collect BeyondTrust EPM logs。
  4. 按一下「建立並繼續」。
  5. 在「將專案存取權授予這個服務帳戶」部分，新增下列角色：
    1. 按一下「選擇角色」。
    2. 搜尋並選取「Storage 物件管理員」。
    3. 點選「+ 新增其他角色」。
    4. 搜尋並選取「Cloud Run Invoker」。
    5. 點選「+ 新增其他角色」。
    6. 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
  6. 按一下「繼續」。
  7. 按一下「完成」。

這些角色適用於：

  • Storage Object Admin：將記錄檔寫入 Cloud Storage bucket，並管理狀態檔案
  • Cloud Run 叫用者：允許 Pub/Sub 叫用函式
  • Cloud Functions 叫用者：允許叫用函式

授予 Cloud Storage bucket 的 IAM 權限

授予服務帳戶 Cloud Storage 值區的寫入權限：

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

建立 Pub/Sub 主題

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

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

建立 Cloud Run 函式來收集記錄

Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發，從 BeyondTrust EPM API 擷取記錄，並將記錄寫入 Cloud Storage。

  1. 前往 Google Cloud 控制台的「Cloud Run」。
  2. 按一下「Create service」(建立服務)。
  3. 選取「函式」。使用內嵌編輯器建立函式。

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

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

  5. 在「觸發條件 (選用)」部分：

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

  6. 在「Authentication」(驗證) 部分：

    1. 選取「需要驗證」。
    2. 檢查 Identity and Access Management (IAM)。

  7. 前往並展開「Containers, Networking, Security」。

  8. 前往「安全性」分頁：

    • 服務帳戶：選取服務帳戶 beyondtrust-epm-collector-sa。

  9. 前往「容器」分頁：

    1. 按一下「變數與密鑰」。
    2. 針對每個環境變數，按一下「+ 新增變數」：
    變數名稱 範例值
    GCS_BUCKET beyondtrust-epm-logs
    GCS_PREFIX beyondtrust-epm/
    STATE_KEY beyondtrust-epm-state.json
    BPT_API_URL https://yourtenant-services.pm.beyondtrustcloud.com
    CLIENT_ID your-client-id
    CLIENT_SECRET your-client-secret
    RECORD_SIZE 1000
    MAX_BATCHES 50
    LOOKBACK_HOURS 24
    • RECORD_SIZE：每個要求記錄。/Events/FromStartDate 接受 1 到 1000，函式會將較高的值限制在 1000。
    • MAX_BATCHES：每次執行的要求安全上限。如果執行作業時遇到這個字元，游標仍會前進到已寫入的內容，因此下一個執行作業會從該處繼續，而不是重複作業。
    • LOOKBACK_HOURS：第一次執行可追溯的時間範圍。BeyondTrust 會保留被動事件 (代碼 106、107、603、706) 30 天，以及所有其他事件 90 天，因此回溯期較長也不會傳回更舊的資料。
    • LOOKBACK_HOURS：在任何游標存在之前，第一次執行時要回溯多遠。

  10. 在「變數與密鑰」分頁中，前往「要求」：

    • 要求逾時：輸入 600 秒 (10 分鐘)。

  11. 前往「容器」中的「設定」分頁：

    • 在「資源」部分：
      • 記憶體：選取 512 MiB 以上。
      • CPU：選取 1。
    • 按一下「完成」。

  12. 捲動至「執行環境」：

    • 選取「預設」 (建議選項)。

  13. 在「修訂版本資源調度」部分：

    • 「執行個體數量下限」：輸入 0。
    • 「Maximum number of instances」(執行個體數量上限)：輸入 100 (或根據預期負載調整)。

  14. 點選「建立」。

  15. 等待服務建立完成 (1 到 2 分鐘)。

  16. 服務建立完成後，系統會自動開啟內嵌程式碼編輯器。

新增函式程式碼

  1. 在「Function entry point」(函式進入點) 中輸入 main

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

    • 第一個檔案 - main.py:
    import hashlib
import json
import os
import re
import time
import urllib3
import uuid
from datetime import datetime, timedelta, timezone
from urllib.parse import urlencode

import functions_framework
from google.cloud import storage
from google.cloud.exceptions import NotFound

# Initialize HTTP client with timeouts
http = urllib3.PoolManager(
    timeout=urllib3.Timeout(connect=5.0, read=60.0),
    retries=False,
)

# Initialize Storage client
storage_client = storage.Client()

TOKEN_PATH = '/oauth/connect/token'
# Events/FromStartDate is the endpoint BeyondTrust documents for bulk extraction:
# "A new API is exposed to extract the events in bulk." It takes only StartDate and
# RecordSize. Events/search is not usable here: it requires an OperatingSystem value,
# it is a single string with no documented "all" value, so it can only ever return one
# operating system per call.
EVENTS_PATH = '/management-api/v3/Events/FromStartDate'
OAUTH_SCOPE = 'urn:management:api'
# RecordSize accepts 1 to 1000.
MAX_RECORD_SIZE = 1000

class FetchError(Exception):
    """Raised when the BeyondTrust API call fails.

    The cursor must never advance on a failed fetch, otherwise the events in the
    failed window are skipped permanently.
    """

def rfc3339(dt: datetime) -> str:
    """Render a datetime the way StartDate accepts it: milliseconds and a literal Z."""
    return dt.astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z'

def parse_iso(stamp: str) -> datetime:
    """Parse any ISO 8601 timestamp shape the API emits.

    event.ingested comes back with a +00:00 offset and up to seven fractional
    digits, e.g. 2026-08-03T13:29:55.1109163+00:00. fromisoformat's tolerance for
    long fractions varies across Python versions, so the fraction is trimmed to
    microseconds first.
    """
    text = stamp.strip()
    if text.endswith('Z'):
        text = text[:-1] + '+00:00'
    text = re.sub(r'\.(\d{1,6})\d*', r'.\1', text, count=1)
    parsed = datetime.fromisoformat(text)
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed

def canonical(stamp: str) -> str:
    """Re-render an API timestamp into the one form StartDate accepts.

    The API returns event.ingested as +00:00-offset ISO 8601 but rejects that
    same shape as a StartDate value with 400 "Invalid Start date format": it
    accepts only Z-suffixed values. Every timestamp that came from the API must
    pass through here before being sent back or persisted.
    """
    return rfc3339(parse_iso(stamp))

def event_id(evt: dict) -> str:
    """Return the event's identity for deduplication.

    Falls back to a content hash when event.id is absent, so an id-less event
    still deduplicates instead of being re-ingested on every boundary re-read.
    """
    explicit = str((evt.get('event') or {}).get('id') or '')
    if explicit:
        return explicit
    digest = hashlib.sha256(
        json.dumps(evt, sort_keys=True, ensure_ascii=False).encode('utf-8')
    ).hexdigest()
    return f'sha256:{digest}'

def event_ingested(evt: dict) -> str:
    """Return the Elastic ingestion timestamp, which is what StartDate filters on."""
    return str((evt.get('event') or {}).get('ingested') or '')

def next_millisecond(stamp: str) -> str:
    """Return the canonical timestamp one millisecond later.

    Used only when a full batch fits inside a single ingestion millisecond. Without
    this the cursor cannot move and the collector stalls on that timestamp forever.
    """
    return rfc3339(parse_iso(stamp) + timedelta(milliseconds=1))

@functions_framework.cloud_event
def main(cloud_event):
    """Fetch BeyondTrust EPM events and write them to Cloud Storage as NDJSON.

    Args:
            cloud_event: CloudEvent object containing the Pub/Sub message.
    """
    bucket_name = os.environ.get('GCS_BUCKET')
    prefix = os.environ.get('GCS_PREFIX', 'beyondtrust-epm/')
    state_key = os.environ.get('STATE_KEY', 'beyondtrust-epm-state.json')

    api_url = (os.environ.get('BPT_API_URL') or '').rstrip('/')
    client_id = os.environ.get('CLIENT_ID')
    client_secret = os.environ.get('CLIENT_SECRET')
    # Clamp both ends: the API rejects RecordSize outside 1 to 1000, and a
    # misconfigured 0 would otherwise crash-loop on a 400 every run.
    record_size = min(max(int(os.environ.get('RECORD_SIZE', '1000')), 1), MAX_RECORD_SIZE)
    max_batches = int(os.environ.get('MAX_BATCHES', '50'))
    lookback_hours = int(os.environ.get('LOOKBACK_HOURS', '24'))

    if not all([bucket_name, api_url, client_id, client_secret]):
        raise RuntimeError(
            'Missing required environment variables: '
            'GCS_BUCKET, BPT_API_URL, CLIENT_ID, CLIENT_SECRET'
        )

    bucket = storage_client.bucket(bucket_name)
    state = load_state(bucket, state_key)

    # StartDate is a rising cursor on the Elastic ingestion timestamp, not a closed
    # window. There is no EndDate on this endpoint.

    if state.get('last_ingested'):
        # canonical() also repairs state written by the previous script revision,
        # which persisted the API's raw +00:00 form that StartDate rejects.
        try:
            start_date = canonical(state['last_ingested'])
        except (ValueError, TypeError, AttributeError) as e:
            raise RuntimeError(
                f'Unparseable last_ingested in gs://{bucket_name}/{state_key}: '
                f'{state["last_ingested"]!r}. Fix or delete that object; deleting '
                f'restarts collection from the lookback window.'
            ) from e
    else:
        start_date = rfc3339(datetime.now(timezone.utc) - timedelta(hours=lookback_hours))

    seen_ids = set(state.get('seen_ids', []))
    print(f'Collecting events ingested from {start_date}')

    token = get_oauth_token(api_url, client_id, client_secret)
    fresh, cursor, drained = fetch_events(
        api_url, token, start_date, record_size, max_batches, seen_ids
    )

    if not fresh:
        print('No new events. Cursor left unchanged.')
        return

    # Everything that can fail is computed before the upload: a crash between the
    # upload and save_state replays the batch on the next run, so the window where
    # side effects exist without recorded state must stay minimal. The retained ids
    # share the cursor millisecond, because those are exactly the ones the
    # inclusive StartDate will return again.
    retained = {
        event_id(e) for e in fresh
        if event_ingested(e) and canonical(event_ingested(e)) == cursor
    }
    if cursor == start_date:
        # The cursor millisecond did not advance, so ids retained by earlier runs
        # are still on the boundary; dropping them would re-ingest their events.
        retained |= seen_ids

    timestamp = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
    # The random suffix keeps concurrent executions (Pub/Sub delivers at least
    # once) from overwriting each other's object within the same second.
    filename = (
        f'{prefix}beyondtrust-epm-events-{timestamp}-'
        f'{uuid.uuid4().hex[:8]}.ndjson'
    )
    ndjson = '\n'.join(json.dumps(e, ensure_ascii=False) for e in fresh) + '\n'
    bucket.blob(filename).upload_from_string(
        ndjson, content_type='application/x-ndjson'
    )
    print(f'Wrote {len(fresh)} events to gs://{bucket_name}/{filename}')

    save_state(bucket, state_key, {
        'last_ingested': cursor,
        'seen_ids': sorted(retained),
        'updated_at': rfc3339(datetime.now(timezone.utc)),
    })

    if not drained:
        print(
            f'Stopped after {max_batches} batches with more events available. '
            'The cursor advanced, so the next run continues from here.'
        )

def fetch_events(api_url, token, start_date, record_size, max_batches, seen_ids):
    """Read forward from start_date until the API returns a short batch.

    The response envelope has no page count and no next-page token, so a batch
    shorter than record_size is the only documented end-of-data signal.

    StartDate is inclusive and the cursor lands on the newest event's timestamp, so
    every batch re-returns the events sharing it. Deduplication by event id therefore
    has to happen as batches arrive, not only between runs.

    Returns:
    Tuple of (new events, cursor to persist, whether the stream was drained).

    Raises:
    FetchError: on any API or transport failure, so the caller cannot mistake a
    failed fetch for an empty result and advance the cursor.
    """
    headers = {
        'Authorization': f'Bearer {token}',
        'Accept': 'application/json',
    }

    all_events = []
    seen = set(seen_ids)
    cursor = start_date

    for batch in range(1, max_batches + 1):
        query = urlencode({'StartDate': cursor, 'RecordSize': record_size})
        body = request_with_retry(f'{api_url}{EVENTS_PATH}?{query}', headers)

        events = body.get('events') or []
        new = [e for e in events if event_id(e) not in seen]
        seen.update(event_id(e) for e in events if event_id(e))
        all_events.extend(new)
        print(f'Batch {batch}: {len(events)} events, {len(new)} new')

        # Termination is judged on what the API returned, not on what survived
        # deduplication: a batch can be full and still be entirely duplicates.
        # The newest stamp in the batch, not events[-1]: ordering is a client-side
        # convention, not a documented guarantee, and a misordered tail would
        # regress the cursor and re-ingest events already written.
        stamps = [canonical(s) for s in (event_ingested(e) for e in events) if s]

        if len(events) < record_size:
            # Short batch: the stream is drained.
            if events and not stamps:
                print(
                    'Warning: no event in the final batch carries event.ingested; '
                    'cursor left unchanged.'
                )
            if stamps:
                cursor = max(max(stamps), cursor)
            return all_events, cursor, True

        if not stamps:
            raise FetchError(
                'No event in a full batch carries event.ingested, '
                'so the cursor cannot advance'
            )
        batch_max = max(stamps)
        if batch_max <= cursor:
            # A full batch fits inside one ingestion millisecond. Stepping past it is the only
            # way to make progress; holding the cursor here stalls collection forever.
            # Events beyond record_size at that exact millisecond are unreachable, which
            # needs more than 1000 events in one millisecond.
            print(
                f'Warning: a full batch shares ingestion timestamp {batch_max}; '
                'stepping past it. Events beyond RecordSize at that timestamp are skipped.'
            )
            cursor = next_millisecond(cursor)
        else:
            cursor = batch_max

    return all_events, cursor, False

def request_with_retry(url, headers, attempts=4):
    """GET with backoff on 429 and 5xx.

    The documented rate limit is 1000 requests per 100 seconds.
    """
    backoff = 1.0
    for attempt in range(1, attempts + 1):
        try:
            response = http.request('GET', url, headers=headers)
        except Exception as e:
            raise FetchError(f'Request to {url} failed: {e}') from e

        if response.status in (429, 500, 502, 503, 504) and attempt < attempts:
            retry_after = response.headers.get('Retry-After')
            try:
                delay = int(retry_after) if retry_after else backoff
            except (TypeError, ValueError):
                delay = backoff
            # Retry-After is server-controlled input: bound it so a bogus value
            # cannot sleep past the function timeout or crash time.sleep.
            delay = min(max(delay, 1.0), 60.0)
            print(f'HTTP {response.status}. Retrying in {delay}s...')
            time.sleep(delay)
            backoff = min(backoff * 2, 30.0)
            continue

        if response.status != 200:
            raise FetchError(
                f'Request failed: {response.status} {response.data.decode("utf-8")}'
            )

        text = response.data.decode('utf-8')
        # BeyondTrust's sample folds "Owner" into "owner" before parsing, but that
        # workaround exists for PowerShell's case-insensitive ConvertFrom-Json.
        # Python parses case-sensitively, and the SecOps parser maps file.Owner.*
        # and file.owner to different UDM fields, so both keys must survive.
        try:
            return json.loads(text)
        except json.JSONDecodeError as e:
            raise FetchError(f'Malformed JSON response: {e}') from e

    raise FetchError(f'Giving up on {url} after {attempts} attempts')

def get_oauth_token(api_url, client_id, client_secret):
    """Get an access token using the OAuth client credentials flow.

    The token is valid for one hour.
    """
    body = urlencode({
        'grant_type': 'client_credentials',
        'client_id': client_id,
        'client_secret': client_secret,
        'scope': OAUTH_SCOPE,
    })
    response = http.request(
        'POST',
        f'{api_url}{TOKEN_PATH}',
        body=body,
        headers={'Content-Type': 'application/x-www-form-urlencoded'},
    )
    if response.status != 200:
        raise FetchError(
            f'Token request failed: {response.status} '
            f'{response.data.decode("utf-8")}'
        )
    return json.loads(response.data.decode('utf-8'))['access_token']

def load_state(bucket, key):
    """Read the collector state from Cloud Storage.

    Only a missing object is a cold start. Any other error is raised: swallowing it
    would reset collection to the lookback window and re-ingest that period.
    """
    blob = bucket.blob(key)
    try:
        return json.loads(blob.download_as_text())
    except NotFound:
        print('No state file found. Starting from the lookback window.')
        return {}

def save_state(bucket, key, state):
    """Write the collector state to Cloud Storage.

    Failures are raised, not logged. A run that cannot record its cursor must fail,
    otherwise the next run repeats the same window.
    """
    bucket.blob(key).upload_from_string(
        json.dumps(state, indent=2), content_type='application/json'
    )
    print(f"Saved state: last_ingested={state.get('last_ingested')}")

    • 第二個檔案：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. 前往 Google Cloud 控制台的「Cloud Scheduler」。
  2. 點選「建立工作」。

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

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

  4. 點選「建立」。

排程頻率選項

根據記錄檔量和延遲時間要求選擇頻率：

頻率 Cron 運算式 用途
每 5 分鐘 */5 * * * * 高容量、低延遲
每 15 分鐘 */15 * * * * 中等
每小時 0 * * * * 標準 (建議)
每 6 小時 0 */6 * * * 少量、批次處理
每天 0 0 * * * 歷來資料集合

測試排程器工作

  1. 在 Cloud Scheduler 控制台中找出您的工作。
  2. 按一下「強制執行」即可手動觸發。
  3. 等待幾秒鐘，然後依序前往「Cloud Run」>「Services」>「beyondtrust-epm-collector」>「Logs」。
  4. 確認函式是否執行成功。
  5. 檢查 Cloud Storage bucket，確認記錄已寫入。

擷取 Google SecOps 服務帳戶

Google SecOps 會使用專屬服務帳戶，從 Cloud Storage bucket 讀取資料。您必須授予這個服務帳戶值區存取權。

取得服務帳戶電子郵件地址

  1. 依序前往「SIEM 設定」>「動態饋給」。
  2. 按一下「新增動態消息」。
  3. 按一下「設定單一動態饋給」。
  4. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 BeyondTrust EPM logs)。
  5. 選取「Google Cloud Storage V2」做為「來源類型」。
  6. 選取「BeyondTrust Endpoint Privilege Management」做為「記錄類型」。

  7. 按一下「取得服務帳戶」。系統會顯示專屬服務帳戶電子郵件地址，例如：

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

  8. 複製這個電子郵件地址，以便在下一步中使用。

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

Google SecOps 服務帳戶需要 Cloud Storage bucket 的兩個角色：Storage 物件檢視者，用於讀取記錄物件；以及 bucket 層級角色，用於讀取 bucket 中繼資料。

  1. 依序前往「Cloud Storage」>「Buckets」。
  2. 按一下 bucket 名稱。
  3. 前往「權限」分頁標籤。
  4. 按一下「授予存取權」。
  5. 提供下列設定詳細資料：
    • 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址。
    • 指派角色：選取下列兩個角色：
    • Storage 物件檢視者：讀取記錄檔物件。
    • Storage 舊版值區讀取者：讀取值區中繼資料。如果您選取「刪除已轉移的檔案和空目錄」刪除選項，請改為選取「Storage Legacy Bucket Writer」，這也會授予刪除權限。

  6. 按一下 [儲存]。

在 Google SecOps 中設定動態饋給，擷取 BeyondTrust EPM 記錄

  1. 依序前往「SIEM 設定」>「動態饋給」。
  2. 按一下「新增動態消息」。
  3. 按一下「設定單一動態饋給」。
  4. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 BeyondTrust EPM logs)。
  5. 選取「Google Cloud Storage V2」做為「來源類型」。
  6. 選取「BeyondTrust Endpoint Privilege Management」做為「記錄類型」。
  7. 點選「下一步」。

  8. 指定下列輸入參數的值：

    • 儲存空間 bucket URL：輸入 Cloud Storage bucket URI，並加上前置路徑：

      gs://beyondtrust-epm-logs/beyondtrust-epm/

      • 取代：

        • beyondtrust-epm-logs：Cloud Storage bucket 名稱。
        • beyondtrust-epm/：儲存記錄的選用前置字元/資料夾路徑 (如為根目錄，請留空)。

      • 範例：

        • 根 bucket：gs://beyondtrust-epm-logs/
        • 含前置字元：gs://beyondtrust-epm-logs/beyondtrust-epm/

    • 來源刪除選項：根據偏好設定選取刪除選項：

      • 一律不刪除檔案：一律不刪除來源中的檔案。建議您基於測試目的調整價格。

      • 刪除已轉移的檔案和空白目錄：成功擷取檔案後，從來源刪除檔案和空白目錄。

    • 檔案存在時間上限：包含在過去天數內修改的檔案。預設值為 180 天。

    • 資產命名空間：資產命名空間。

    • 擷取標籤：要套用至這個動態饋給事件的標籤。

  9. 點選「下一步」。

  10. 在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。

UDM 對應表

記錄欄位 UDM 對應 邏輯
DomainIdentifier_label additional.fields 已合併
WatsonEventType_label additional.fields 已合併
action_label additional.fields 已合併
activity_id_label additional.fields 已合併
app_id_label additional.fields 已合併
app_insights_instrumentation_key_label additional.fields 已合併
asimov_instrumentation_key_label additional.fields 已合併
authorization_request_control_authorization_label additional.fields 已合併
bundle_name_label additional.fields 已合併
changedBy_label additional.fields 已合併
code_signature_subject_name_label additional.fields 已合併
collector_api_key_label additional.fields 已合併
command_label additional.fields 已合併
configuration_application_group_description_label additional.fields 已合併
configuration_application_group_identifier_label additional.fields 已合併
configuration_application_group_name_label additional.fields 已合併
configuration_application_identifier_label additional.fields 已合併
configuration_application_type_label additional.fields 已合併
configuration_identifier_label additional.fields 已合併
configuration_message_identifier_label additional.fields 已合併
configuration_message_name_label additional.fields 已合併
configuration_message_type_label additional.fields 已合併
configuration_name_label additional.fields 已合併
configuration_revision_number_label additional.fields 已合併
configuration_rule_identifier_label additional.fields 已合併
configuration_rule_on_demand_label additional.fields 已合併
configuration_rule_script_outcome_rule_affected_label additional.fields 已合併
configuration_token_identifier_label additional.fields 已合併
configuration_token_name_label additional.fields 已合併
configuration_workstyle_description_label additional.fields 已合併
configuration_workstyle_identifier_label additional.fields 已合併
configuration_workstyle_name_label additional.fields 已合併
content_length_label additional.fields 已合併
content_type_label additional.fields 已合併
domainNetBIOSName_label additional.fields 已合併
domain_label additional.fields 已合併
entity_label additional.fields 已合併
entity_name_label additional.fields 已合併
event_action_label additional.fields 已合併
exitstatus_label additional.fields 已合併
file_hash_md5_label additional.fields 已合併
file_hash_sha1_label additional.fields 已合併
file_hash_sha256_label additional.fields 已合併
file_version_label additional.fields 已合併
gid_label additional.fields 已合併
group_data_id_label additional.fields 已合併
group_id_label additional.fields 已合併
group_label additional.fields 已合併
handle_label additional.fields 已合併
host_name_label additional.fields 已合併
host_uptime_labels additional.fields 已合併
http_host_label additional.fields 已合併
id_label additional.fields 已合併
iolog_label additional.fields 已合併
is_opted_in_label additional.fields 已合併
linenum_label additional.fields 已合併
local_identifier_label additional.fields 已合併
locale_label additional.fields 已合併
master_utcoffset_label additional.fields 已合併
masterlocale_label additional.fields 已合併
owner_identifier_label additional.fields 已合併
parent_entity_id_label additional.fields 已合併
parent_process_exec_label additional.fields 已合併
parent_process_label additional.fields 已合併
pbmasterdnodename_label additional.fields 已合併
pipeName_label additional.fields 已合併
process_entity_id_label additional.fields 已合併
process_hash_label additional.fields 已合併
process_name_label additional.fields 已合併
process_parent_name_label additional.fields 已合併
process_start_time_label additional.fields 已合併
processexe_label additional.fields 已合併
product_label additional.fields 已合併
product_type_label additional.fields 已合併
product_version_label additional.fields 已合併
requestuser_label additional.fields 已合併
runargv_label additional.fields 已合併
runcwd_label additional.fields 已合併
runeffectivegroup_label additional.fields 已合併
runeffectiveuser_label additional.fields 已合併
runhost_label additional.fields 已合併
schema_version_label additional.fields 已合併
sku_name_label additional.fields 已合併
telemetry_level_label additional.fields 已合併
tenant_id_label additional.fields 已合併
type_label additional.fields 已合併
uid_label additional.fields 已合併
user_id_label additional.fields 已合併
user_name_label additional.fields 已合併
vs_exe_version_label additional.fields 已合併
inter_host intermediary.hostname 直接對應
Processes.description metadata.description 直接對應
details metadata.description 直接對應
event_data.reason metadata.description 直接對應
file.pe.description metadata.description 直接對應
created metadata.event_timestamp 已剖析為 ISO8601
datetime metadata.event_timestamp 已剖析為 MMM dd HH:mm:ss
has_principal metadata.event_type 已對應：true → STATUS_UPDATE
has_user metadata.event_type 已對應：true → USER_UNCATEGORIZED
parent_working_directory_label metadata.ingestion_labels 已合併
working_directory_label metadata.ingestion_labels 已合併
auditType metadata.product_event_type 直接對應
event_datas.ActionId metadata.product_log_id 直接對應
labels.related_item_id metadata.product_log_id 直接對應
uniqueid metadata.product_log_id 直接對應
masterdversion metadata.product_version 直接對應
headers.http_version network.application_protocol_version 直接對應
headers.request_method network.http.method 直接對應
host.os.platform principal.administrative_domain 直接對應
Processes.process principal.application 直接對應
agent_ephemeral_id_label principal.asset.attribute.labels 已合併
agent_id_label principal.asset.attribute.labels 已合併
agent_version_label principal.asset.attribute.labels 已合併
ecs_version_label principal.asset.attribute.labels 已合併
_hardware principal.asset.hardware 已合併
host.hostname principal.asset.hostname 直接對應
submithost principal.asset.hostname 直接對應
ip_address principal.asset.ip 已合併
masterhostip principal.asset.ip 已合併
submithostip principal.asset.ip 已合併
file.path principal.file.full_path 直接對應
lineinfile principal.file.full_path 直接對應
host.hostname principal.hostname 直接對應
submithost principal.hostname 直接對應
ip_address principal.ip 已合併
masterhostip principal.ip 已合併
submithostip principal.ip 已合併
mac principal.mac 已合併
host.os.name principal.platform_version 直接對應
host.os.version principal.platform_version 直接對應
process.command_line principal.process.command_line 直接對應
runcommand principal.process.command_line 直接對應
process.executable principal.process.file.full_path 直接對應
cmd principal.process.parent_process.command_line 直接對應
Processes.process_path principal.process.parent_process.file.full_path 直接對應
process.parent.executable principal.process.parent_process.file.full_path 直接對應
Processes.parent_process_id principal.process.parent_process.pid 直接對應
Processes.process_id principal.process.pid 直接對應
logpid principal.process.pid 直接對應
file.Owner.DomainName principal.user.company_name 直接對應
file.Owner.Name principal.user.user_display_name 直接對應
runuser principal.user.user_display_name 直接對應
userName principal.user.user_display_name 直接對應
Processes.user principal.user.userid 直接對應
userId principal.user.userid 直接對應
userid principal.user.userid 直接對應
Processes.user_id principal.user.windows_sid 直接對應
EPMWinMac.Configuration.Rule.Action security_result.action 已合併
security_result_action security_result.action 已合併
event_data.outcome security_result.category_details 已合併
host.os.version security_result.category_details 已合併
EPMWinMac.Configuration.Application.Description security_result.description 直接對應
EPMWinMac.Configuration.Message.Description security_result.description 直接對應
host.os.type src.administrative_domain 直接對應
file.name src.file.names 已合併
host.os.full src.platform_version 直接對應
host.os.family target.administrative_domain 直接對應
Processes.dest target.asset.hostname 直接對應
file.extension target.file.mime_type 直接對應
Processes.dest target.hostname 直接對應
host.domain target.hostname 直接對應
file_DriveType_label target.resource.attribute.labels 已合併
file_drive_letter_label target.resource.attribute.labels 已合併
owner_label target.resource.attribute.labels 已合併
不適用 metadata.event_type 常數：USER_UNCATEGORIZED
不適用 metadata.product_name 常數：Beyondtrust Privilege Management
不適用 metadata.vendor_name 常數：Beyondtrust Privilege Management
不適用 network.application_protocol 常數：HTTP
不適用 principal.platform 常數：MAC

變更記錄

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

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