Proofpoint Secure Email Relay 로그 수집

다음에서 지원:

이 문서에서는 Cloud Storage V2를 사용하여 Proofpoint Secure Email Relay 로그를 Google Security Operations에 수집하는 방법을 설명합니다.

Proofpoint Secure Email Relay (SER)는 거래 및 알림 메시지와 같은 애플리케이션 생성 메일을 위한 관리형 릴레이입니다. SER은 SER 구성 API 중 하나인 Reporting API를 통해 해당 트래픽을 보고합니다. API는 라이선스가 부여된 처리량 위치, 일일 볼륨 및 메시지 수, 각 중계 사용자 카운터와 같은 집계 사용량 수치를 반환합니다. 파서는 이러한 수치를 통합 데이터 모델 (UDM)에 매핑하여 중계 사용자 ID와 함께 라벨로 기록합니다.

시작하기 전에

다음 기본 요건이 충족되었는지 확인합니다.

  • Google SecOps 인스턴스
  • Cloud Storage API가 사용 설정된 Google Cloud 프로젝트
  • Cloud Storage 버킷을 만들고 관리할 수 있는 권한
  • Cloud Storage 버킷의 Identity and Access Management (IAM) 정책을 관리할 수 있는 권한
  • Cloud Run 서비스, Pub/Sub 주제, Cloud Scheduler 작업을 만들 수 있는 권한
  • API 키를 생성할 권한을 포함한 Proofpoint Secure Email Relay 포털에 대한 관리자 액세스 권한

Cloud Storage 버킷 만들기

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

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

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

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

Proofpoint SER API 사용자 인증 정보 수집

Reporting API에 대한 액세스 권한은 Proofpoint API 키 관리 서비스에서 발급된 API 키를 통해 부여됩니다. 키는 SER 콘솔 내에서 생성되지 않으므로 SER 탐색 항목에 표시되지 않습니다.

API 키 만들기

  1. 관리자 사용자 인증 정보로 Proofpoint Secure Email Relay 포털에 로그인합니다.
  2. 왼쪽 상단에서 앱 전환 메뉴를 엽니다.
  3. 서비스 > API 키 관리로 이동합니다.
  4. 키 만들기를 클릭합니다.
  5. 제품으로 Secure Email Relay를 선택합니다.

  6. 다음 값을 복사하여 안전하게 저장합니다.

    • 키: client_id로 사용됩니다.
    • 보안 비밀: client_secret로 사용됩니다.

권한 확인

API 키 관리 서비스는 여러 Proofpoint 제품에서 공유되므로 SER 탐색 메뉴가 아닌 앱 전환 메뉴를 통해 액세스합니다. 보고 API를 비롯한 SER 구성 API에 액세스하려면 키를 만들 때 제품으로 Secure Email Relay를 선택해야 합니다.

앱 전환 메뉴에 API 키 관리가 표시되지 않으면 계정에 API 관리 권한이 없는 것입니다. Proofpoint 관리자 또는 계정팀에 문의하여 권한을 부여받으세요.

API 액세스 테스트

  • Reporting API는 OAuth 2.0 클라이언트 사용자 인증 정보 부여를 사용합니다. 키와 보안 비밀을 전달자 토큰으로 교환한 다음 해당 토큰으로 API를 호출합니다.

    # Replace with the values you copied
CLIENT_ID="<your-key>"
CLIENT_SECRET="<your-secret>"

# 1. Exchange the key and secret for an access token
ACCESS_TOKEN=$(curl -s -X POST "https://auth.proofpoint.com/v1/token" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "grant_type=client_credentials" \
  -d "client_id=${CLIENT_ID}" \
  -d "client_secret=${CLIENT_SECRET}" \
  | python3 -c "import json,sys; print(json.load(sys.stdin)['access_token'])")

# 2. Call the Reporting API with the token
curl -v -H "Authorization: Bearer ${ACCESS_TOKEN}" \
  "https://reporting.ser.proofpoint.com/v1/usage/overview"

호출이 성공하면 data 멤버에 throughputLimit, licenseStartDate, licenseEndDate, 처리량 평균이 포함된 JSON 객체가 반환됩니다.

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

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

