SAP BTP 로그 수집

다음에서 지원:

이 문서에서는 Cloud Storage V2를 사용하여 SAP Business Technology Platform(BTP) 로그를 Google Security Operations에 수집하는 방법을 설명합니다.

SAP Business Technology Platform(BTP)은 애플리케이션 개발, 통합, 데이터 관리, 분석 기능을 제공하는 클라우드 플랫폼입니다. 플랫폼 활동을 모니터링하고 보안 위협을 감지하는 데 중요한 보안 감사 로그, 액세스 로그, 서비스 이벤트 로그를 생성합니다. Cloud Run 함수는 일정에 따라 SAP 감사 로그 관리 서비스의 감사 로그 검색 API를 폴링하고, 로그를 NDJSON 형식으로 Cloud Storage 버킷에 기록하며, Google SecOps는 Cloud Storage V2 피드를 통해 이를 수집합니다.

시작하기 전에

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

  • Google SecOps 인스턴스
  • Cloud Storage API가 사용 설정된 Google Cloud 프로젝트
  • Cloud Storage 버킷을 만들고 관리할 수 있는 권한
  • Cloud Storage 버킷에서 Identity and Access Management(IAM) 정책을 관리할 수 있는 권한
  • Cloud Run 서비스, Pub/Sub 주제, Cloud Scheduler 작업을 만들 수 있는 권한
  • 감사 로그 검색을 구성할 수 있는 권한이 있는 SAP BTP에 대한 관리자 액세스 권한(대상 Cloud Foundry 공간의 공간 개발자 및 전역 계정 수준에서 사용 권한을 추가할 수 있는 기능)

  • SAP BTP 하위 계정에 대한 감사 로그 관리 서비스(기술 이름: auditlog-management)

Cloud Storage 버킷 만들기

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

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

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

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

SAP BTP API 사용자 인증 정보 수집

감사 로그 관리 서비스 인스턴스 만들기

  1. SAP BTP Cockpit에 로그인합니다.
  2. 감사 로그 관리 서비스 사용 권한이 하위 계정에 할당되어 있는지 확인합니다. 하위 계정 뷰에서 사용 권한 > 수정 > 서비스 요금제 추가로 이동합니다(사용 권한은 전역 계정에서 할당됨).
  3. 하위 계정으로 이동합니다.
  4. 서비스 > 인스턴스 및 구독으로 이동합니다.
  5. 만들기를 클릭합니다.
  6. 다음 구성 세부정보를 제공합니다.
    • 서비스: 감사 로그 관리 서비스(기술 이름: auditlog-management)를 선택합니다.
    • 계획: 기본값을 선택합니다.
    • 런타임 환경: Cloud Foundry를 선택합니다.
    • 스페이스: 대상 스페이스를 선택합니다.
    • 인스턴스 이름: 이름을 입력합니다(예: auditlog-management-secops).
  7. 만들기를 클릭합니다.

서비스 키 만들기

  1. SAP BTP Cockpit에서 서비스 > 인스턴스 및 구독으로 이동하고 auditlog-management 서비스 인스턴스를 선택합니다.
  2. 서비스 키 만들기를 선택합니다.
  3. 이름을 입력합니다(예: google-secops-integration).
  4. 만들기를 클릭합니다.

  5. 서비스 키에서 다음 세부정보를 복사하여 안전한 위치에 저장합니다.

    • url: 감사 로그 검색 요청의 기준 URL입니다.
    • uaa.url: OAuth 서버 기준 URL입니다. 토큰 엔드포인트는 <uaa.url>/oauth/token입니다.
    • uaa.clientid: OAuth 클라이언트 ID입니다.
    • uaa.clientsecret: OAuth 클라이언트 보안 비밀번호입니다.

테스트 API 액세스

  • 통합을 진행하기 전에 사용자 인증 정보를 테스트합니다.

    # Replace with your actual credentials from the service key
