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 버킷을 만들고 관리할 수 있는 권한
  • Cloud Storage 버킷의 Identity and Access Management (IAM) 정책을 관리할 수 있는 권한
  • Cloud Run 함수 서비스, Pub/Sub 주제, Cloud Scheduler 작업을 만들 수 있는 권한
  • BeyondTrust Endpoint Privilege Management 테넌트 또는 API에 대한 권한 있는 액세스

Cloud Storage 버킷 만들기

  1. Google Cloud 콘솔로 이동합니다.
  2. 프로젝트를 선택하거나 새 프로젝트를 만듭니다.
  3. 탐색 메뉴에서 Cloud Storage> 버킷으로 이동합니다.
  4. 버킷 만들기를 클릭합니다.

  5. 다음 구성 세부정보를 제공합니다.

    설정 값
    버킷 이름 지정 전역적으로 고유한 이름 (예: beyondtrust-epm-logs)을 입력합니다.
    위치 유형 필요에 따라 선택 (리전, 이중 리전, 멀티 리전)
    위치 위치를 선택합니다 (예: us-central1).
    스토리지 클래스 Standard (자주 액세스하는 로그에 권장)
    액세스 제어 균일 (권장)
    보호 조치 선택사항: 객체 버전 관리 또는 보관 정책 사용 설정

  6. 만들기를 클릭합니다.

BeyondTrust EPM API 사용자 인증 정보 수집

  1. 관리자로 BeyondTrust Privilege Management 웹 콘솔에 로그인합니다.
  2. 구성 > 설정 > API 설정으로 이동합니다.
  3. API 계정 만들기를 클릭합니다.
  4. 다음 구성 세부정보를 제공합니다.
    • 이름: Google SecOps Collector를 입력합니다.
    • API 액세스: 보고 (읽기 전용)을 사용 설정합니다. 이 수집기에서 사용하는 /management-api/v3/Events/FromStartDate 엔드포인트는 보고 권한 카테고리에 속합니다. 웹 콘솔 활동 감사도 수집하려는 경우에만 감사 (읽기 전용)도 사용 설정하세요.
  5. 클라이언트 ID와 클라이언트 보안 비밀번호를 복사하여 저장합니다.

  6. API 설정 페이지 상단에 표시된 API 기본 URL을 복사합니다. 일반적으로 https://<your-tenant>-services.pm.beyondtrustcloud.com입니다. 이를 BPT_API_URL로 사용합니다.

Cloud Run 함수의 서비스 계정 만들기

Cloud Run 함수에는 Cloud Storage 버킷에 쓸 수 있고 Pub/Sub에 의해 호출될 수 있는 권한이 있는 서비스 계정이 필요합니다.

서비스 계정 만들기

  1. Google Cloud 콘솔에서 IAM 및 관리자 > 서비스 계정으로 이동합니다.
  2. 서비스 계정 만들기를 클릭합니다.
  3. 다음 구성 세부정보를 제공합니다.
    • 서비스 계정 이름: beyondtrust-epm-collector-sa을 입력합니다.
    • 서비스 계정 설명: Service account for Cloud Run function to collect BeyondTrust EPM logs을 입력합니다.
  4. 만들고 계속하기를 클릭합니다.
  5. 이 서비스 계정에 프로젝트에 대한 액세스 권한 부여 섹션에서 다음 역할을 추가합니다.
    1. 역할 선택을 클릭합니다.
    2. 스토리지 객체 관리자를 검색하여 선택합니다.
    3. + 다른 역할 추가를 클릭합니다.
    4. Cloud Run 호출자를 검색하여 선택합니다.
    5. + 다른 역할 추가를 클릭합니다.
    6. Cloud Functions 호출자를 검색하여 선택합니다.
  6. 계속을 클릭합니다.
  7. 완료를 클릭합니다.

이러한 역할은 다음 작업에 필요합니다.

  • 스토리지 객체 관리자: Cloud Storage 버킷에 로그를 쓰고 상태 파일을 관리합니다.
  • Cloud Run 호출자: Pub/Sub가 함수를 호출할 수 있도록 허용
  • Cloud Functions 호출자: 함수 호출을 허용합니다.

Cloud Storage 버킷에 대한 IAM 권한 부여