서비스 계정 만들기

  1. GCP 콘솔에서 IAM 및 관리자 > 서비스 계정으로 이동합니다.
  2. 서비스 계정 만들기를 클릭합니다.
  3. 다음 구성 세부정보를 제공합니다.
    • 서비스 계정 이름: proofpoint-ser-collector-sa을 입력합니다.
    • 서비스 계정 설명: Service account for Cloud Run function to collect Proofpoint Secure Email Relay 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. 버킷 이름 (예: proofpoint-ser-logs)을 클릭합니다.
  3. 권한 탭으로 이동합니다.
  4. 액세스 권한 부여를 클릭합니다.
  5. 다음 구성 세부정보를 제공합니다.
    • 주 구성원 추가: 서비스 계정 이메일 (예: proofpoint-ser-collector-sa@PROJECT_ID.iam.gserviceaccount.com)을 입력합니다.
    • 역할 할당: 스토리지 객체 관리자 선택
  6. 저장을 클릭합니다.

Pub/Sub 주제 만들기

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

  1. GCP Console에서 Pub/Sub > 주제로 이동합니다.
  2. 주제 만들기를 클릭합니다.
  3. 다음 구성 세부정보를 제공합니다.
    • 주제 ID: proofpoint-ser-trigger를 입력합니다.
    • 다른 설정은 기본값으로 둡니다.
  4. 만들기를 클릭합니다.

로그를 수집하는 Cloud Run 함수 만들기

Cloud Run 함수는 Cloud Scheduler의 Pub/Sub 메시지에 의해 트리거되어 SER Reporting API에서 사용 보고서를 가져와 Cloud Storage에 씁니다.

  1. GCP 콘솔에서 Cloud Run으로 이동합니다.
  2. 서비스 만들기를 클릭합니다.
  3. 함수를 선택합니다 (인라인 편집기를 사용하여 함수 만들기).

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

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

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

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

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

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

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

  8. 보안 탭으로 이동합니다.

    • 서비스 계정: 서비스 계정을 선택합니다 (proofpoint-ser-collector-sa).

  9. 컨테이너 탭으로 이동합니다.

    1. 변수 및 보안 비밀을 클릭합니다.
    2. 각 환경 변수에 대해 + 변수 추가를 클릭합니다.
    변수 이름 예시 값 설명
    GCS_BUCKET proofpoint-ser-logs Cloud Storage 버킷 이름
    GCS_PREFIX ser-logs 로그 파일의 접두사
    STATE_KEY ser-logs-state.json 로그 접두사 외부의 상태 경로
    TOKEN_URL https://auth.proofpoint.com/v1/token OAuth 2.0 토큰 엔드포인트
    API_BASE https://reporting.ser.proofpoint.com SER Reporting API 기본 URL
    CLIENT_ID your-key API 키 관리의 키
    CLIENT_SECRET your-secret API 키 관리의 Secret
    PAGE_SIZE 50 페이지당 요청된 중계 사용자
    MAX_PAGES 200 릴레이 사용자 페이지로 나누기의 페이지 상한
    REPORT_DAY_OFFSET 1 이 날짜로부터 이만큼 전의 날짜에 대한 보고서
    MAX_BACKFILL_DAYS 7 단일 실행으로 백필되는 일수
    SEEN_RETENTION_DAYS 7 상태에 보관된 중복 삭제 키의 일수

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

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

  11. 설정 탭으로 이동합니다.

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

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

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

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

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

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

함수 코드 추가

  1. 진입점 필드에 main을 입력합니다.
  2. 인라인 코드 편집기에서 다음 두 파일을 만듭니다.

  • 첫 번째 파일 - main.py:

    import functions_framework
from google.cloud import storage
from google.cloud.exceptions import NotFound
import hashlib
import json
import os
import time
import urllib.parse
import urllib3
from datetime import date, datetime, timezone, timedelta

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

# Initialize Storage client
storage_client = storage.Client()

# Environment variables
GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'ser-logs')
# STATE_KEY must stay OUTSIDE GCS_PREFIX. The feed ingests every object under
# its bucket URI and, with a deletion option selected, deletes what it
# transferred. A state file inside the prefix would be ingested as log data and
# then deleted, resetting collection and re-ingesting duplicates.
STATE_KEY = os.environ.get('STATE_KEY', 'ser-logs-state.json')
TOKEN_URL = os.environ.get('TOKEN_URL', 'https://auth.proofpoint.com/v1/token')
API_BASE = os.environ.get('API_BASE', 'https://reporting.ser.proofpoint.com')
CLIENT_ID = os.environ.get('CLIENT_ID')
CLIENT_SECRET = os.environ.get('CLIENT_SECRET')
PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '50'))
MAX_PAGES = int(os.environ.get('MAX_PAGES', '200'))
# The Reporting API filters on whole calendar days, so the collector reports on
# a day that has already finished rather than on a partial one.
REPORT_DAY_OFFSET = int(os.environ.get('REPORT_DAY_OFFSET', '1'))
MAX_BACKFILL_DAYS = int(os.environ.get('MAX_BACKFILL_DAYS', '7'))
SEEN_RETENTION_DAYS = int(os.environ.get('SEEN_RETENTION_DAYS', '7'))