UAA_URL="https://your-subdomain.authentication.region.hana.ondemand.com"
CLIENT_ID="your-client-id"
CLIENT_SECRET="your-client-secret"
API_URL="https://auditlog-management.cfapps.us10.hana.ondemand.com"

# Get OAuth token
TOKEN=$(curl -s -X POST "${UAA_URL}/oauth/token" \
  -u "${CLIENT_ID}:${CLIENT_SECRET}" \
  -d "grant_type=client_credentials" \
  -H "Content-Type: application/x-www-form-urlencoded" | jq -r '.access_token')

# Test API access: the API takes time_from/time_to in UTC without a
# timezone suffix and returns a JSON array at the root. When the result
# set is chunked, the next chunk's handle arrives in the "Paging"
# response header.
curl -v -H "Authorization: Bearer ${TOKEN}" \
  "${API_URL}/auditlog/v2/auditlogrecords?time_from=2026-08-07T00:00:00&time_to=2026-08-07T01:00:00"

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

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

서비스 계정 만들기

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

Pub/Sub 주제 만들기

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

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

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

Cloud Run 함수는 Cloud Scheduler의 Pub/Sub 메시지에 의해 트리거되어 감사 로그 관리 서비스 검색 API에서 감사 로그를 가져와 Cloud Storage에 기록합니다.

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

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

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

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

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

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

    1. 인증 필요를 선택합니다.
    2. Identity and Access Management(IAM)를 확인합니다.

  7. 컨테이너, 네트워킹, 보안을 펼칩니다.

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

    • 서비스 계정: 서비스 계정 sap-btp-collector-sa를 선택합니다.

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

    1. 변수 및 보안 비밀을 클릭합니다.
    2. 각 환경 변수에 대해 + 변수 추가를 클릭합니다.
    변수 이름 예시 값 설명
    GCS_BUCKET sap-btp-logs Cloud Storage 버킷 이름
    GCS_PREFIX sap-btp 로그 파일의 프리픽스
    STATE_KEY sap-btp-state.json 로그 프리픽스 외부의 상태 파일 경로
    SAP_API_URL https://auditlog-management.cfapps.us10.hana.ondemand.com 서비스 키의 url 값
    SAP_UAA_URL https://your-subdomain.authentication.region.hana.ondemand.com OAuth 토큰 엔드포인트 URL
    SAP_CLIENT_ID your-client-id OAuth 클라이언트 ID
    SAP_CLIENT_SECRET your-client-secret OAuth 클라이언트 보안 비밀번호
    LOOKBACK_HOURS 24 초기 대상 기간

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

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

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

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

  12. 실행 환경으로 이동합니다.

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

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

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

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

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

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

함수 코드 추가

  1. 함수 진입점에 main을 입력합니다.

  2. 인라인 코드 편집기에서 다음 두 파일을 만듭니다.

    • 첫 번째 파일 - main.py:

      import functions_framework
from google.cloud import storage
import hashlib
import json
import os
import urllib3
from datetime import datetime, timezone, timedelta
import time
import base64
import uuid

# 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', 'sap-btp')
STATE_KEY = os.environ.get('STATE_KEY', 'sap-btp-state.json')
SEEN_KEY = STATE_KEY + '.seen'
API_URL = os.environ.get('SAP_API_URL')
UAA_URL = os.environ.get('SAP_UAA_URL')
CLIENT_ID = os.environ.get('SAP_CLIENT_ID')
CLIENT_SECRET = os.environ.get('SAP_CLIENT_SECRET')
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))

def parse_datetime(value: str) -> datetime:
    """Parse ISO datetime string to datetime object."""
    if value.endswith("Z"):
        value = value[:-1] + "+00:00"
    return datetime.fromisoformat(value)

# message_uuid is the identifier of an audit log record; the parser maps it to
# metadata.product_log_id.
ID_FIELDS = ('message_uuid',)