서비스 계정에 Cloud Storage 버킷에 대한 쓰기 권한을 부여합니다.

  1. Cloud Storage> 버킷으로 이동합니다.
  2. 버킷 이름을 클릭합니다.
  3. 권한 탭으로 이동합니다.
  4. 액세스 권한 부여를 클릭합니다.
  5. 다음 구성 세부정보를 제공합니다.
    • 주 구성원 추가: 서비스 계정 이메일을 입력합니다. 예를 들면 beyondtrust-epm-collector-sa@PROJECT_ID.iam.gserviceaccount.com입니다.
    • 역할 할당: 스토리지 객체 관리자를 선택합니다.
  6. 저장을 클릭합니다.

Pub/Sub 주제 만들기

Cloud Scheduler가 게시하고 Cloud Run 함수가 구독할 Pub/Sub 주제를 만듭니다.

  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. 서비스 만들기를 클릭합니다.
  3. 함수를 선택합니다. 인라인 편집기를 사용하여 함수를 만듭니다.

  4. 구성 섹션에서 다음 구성 세부정보를 입력합니다.

    설정 값
    서비스 이름 beyondtrust-epm-collector
    리전 Cloud Storage 버킷과 일치하는 리전을 선택합니다 (예: us-central1).
    런타임 Python 3.12 이상 선택

  5. 트리거 (선택사항) 섹션에서 다음을 수행합니다.

    1. + 트리거 추가를 클릭합니다.
    2. Cloud Pub/Sub을 선택합니다.
    3. Cloud Pub/Sub 주제 선택에서 beyondtrust-epm-trigger 주제를 선택합니다.
    4. 저장을 클릭합니다.

  6. 인증 섹션에서 다음을 구성합니다.

    1. 인증 필요를 선택합니다.
    2. ID 및 액세스 관리 (IAM)를 확인합니다.

  7. 컨테이너, 네트워킹, 보안으로 이동하여 펼칩니다.

  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을 허용하며, 함수는 더 높은 값을 클램프합니다.
    • MAX_BATCHES: 실행당 요청에 대한 안전 한계입니다. 실행이 이 지점에 도달하면 커서는 작성된 내용으로 계속 이동하므로 다음 실행은 작업을 반복하지 않고 이 지점에서 계속됩니다.
    • LOOKBACK_HOURS: 첫 번째 실행이 도달하는 시점입니다. BeyondTrust는 수동 이벤트 (코드 106, 107, 603, 706)를 30일 동안, 기타 모든 이벤트를 90일 동안 보관하므로 더 긴 전환 확인 기간을 설정해도 그보다 오래된 데이터는 반환되지 않습니다.
    • LOOKBACK_HOURS: 커서가 존재하기 전 첫 실행에서 도달할 과거의 범위입니다.

  10. 변수 및 보안 비밀 탭에서 요청으로 이동합니다.

    • 요청 제한 시간: 600초 (10분)를 입력합니다.

  11. 컨테이너의 설정 탭으로 이동합니다.

    • 리소스 섹션에서 다음을 수행합니다.
      • 메모리: 512MiB 이상을 선택합니다.
      • CPU: 1을 선택합니다.
    • 완료를 클릭합니다.

  12. 실행 환경으로 스크롤합니다.

    • 기본값 (권장)을 선택합니다.

  13. 버전 확장 섹션에서 다음을 수행합니다.

    • 최소 인스턴스 수: 0를 입력합니다.
    • 최대 인스턴스 수: 100을 입력합니다 (또는 예상 부하에 따라 조정).

  14. 만들기를 클릭합니다.

  15. 서비스가 생성될 때까지 기다립니다 (1~2분).

  16. 서비스가 생성되면 인라인 코드 편집기가 자동으로 열립니다.

함수 코드 추가

  1. 함수 진입점에 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 * * * * (매시간 정각)
    시간대 시간대 선택 (UTC 권장)
    타겟 유형 Pub/Sub
    주제 beyondtrust-epm-trigger 주제를 선택합니다.
    메일 본문 {} (빈 JSON 객체)

  4. 만들기를 클릭합니다.

일정 빈도 옵션

로그 볼륨 및 지연 시간 요구사항에 따라 빈도를 선택합니다.

빈도 크론 표현식 사용 사례
5분마다 */5 * * * * 대용량, 저지연
15분마다 */15 * * * * 검색량 보통
1시간마다 0 * * * * 표준 (권장)
6시간마다 0 */6 * * * 적은 볼륨, 일괄 처리
매일 0 0 * * * 이전 데이터 수집