MAX_RATE_LIMIT_RETRIES = 5

class FetchError(Exception):
    """Raised when a Proofpoint SER API call fails.

    Collection must fail loudly. Returning an empty result on an API error is
    indistinguishable from a day with no traffic, and would let the run record
    a day it never actually read.
    """

def get_access_token():
    """Exchange the API key and secret for an OAuth 2.0 bearer token.

    The SER configuration APIs use the client credentials grant with the
    credentials in the form body. The returned token is sent to each API
    host as an Authorization: Bearer header.
    """
    body = urllib.parse.urlencode({
        'grant_type': 'client_credentials',
        'client_id': CLIENT_ID,
        'client_secret': CLIENT_SECRET,
    })
    headers = {
        'Content-Type': 'application/x-www-form-urlencoded',
        'Accept': 'application/json',
    }

    try:
        response = http.request('POST', TOKEN_URL, body=body, headers=headers)
    except Exception as e:
        raise FetchError(f'Token request to {TOKEN_URL} failed: {e}') from e

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

    try:
        token = json.loads(response.data.decode('utf-8')).get('access_token')
    except json.JSONDecodeError as e:
        raise FetchError(f'Malformed token response from {TOKEN_URL}: {e}') from e

    if not token:
        raise FetchError('Token response did not contain an access_token')

    return token

def call_api(token, method, path, body=None):
    """Call one Reporting API endpoint and return the decoded JSON body.

    Every failure raises. A caller that mistook an error for an empty report
    would record the day as collected and never come back to it.
    """
    url = f'{API_BASE}{path}'
    headers = {
        'Authorization': f'Bearer {token}',
        'Accept': 'application/json',
        'Content-Type': 'application/json',
        'User-Agent': 'GoogleSecOps-ProofpointSERCollector/1.0',
    }
    payload = json.dumps(body) if body is not None else None
    backoff = 1.0
    retries = 0

    while True:
        try:
            response = http.request(method, url, body=payload, headers=headers)
        except Exception as e:
            raise FetchError(f'Request to {url} failed: {e}') from e

        if response.status == 429:
            retries += 1
            if retries > MAX_RATE_LIMIT_RETRIES:
                raise FetchError(f'Rate limited repeatedly by {url}; giving up without recording the day')
            raw_retry_after = response.headers.get('Retry-After')
            try:
                # Retry-After may also be an HTTP date, which int() cannot parse.
                delay = int(raw_retry_after) if raw_retry_after else int(backoff)
            except (TypeError, ValueError):
                delay = int(backoff)
            print(f'Rate limited (429) on {path}. Retrying after {delay}s...')
            time.sleep(delay)
            backoff = min(backoff * 2, 30.0)
            continue

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

        try:
            return json.loads(response.data.decode('utf-8'))
        except json.JSONDecodeError as e:
            raise FetchError(f'Malformed JSON response from {url}: {e}') from e

def as_list(value):
    """Return a response data member as a list.

    The Reporting API returns data as an object on some endpoints and as an
    array on others, and the published schema declares neither.
    """
    if value is None:
        return []
    if isinstance(value, list):
        return value
    return [value]