def event_keys(event):
    """Return every identity this event can be recognised by."""
    keys = {'sha256:' + hashlib.sha256(
        json.dumps(event, sort_keys=True, ensure_ascii=False).encode('utf-8')
    ).hexdigest()}
    for field in ID_FIELDS:
        value = event.get(field)
        if value:
            keys.add(f'{field}:{value}')
    return keys

def get_oauth_token():
    """Get OAuth 2.0 access token using client credentials flow."""
    token_url = f"{UAA_URL.rstrip('/')}/oauth/token"

    credentials = f"{CLIENT_ID}:{CLIENT_SECRET}"
    encoded_credentials = base64.b64encode(credentials.encode('utf-8')).decode('utf-8')

    headers = {
        'Authorization': f'Basic {encoded_credentials}',
        'Content-Type': 'application/x-www-form-urlencoded',
        'Accept': 'application/json'
    }

    body = 'grant_type=client_credentials'

    backoff = 1.0
    max_retries = 3

    for attempt in range(max_retries):
        response = http.request('POST', token_url, body=body, headers=headers)

        if response.status == 429:
            retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
            print(f"Rate limited (429) on token request. Retrying after {retry_after}s...")
            time.sleep(retry_after)
            backoff = min(backoff * 2, 30.0)
            continue

        if response.status != 200:
            raise RuntimeError(f"Failed to get access token: {response.status} - {response.data.decode('utf-8')}")

        data = json.loads(response.data.decode('utf-8'))
        return data['access_token']

    raise RuntimeError(f"Failed to get token after {max_retries} retries due to rate limiting")