스케줄러 작업 테스트

  1. Cloud Scheduler 콘솔에서 작업을 찾습니다.
  2. 강제 실행을 클릭하여 수동으로 트리거합니다.
  3. 몇 초간 기다린 후 Cloud Run > 서비스 > beyondtrust-epm-collector > 로그로 이동합니다.
  4. 함수가 성공적으로 실행되었는지 확인합니다.
  5. Cloud Storage 버킷을 확인하여 로그가 작성되었는지 확인합니다.

Google SecOps 서비스 계정 가져오기

Google SecOps는 고유한 서비스 계정을 사용하여 Cloud Storage 버킷에서 데이터를 읽습니다. 이 서비스 계정에 버킷 액세스 권한을 부여해야 합니다.

서비스 계정 이메일 가져오기

  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. 다음 단계에서 사용할 수 있도록 이 이메일 주소를 복사합니다.

Google SecOps 서비스 계정에 IAM 권한 부여

Google SecOps 서비스 계정에는 Cloud Storage 버킷에 대한 두 가지 역할이 필요합니다. 로그 객체를 읽기 위한 스토리지 객체 뷰어와 버킷 메타데이터를 읽기 위한 버킷 수준 역할입니다.

  1. Cloud Storage> 버킷으로 이동합니다.
  2. 버킷 이름을 클릭합니다.
  3. 권한 탭으로 이동합니다.
  4. 액세스 권한 부여를 클릭합니다.
  5. 다음 구성 세부정보를 제공합니다.
    • 주 구성원 추가: Google SecOps 서비스 계정 이메일을 붙여넣습니다.
    • 역할 할당: 다음 두 가지를 모두 선택합니다.
    • 스토리지 객체 뷰어: 로그 객체를 읽습니다.
    • 스토리지 기존 버킷 리더: 버킷 메타데이터를 읽습니다. 전송된 파일 및 빈 디렉터리 삭제 삭제 옵션을 선택한 경우 삭제 권한도 부여하는 스토리지 기존 버킷 작성자를 대신 선택합니다.

  6. 저장을 클릭합니다.

BeyondTrust EPM 로그를 수집하도록 Google SecOps에서 피드 구성

  1. SIEM 설정 > 피드로 이동합니다.
  2. 새 피드 추가를 클릭합니다.
  3. 단일 피드 구성을 클릭합니다.
  4. 피드 이름 필드에 피드 이름을 입력합니다(예: BeyondTrust EPM logs).
  5. 소스 유형으로 Google Cloud Storage V2를 선택합니다.
  6. 로그 유형으로 BeyondTrust Endpoint Privilege Management를 선택합니다.
  7. 다음을 클릭합니다.

  8. 다음 입력 매개변수의 값을 지정합니다.

    • 스토리지 버킷 URL: 다음 접두사 경로를 사용하여 Cloud Storage 버킷 URI를 입력합니다.

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

      • 다음과 같이 바꿉니다.

        • beyondtrust-epm-logs: Cloud Storage 버킷 이름
        • beyondtrust-epm/: 로그가 저장되는 선택적 접두사/폴더 경로입니다 (루트의 경우 비워 둠).

      • 예:

        • 루트 버킷: gs://beyondtrust-epm-logs/
        • 접두사 사용: gs://beyondtrust-epm-logs/beyondtrust-epm/

    • 소스 삭제 옵션: 환경설정에 따라 삭제 옵션을 선택합니다.

      • 파일 삭제 안함: 소스에서 파일을 삭제하지 않습니다. 테스트 목적으로 권장됩니다.

      • 전송된 파일 및 빈 디렉터리 삭제: 가져오기가 완료되면 소스에서 파일과 빈 디렉터리를 삭제합니다.

    • 최대 파일 기간: 지난 일수 동안 수정된 파일을 포함합니다. 기본값은 180일입니다.

    • 애셋 네임스페이스: 애셋 네임스페이스입니다.

    • 수집 라벨: 이 피드의 이벤트에 적용할 라벨입니다.

  9. 다음을 클릭합니다.

  10. 확정 화면에서 새 피드 구성을 검토한 다음 제출을 클릭합니다.

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 전문가에게 문의하여 답변을 받으세요.