def collect_day(token, day, include_overview):
    """Collect one reporting day and return the records to write.

    The nesting of each record is chosen to match what the parser expects.
    Rows carrying acceptedMessages are emitted bare, because the parser nests
    those under data itself. The other two responses are emitted whole, so
    that their data and metadata envelope survives to the parser.
    """
    records = []

    # Licensed throughput position. This endpoint takes no date and reports
    # the tenant's standing right now, so it is collected once per run rather
    # than once per day: a backfill would otherwise write the same snapshot
    # against every day it catches up on.
    if include_overview:
        records.append(call_api(token, 'GET', '/v1/usage/overview'))

    # Volume in bytes for the day. One envelope per row, so that the
    # totalThroughput in metadata stays attached to the row it describes.
    data_trend = call_api(token, 'POST', '/v1/usage/data-trend', {
        'dates': day,
        'interval': 'day',
    })
    for element in as_list(data_trend.get('data')):
        records.append({'data': [element], 'metadata': data_trend.get('metadata', {})})

    # Message counters for the day.
    message_trend = call_api(token, 'POST', '/v1/usage/message-trend', {
        'dates': day,
        'interval': 'day',
    })
    records.extend(as_list(message_trend.get('data')))

    # Per relay user counters, paginated.
    page = 1
    while True:
        if page > MAX_PAGES:
            raise FetchError(f'Relay user pagination for {day} exceeded MAX_PAGES ({MAX_PAGES})')

        payload = call_api(token, 'POST', '/v1/usage/relay-users', {
            'dates': day,
            'pageNum': page,
            'pageSize': PAGE_SIZE,
        })
        elements = as_list(payload.get('data'))
        if not elements:
            break

        records.extend(elements)

        pagination = (payload.get('metadata') or {}).get('pagination') or {}
        total_pages = pagination.get('totalPages')
        if not total_pages or page >= total_pages:
            break
        page += 1

    return records

def record_key(record):
    """Return the deduplication identity of one record.

    Reporting rows are aggregates and carry no identifier of their own, so
    identity is the content hash. Keys are held per reporting day, which makes
    a repeated run of the same day a no-op without ever suppressing a new day
    whose figures happen to be identical.
    """
    return 'sha256:' + hashlib.sha256(
        json.dumps(record, sort_keys=True, ensure_ascii=False).encode('utf-8')
    ).hexdigest()

def pending_days(last_report_date, target):
    """Return the reporting days still to collect, oldest first."""
    if not last_report_date:
        start = target
    else:
        start = date.fromisoformat(last_report_date) + timedelta(days=1)

    if start > target:
        return []

    if (target - start).days >= MAX_BACKFILL_DAYS:
        start = target - timedelta(days=MAX_BACKFILL_DAYS - 1)
        print(f'Backfill capped at {MAX_BACKFILL_DAYS} days. Days before {start.isoformat()} are not collected.')

    return [start + timedelta(days=offset) for offset in range((target - start).days + 1)]

def prune_seen(seen, target):
    """Drop deduplication keys for days outside the retention window."""
    cutoff = target - timedelta(days=SEEN_RETENTION_DAYS)
    return {day: keys for day, keys in seen.items() if date.fromisoformat(day) >= cutoff}

@functions_framework.cloud_event
def main(cloud_event):
    """Fetch Proofpoint SER usage reports and write them to Cloud Storage.

    Args:
        cloud_event: CloudEvent object containing the Pub/Sub message.
    """
    if not all([GCS_BUCKET, CLIENT_ID, CLIENT_SECRET]):
        # Raise rather than return: a bare return acks the Pub/Sub message and
        # reports the run as successful, silently discarding the schedule tick.
        raise RuntimeError('Missing required environment variables')

    bucket = storage_client.bucket(GCS_BUCKET)
    state = load_state(bucket, STATE_KEY)
    seen = state.get('seen') or {}

    now = datetime.now(timezone.utc)
    target = (now - timedelta(days=REPORT_DAY_OFFSET)).date()

    days = pending_days(state.get('last_report_date'), target)
    if not days:
        print(f'Nothing to collect. {target.isoformat()} is already recorded.')
        return

    token = get_access_token()

    for day in days:
        key = day.isoformat()
        print(f'Collecting reporting day {key}')

        # A FetchError here propagates: the run fails, the state is left at the
        # last day that was fully written, and the next run retries this day.
        records = collect_day(token, key, include_overview=(day == days[-1]))

        day_seen = set(seen.get(key, []))
        fresh = []
        for record in records:
            digest = record_key(record)
            if digest in day_seen:
                continue
            day_seen.add(digest)
            fresh.append(record)

        print(f'{key}: fetched {len(records)} records, {len(fresh)} new after deduplication')

        if fresh:
            timestamp = now.strftime('%Y%m%dT%H%M%SZ')
            object_key = f'{GCS_PREFIX}/usage_{key}_{timestamp}.ndjson'
            blob = bucket.blob(object_key)

            ndjson = '\n'.join(json.dumps(record, ensure_ascii=False) for record in fresh) + '\n'
            blob.upload_from_string(ndjson, content_type='application/x-ndjson')

            print(f'Wrote {len(fresh)} records to gs://{GCS_BUCKET}/{object_key}')

        # Record the day only after its data is durably written.
        seen[key] = sorted(day_seen)
        state['last_report_date'] = key
        state['seen'] = prune_seen(seen, target)
        save_state(bucket, STATE_KEY, state)

    print(f'Successfully processed {len(days)} reporting day(s)')

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

    Only a missing object is treated as a cold start. Any other error is
    raised: swallowing it would silently restart collection and re-ingest the
    whole backfill window.
    """
    blob = bucket.blob(key)
    try:
        return json.loads(blob.download_as_text())
    except NotFound:
        print('No state file found. Starting from the most recent completed day.')
        return {}

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

    Failures are raised, not logged. If the state write fails after the data
    was uploaded, the next run repeats the same day and duplicates it.
    """
    blob = bucket.blob(key)
    blob.upload_from_string(
        json.dumps(state, indent=2),
        content_type='application/json',
    )
    print(f'Saved state: last_report_date={state.get("last_report_date")}')

  • 두 번째 파일 - requirements.txt:

    functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0
  1. 배포를 클릭하여 함수를 저장하고 배포합니다.

  2. 배포가 완료될 때까지 기다립니다 (2~3분).