@functions_framework.cloud_event
def main(cloud_event):
    """
    Cloud Run function triggered by Pub/Sub to fetch SAP BTP
    audit logs and write to GCS.

    Args:
        cloud_event: CloudEvent object containing Pub/Sub message
    """

    if not all([GCS_BUCKET, API_URL, UAA_URL, 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')

    try:
        bucket = storage_client.bucket(GCS_BUCKET)

        # Load state
        state = load_state(bucket, STATE_KEY)
        seen_keys = load_seen(bucket, SEEN_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"])
                last_time = last_time - timedelta(minutes=2)
            except Exception as e:
                print(f"Warning: Could not parse last_event_time: {e}")

        if last_time is None:
            last_time = now - timedelta(hours=LOOKBACK_HOURS)

        print(f"Fetching audit logs from {last_time.isoformat()} to {now.isoformat()}")

        # Get OAuth token
        token = get_oauth_token()

        # Fetch audit logs
        records, newest_event_time = fetch_audit_logs(
            token=token,
            start_time=last_time,
            end_time=now,
        )

        # Drop the events already written by a previous run (overlap window)
        fresh = [r for r in records if not (event_keys(r) & seen_keys)]

        if not fresh:
            print("No new audit log records found.")
            return

        # Write to GCS as NDJSON
        timestamp = now.strftime('%Y%m%d_%H%M%S')
        # The random suffix keeps concurrent executions (Pub/Sub delivers at
        # least once) from overwriting each other's object within one second.
        object_key = f"{GCS_PREFIX}/audit_{timestamp}_{uuid.uuid4().hex[:8]}.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}")

        # Remember every fetched record so the next overlap window is deduplicated
        save_seen(bucket, SEEN_KEY, records)

        # Update state with newest event time. The watermark must be an event
        # timestamp, never wall clock: a wall-clock watermark skips every event
        # the vendor indexes late.
        if not newest_event_time:
            raise RuntimeError('Records were returned but no event timestamp could be parsed')
        save_state(bucket, STATE_KEY, newest_event_time)

        print(f"Successfully processed {len(fresh)} records")

    except Exception as e:
        print(f'Error processing audit logs: {str(e)}')
        raise

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}")
        raise

    return {}

def save_state(bucket, key, last_event_time_iso: str):
    """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}")
        raise

def load_seen(bucket, key):
    """Load the event keys written by the previous run."""
    try:
        blob = bucket.blob(key)
        if blob.exists():
            return set(json.loads(blob.download_as_text()))
    except Exception as e:
        print(f"Warning: Could not load seen keys: {e}")
        raise

    return set()

def save_seen(bucket, key, records):
    """Save the keys of every fetched record for the next overlap window."""
    try:
        keys = set()
        for record in records:
            keys |= event_keys(record)
        blob = bucket.blob(key)
        blob.upload_from_string(
            json.dumps(sorted(keys)),
            content_type='application/json'
        )
        print(f"Saved {len(keys)} seen keys")
    except Exception as e:
        print(f"Warning: Could not save seen keys: {e}")
        raise

def fetch_audit_logs(token: str, start_time: datetime, end_time: datetime):
    """
    Fetch audit logs from the Audit Log Management Service retrieval API.

    The API takes time_from/time_to in UTC without a timezone designator and
    returns a JSON array at the root (HTTP 204 when the window is empty).
    When the result set is chunked, the next chunk's handle arrives in the
    "Paging" response header as handle=<value> and is passed back as the
    handle query parameter; the server chunk size is fixed at 500 records.

    Args:
        token: OAuth 2.0 access token
        start_time: Start time for log query
        end_time: End time for log query

    Returns:
        Tuple of (records list, newest_event_time ISO string)

    Raises:
        RuntimeError: on any API failure, so a failed fetch can never be
            mistaken for an empty result.
    """
    base_url = API_URL.rstrip('/')
    endpoint = f"{base_url}/auditlog/v2/auditlogrecords"

    headers = {
        'Authorization': f'Bearer {token}',
        'Accept': 'application/json',
        'User-Agent': 'GoogleSecOps-SAPBTPCollector/1.0'
    }

    records = []
    newest_time = None
    page_num = 0
    backoff = 1.0
    rate_limit_retries = 0
    token_refreshed = False

    # The API expects UTC timestamps with no timezone suffix.
    time_from = start_time.strftime('%Y-%m-%dT%H:%M:%S')
    time_to = end_time.strftime('%Y-%m-%dT%H:%M:%S')
    handle = None

    from urllib.parse import urlencode

    while True:
        page_num += 1

        # Rebuild the full parameter set on every request.
        params = {'time_from': time_from, 'time_to': time_to}
        if handle:
            params['handle'] = handle
        url = f"{endpoint}?{urlencode(params)}"

        response = http.request('GET', url, headers=headers)

        # The documented rate limit is 4 requests per second (burst 20).
        if response.status == 429:
            rate_limit_retries += 1
            if rate_limit_retries > 5:
                raise RuntimeError('Rate limited repeatedly; giving up without advancing the watermark')
            retry_after = response.headers.get('Retry-After')
            try:
                delay = float(retry_after) if retry_after else backoff
            except (TypeError, ValueError):
                delay = backoff
            delay = min(max(delay, 1.0), 60.0)
            print(f"Rate limited (429). Retrying after {delay}s...")
            time.sleep(delay)
            backoff = min(backoff * 2, 30.0)
            continue

        backoff = 1.0
        rate_limit_retries = 0

        if response.status == 401:
            # One refresh per run: a second consecutive 401 means the audit
            # service rejects tokens UAA still issues, and retrying forever
            # would hammer both services until the platform kills the run.
            if token_refreshed:
                raise RuntimeError('Audit log request still unauthorized after a token refresh')
            print("Token expired, refreshing...")
            token = get_oauth_token()
            headers['Authorization'] = f'Bearer {token}'
            token_refreshed = True
            continue

        token_refreshed = False

        if response.status == 204:
            # Documented empty result for the requested window.
            break

        if response.status != 200:
            raise RuntimeError(
                f"Audit log request failed: {response.status} - "
                f"{response.data.decode('utf-8')}"
            )

        data = json.loads(response.data.decode('utf-8'))

        # The records arrive as a JSON array at the root. Tolerate wrapped
        # shapes defensively, but never call .get on a list: doing so
        # crashed previous revisions of this function.
        if isinstance(data, list):
            page_results = data
        elif isinstance(data, dict):
            page_results = data.get('value', data.get('results', []))
        else:
            page_results = []

        print(f"Page {page_num}: Retrieved {len(page_results)} audit records")
        records.extend(page_results)

        # Track newest event time. Parse before assigning: an unparseable
        # stamp must never become the persisted watermark.
        for record in page_results:
            event_time = record.get('time') or record.get('timestamp')
            if not event_time:
                continue
            try:
                parsed = parse_datetime(event_time)
            except Exception as e:
                print(f"Warning: Could not parse event time: {e}")
                continue
            if newest_time is None or parsed > parse_datetime(newest_time):
                newest_time = event_time

        # More chunks are signaled only by the Paging response header.
        paging = response.headers.get('Paging') or response.headers.get('paging') or ''
        handle = paging.split('=', 1)[1] if '=' in paging else None
        if not handle:
            break

    print(f"Retrieved {len(records)} total audit records 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-btp-collector-hourly
    리전 Cloud Run 함수와 동일한 리전 선택
    주파수 0 * * * *(매시간 정각)
    시간대 시간대 선택(UTC 권장)
    타겟 유형 Pub/Sub
    주제 sap-btp-trigger 주제를 선택합니다.
    메일 본문 {}(빈 JSON 객체)

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

일정 빈도 옵션

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

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

통합 테스트

  1. Cloud Scheduler 콘솔에서 작업(sap-btp-collector-hourly)을 찾습니다.
  2. 강제 실행을 클릭하여 수동으로 트리거합니다.
  3. 몇 초 기다린 후 Cloud Run > 서비스 > sap-btp-collector > 로그로 이동합니다.

  4. 함수가 실행되었는지 확인합니다. 다음 항목을 찾습니다.

    Fetching audit logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
Page 1: Retrieved X audit records
Wrote X records to gs://sap-btp-logs/sap-btp/audit_YYYYMMDD_HHMMSS.ndjson
Successfully processed X records

  5. Cloud Storage 버킷(sap-btp-logs)을 확인하여 감사 로그가 기록되었는지 확인합니다.

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

  • HTTP 401: 환경 변수의 OAuth 사용자 인증 정보를 확인하거나 토큰이 만료되었을 수 있습니다(함수가 자동으로 새로고침을 처리함).
  • HTTP 403: 서비스 키가 로그를 예상하는 동일한 하위 계정의 auditlog-management 인스턴스에 속하는지 확인합니다. 사용자 역할 컬렉션은 클라이언트 사용자 인증 정보 토큰에 적용되지 않습니다.
  • HTTP 429: 비율 제한 - 함수가 백오프를 사용하여 자동으로 재시도합니다.
  • 액세스 토큰을 가져올 수 없음: SAP_UAA_URL, SAP_CLIENT_ID, SAP_CLIENT_SECRET가 올바른지 확인합니다.

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

  1. 설정 > 데이터 관리 > 피드로 이동합니다. SIEM 전용 플랫폼을 사용하는 경우 설정 > 피드로 이동합니다.
  2. 새 피드 추가를 클릭합니다.
  3. 단일 피드 구성을 클릭합니다.
  4. 피드 이름 필드에 피드 이름을 입력합니다(예: SAP BTP Logs).
  5. 소스 유형으로 Google Cloud Storage V2를 선택합니다.
  6. 로그 유형으로 SAP Business Technology Platform을 선택합니다.

  7. 서비스 계정 가져오기를 클릭합니다. 고유한 서비스 계정 이메일이 표시됩니다. 예를 들면 다음과 같습니다.

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

  8. 이 이메일 주소를 복사합니다. 다음 단계에서 사용합니다.

  9. 다음을 클릭합니다.

  10. 다음 입력 파라미터의 값을 지정합니다.

    • 스토리지 버킷 URL: 프리픽스 경로가 있는 Cloud Storage 버킷 URI를 입력합니다.

      gs://sap-btp-logs/sap-btp/
      • 다음과 같이 바꿉니다.
        • sap-btp-logs: Cloud Storage 버킷 이름입니다.
        • sap-btp/: 로그가 저장되는 경로입니다.

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

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

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

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

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

  11. 다음을 클릭합니다.

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

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

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

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

  6. 저장을 클릭합니다.

UDM 매핑 테이블

로그 필드 UDM 매핑 로직
time metadata.event_timestamp ISO8601(으)로 파싱됨
principal_present metadata.event_type 매핑됨: true → STATUS_UPDATE
user_present metadata.event_type 매핑됨: true → USER_RESOURCE_ACCESS, true → USER_UNCATEGORIZED
json_message_uuid metadata.product_log_id 직접 매핑됨
host principal.asset.hostname 직접 매핑됨
key principal.asset.ip 매핑됨: ip → value
principal_ip principal.asset.ip 병합됨
value principal.asset.ip 병합됨
host principal.hostname 직접 매핑됨
key principal.ip 매핑됨: ip → value
principal_ip principal.ip 병합됨
value principal.ip 병합됨
principal_user_role principal.user.attribute.roles 병합됨
key principal.user.email_addresses 매핑됨: "new","old" → value2, "new","old" → value1
key1 principal.user.email_addresses 매핑됨: emails → value2, userName → value1
key2 principal.user.email_addresses 매핑됨: value → value2
value1 principal.user.email_addresses 매핑됨: ^.+@.+$ → value1
value2 principal.user.email_addresses 병합됨
user principal.user.user_display_name 직접 매핑됨
json_message.data_subject.id.userid principal.user.userid 직접 매핑됨
value1 principal.user.userid 직접 매핑됨
category security_result.category_details 병합됨
als_service_id_label security_result.detection_fields 병합됨
app_or_service_id_label security_result.detection_fields 병합됨
attri_label security_result.detection_fields 병합됨
attribute_label security_result.detection_fields 병합됨
attribute_label1 security_result.detection_fields 병합됨
custom_status_label security_result.detection_fields 병합됨
custom_success_label security_result.detection_fields 병합됨
custom_type_label security_result.detection_fields 병합됨
desc_label security_result.detection_fields 병합됨
format_version_label security_result.detection_fields 병합됨
key security_result.detection_fields 매핑됨: json_message → message_label, id → attribute_label, "new","old" → `attribu…
message_id_label security_result.detection_fields 병합됨
message_label security_result.detection_fields 병합됨
message_label_json_message security_result.detection_fields 병합됨
message_label_msgId security_result.detection_fields 병합됨
message_label_msgNo security_result.detection_fields 병합됨
message_label_uuid security_result.detection_fields 병합됨
message_label_version security_result.detection_fields 병합됨
org_id_label security_result.detection_fields 병합됨
space_id_label security_result.detection_fields 병합됨
tenant_label security_result.detection_fields 병합됨
key security_result.severity 매핑됨: level → INFORMATIONAL, level → MEDIUM, level → UNKNOWN_SEVERITY
level security_result.severity 매핑됨: INFO → INFORMATIONAL
value security_result.severity 매핑됨: INFO → INFORMATIONAL, WARN → MEDIUM
level security_result.severity_details 직접 매핑됨
value1 target.resource.product_object_id 직접 매핑됨
key1 target.resource.resource_type 매핑됨: clusterID → CLUSTER
해당 사항 없음 metadata.event_type 상수: GENERIC_EVENT
해당 사항 없음 metadata.product_name 상수: SAP_BTP
해당 사항 없음 metadata.vendor_name 상수: SAP_BTP
해당 사항 없음 security_result.severity 상수: INFORMATIONAL
해당 사항 없음 target.resource.resource_type 상수: CLUSTER

변경 로그

이 파서의 변경 로그 보기

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