Cloud Scheduler 작업 만들기

Cloud Scheduler는 일정 간격으로 Pub/Sub 주제에 메시지를 게시하여 Cloud Run 함수를 트리거합니다.

  1. GCP Console에서 Cloud Scheduler로 이동합니다.
  2. 작업 만들기를 클릭합니다.

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

    설정 값
    이름 proofpoint-ser-collector-daily
    리전 Cloud Run 함수와 동일한 리전 선택
    주파수 0 2 * * * (매일 오전 2시)
    시간대 시간대 선택 (UTC 권장)
    타겟 유형 Pub/Sub
    주제 Pub/Sub 주제 (proofpoint-ser-trigger)를 선택합니다.
    메일 본문 {} (빈 JSON 객체)

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

일정 빈도 옵션

Reporting API는 일별로 집계하므로 완료된 날짜는 한 번 수집됩니다. 테넌트의 시간대에서 보고일이 종료된 후 매일 작업을 실행합니다.

빈도 크론 표현식 사용 사례
매일 02:00 0 2 * * * 표준 (권장)
매일 06:00 0 6 * * * 보고가 나중에 정산되는 테넌트
하루 두 번 0 2,14 * * * 첫 번째 실행이 실패한 경우 당일 재시도 추가

이미 기록된 날짜를 찾는 실행은 아무것도 쓰지 않고 종료되므로 추가 실행에는 API 호출이 한 번만 필요하며 데이터가 중복되지 않습니다.

통합 테스트

  1. Cloud Scheduler 콘솔에서 작업을 찾습니다.
  2. 강제 실행을 클릭하여 작업을 수동으로 트리거합니다.
  3. 몇 초 동안 기다립니다.
  4. Cloud Run > 서비스로 이동합니다.
  5. 함수 이름 (proofpoint-ser-collector)을 클릭합니다.
  6. 로그 탭을 클릭합니다.

  7. 함수가 성공적으로 실행되었는지 확인합니다. 다음 항목을 찾습니다.

    Collecting reporting day YYYY-MM-DD
YYYY-MM-DD: fetched X records, X new after deduplication
Wrote X records to gs://proofpoint-ser-logs/ser-logs/usage_YYYY-MM-DD_YYYYMMDDTHHMMSSZ.ndjson
Saved state: last_report_date=YYYY-MM-DD
Successfully processed 1 reporting day(s)

  8. Cloud Storage> 버킷으로 이동합니다.

  9. 버킷 이름 (proofpoint-ser-logs)을 클릭합니다.

  10. 접두사 폴더 (ser-logs/)로 이동합니다.

  11. 현재 타임스탬프를 사용하여 새 .ndjson 파일이 생성되었는지 확인합니다.

로그에 오류가 표시되는 경우:

  • 토큰 엔드포인트에서 HTTP 400: 키 또는 비밀번호가 잘못되었거나 키가 만료되었습니다. invalid_client는 쌍이 거부되었음을 의미합니다. 대체 키를 만드세요.
  • Reporting API의 HTTP 401: 전달자 토큰이 전송되지 않았거나 더 이상 유효하지 않습니다.
  • HTTP 429: 비율 제한입니다. 함수는 지수 백오프로 재시도하고 제한이 지속되면 실행이 실패하여 다음 실행으로 넘어갑니다.
  • 환경 변수 누락: 필요한 모든 변수가 설정되어 있는지 확인합니다.

Google SecOps 서비스 계정 가져오기

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

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

  1. SIEM 설정> 피드로 이동합니다.
  2. 새 피드 추가를 클릭합니다.
  3. 단일 피드 구성을 클릭합니다.
  4. 피드 이름 필드에 피드 이름을 입력합니다(예: Proofpoint SER Logs).
  5. 소스 유형으로 Google Cloud Storage V2를 선택합니다.
  6. 로그 유형으로 ProofPoint Secure Email Relay를 선택합니다.
  7. 서비스 계정 가져오기를 클릭합니다.

  8. 고유한 서비스 계정 이메일이 표시됩니다(예:

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

  9. 다음 단계에서 사용할 수 있도록 이 이메일 주소를 복사합니다.

  10. 다음을 클릭합니다.

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

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

      gs://proofpoint-ser-logs/ser-logs/
      • 다음과 같이 바꿉니다.
        • proofpoint-ser-logs: Cloud Storage 버킷 이름입니다.
        • ser-logs: 로그가 저장되는 선택적 접두사 또는 폴더 경로입니다 (루트의 경우 비워 둠).

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

      • 파일 삭제 안함: 소스에서 파일을 삭제하지 않습니다 (테스트에 권장).
      • 전송된 파일 및 빈 디렉터리 삭제: 가져오기가 완료되면 소스에서 파일과 빈 디렉터리를 삭제합니다.

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

    • 애셋 네임스페이스: 애셋 네임스페이스

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

  12. 다음을 클릭합니다.

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

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

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

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

  6. 저장을 클릭합니다.

UDM 매핑 테이블

로그 필드 UDM 매핑 로직
acceptedMessages_label additional.fields 병합됨
acceptedThroughput_label additional.fields 병합됨
average30DayThroughput_label additional.fields 병합됨
average7DayThroughput_label additional.fields 병합됨
averageDailyThroughput_label additional.fields 병합됨
avgAcceptedMessageSize_label additional.fields 병합됨
blockedMessages_label additional.fields 병합됨
data_totalThroughput_label additional.fields 병합됨
deliveredMessages_label additional.fields 병합됨
licenseEndDate_label additional.fields 병합됨
licenseStartDate_label additional.fields 병합됨
map_label additional.fields 병합됨
metadata_totalThroughput_label additional.fields 병합됨
quarantinedMessages_label additional.fields 병합됨
rejectedMessages_label additional.fields 병합됨
remainingThroughput_label additional.fields 병합됨
requestedMessages_label additional.fields 병합됨
requestedThroughput_label additional.fields 병합됨
sentMessages_label additional.fields 병합됨
throughputForecast_label additional.fields 병합됨
throughputLimit_label additional.fields 병합됨
throughput_label additional.fields 병합됨
totalMessages_label additional.fields 병합됨
undeliveredMessages_label additional.fields 병합됨
data.name metadata.description 직접 매핑됨
desc metadata.description 직접 매핑됨
data.date metadata.event_timestamp yyyy-MM-dd로 파싱됨
event_type metadata.event_type 직접 매핑됨
fromEnvelope network.email.bounce_address 직접 매핑됨
fromHeader network.email.from 직접 매핑됨
applicationName principal.administrative_domain 직접 매핑됨
principal_host principal.asset.hostname 직접 매핑됨
principal_host principal.hostname 직접 매핑됨
principal_port principal.port 직접 매핑됨
data.relayUserId principal.user.product_object_id 직접 매핑됨
userId principal.user.product_object_id 직접 매핑됨
applicationUserName principal.user.user_display_name 직접 매핑됨
senderName target.administrative_domain 직접 매핑됨
senderId target.user.product_object_id 직접 매핑됨
해당 사항 없음 metadata.product_name 상수: PROOFPOINT SER
해당 사항 없음 metadata.vendor_name 상수: PROOFPOINT

변경 로그

이 파서의 변경 로그 보기

도움이 더 필요하신가요? 커뮤니티 회원 및 Google SecOps 전문가에게 문의하여 답변을 받으세요.