MuleSoft Anypoint 플랫폼 로그 수집

다음에서 지원:

이 문서에서는 Cloud Storage를 사용하여 MuleSoft Anypoint 플랫폼 로그에서 Google Security Operations로 감사 추적 이벤트를 수집하는 방법을 설명합니다.

시작하기 전에

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

  • Google SecOps 인스턴스
  • Cloud Storage API가 사용 설정된 Google Cloud 프로젝트
  • Cloud Storage 버킷을 만들고 관리할 수 있는 권한
  • Cloud Storage 버킷에서 Identity and Access Management (IAM) 정책을 관리할 수 있는 권한
  • Cloud Run 함수, Pub/Sub 주제, Cloud Scheduler 작업을 만들 수 있는 권한
  • 서비스 계정을 만들 수 있는 권한
  • MuleSoft Anypoint Platform에 대한 액세스 권한 관리

MuleSoft 조직 ID 가져오기

  1. Anypoint Platform에 로그인합니다.
  2. 액세스 관리 > 조직으로 이동합니다.
  3. 비즈니스 그룹 표에서 조직 이름을 클릭합니다.
  4. 조직 ID (예: 0a12b3c4-d5e6-789f-1021-1a2b34cd5e6f)를 복사합니다.

또는 MuleSoft 비즈니스 그룹으로 이동하여 URL에서 ID를 복사합니다.

MuleSoft 연결된 앱 만들기

  1. Anypoint Platform에 로그인합니다.
  2. 액세스 관리 > 연결된 앱 > 앱 만들기로 이동합니다.
  3. 다음 구성 세부정보를 제공합니다.
    • 앱 이름: 고유한 이름을 입력합니다 (예: Google SecOps export).
    • 앱이 자체적으로 작동 (클라이언트 사용자 인증 정보)을 선택합니다.
  4. 범위 추가 > 감사 로그 뷰어 > 다음을 클릭합니다.
  5. 로그가 필요한 모든 비즈니스 위치 그룹을 선택합니다.
  6. 다음 > 범위 추가를 클릭합니다.

  7. 저장을 클릭하고 클라이언트 ID와 클라이언트 보안 비밀번호를 복사합니다.

Cloud Storage 버킷 만들기

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

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

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

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

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

Cloud Run 함수에는 Cloud Storage 버킷에 쓸 수 있는 권한이 있는 서비스 계정이 필요합니다.

서비스 계정 만들기

  1. GCP 콘솔에서 IAM 및 관리자 > 서비스 계정으로 이동합니다.
  2. 서비스 계정 만들기를 클릭합니다.
  3. 다음 구성 세부정보를 제공합니다.
    • 서비스 계정 이름: mulesoft-logs-collector-sa을 입력합니다.
    • 서비스 계정 설명: Service account for Cloud Run function to collect MuleSoft Anypoint 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 버킷에 대한 서비스 계정 write 권한을 부여합니다.

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

Pub/Sub 주제 만들기

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

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

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

Cloud Run 함수는 Cloud Scheduler의 Pub/Sub 메시지에 의해 트리거되어 MuleSoft Anypoint API에서 로그를 가져와 Cloud Storage에 기록합니다.

  1. GCP 콘솔에서 Cloud Run으로 이동합니다.
  2. 함수 작성을 클릭합니다.

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

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

  4. 트리거 섹션에서 트리거 추가를 클릭하고 Pub/Sub 트리거를 선택합니다.

  5. Eventarc 트리거 창에서 다음 구성 세부정보를 제공합니다.

    • 트리거 이름: 생성된 이름을 유지하거나 트리거 이름을 입력합니다.
    • 트리거 유형: Google 소스를 선택합니다.
    • 이벤트 제공자: Pub/Sub를 선택합니다.
    • 이벤트 유형: google.cloud.pubsub.topic.v1.messagePublished를 선택합니다.
    • Cloud Pub/Sub 주제 선택: mulesoft-audit-trigger 주제를 선택합니다.
    • 리전: 함수와 동일한 리전을 선택합니다.
    • 서비스 계정: 서비스 계정 mulesoft-logs-collector-sa를 선택합니다.

  6. 트리거 저장을 클릭합니다.

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

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

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

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

    • 서비스 계정: mulesoft-logs-collector-sa 서비스 계정을 선택합니다.

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

    1. 변수 및 보안 비밀을 클릭합니다.
    2. 각 환경 변수에 대해 + 변수 추가를 클릭합니다.
    변수 이름 예시 값 설명
    GCS_BUCKET mulesoft-audit-logs Cloud Storage 버킷 이름
    GCS_PREFIX mulesoft-audit 로그 파일의 접두사
    STATE_KEY mulesoft-audit-state.json 로그 접두사 외부의 상태 경로
    MULE_ORG_ID your_org_id MuleSoft 조직 ID
    CLIENT_ID your_client_id 연결된 앱 클라이언트 ID
    CLIENT_SECRET your_client_secret 연결된 앱 클라이언트 보안 비밀번호
    MAX_RECORDS 10000 실행당 최대 레코드 수
    PAGE_SIZE 200 페이지당 항목 수 (API 최대 200개)
    LOOKBACK_HOURS 24 첫 실행 시 초기 대상 기간

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

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

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

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

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

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

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

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

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

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

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

함수 코드 추가

  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 urllib3
from urllib.parse import urlencode
from datetime import datetime, timezone, timedelta
import time

# 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()

# MuleSoft API endpoints
TOKEN_URL = "https://anypoint.mulesoft.com/accounts/api/v2/oauth2/token"
AUDIT_URL = "https://anypoint.mulesoft.com/audit/v2/organizations/{org_id}/query?cursorPagination=true"

# Environment variables
GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'mulesoft-audit')
# 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', 'mulesoft-audit-state.json')
MULE_ORG_ID = os.environ.get('MULE_ORG_ID')
CLIENT_ID = os.environ.get('CLIENT_ID')
CLIENT_SECRET = os.environ.get('CLIENT_SECRET')
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000'))
# The Audit Log Query API returns at most 200 entries per JSON query.
PAGE_SIZE = min(int(os.environ.get('PAGE_SIZE', '200')), 200)
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
# The query re-reads this many minutes before the watermark so that events the
# platform records late are still collected. Re-read events are removed by hash.
OVERLAP_MINUTES = int(os.environ.get('OVERLAP_MINUTES', '2'))

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

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

def to_unix_millis(dt: datetime) -> int:
    """Convert datetime to Unix epoch milliseconds."""
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return int(dt.astimezone(timezone.utc).timestamp() * 1000)

def parse_datetime(value: str) -> datetime:
    """Parse an ISO 8601 datetime string into an aware datetime."""
    if value.endswith("Z"):
        value = value[:-1] + "+00:00"
    dt = datetime.fromisoformat(value)
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt

def event_keys(event: dict) -> set:
    """Return the identity this event can be recognised by.

    Audit log entries carry no unique event ID, so a record counts as already
    collected when its full JSON content was seen before. The overlap window
    returns the same JSON again, which this catches without any knowledge of
    the payload schema.
    """
    return {'sha256:' + hashlib.sha256(
        json.dumps(event, sort_keys=True, ensure_ascii=False).encode('utf-8')
    ).hexdigest()}

def event_time(event: dict) -> str:
    """Return the event timestamp as an ISO 8601 string, or an empty string.

    Audit log entries carry the event time in `timestamp` as Unix milliseconds.
    """
    raw = event.get('timestamp')
    if raw is None or raw == '':
        return ''
    try:
        return datetime.fromtimestamp(int(raw) / 1000, tz=timezone.utc).isoformat()
    except (TypeError, ValueError):
        return ''

@functions_framework.cloud_event
def main(cloud_event):
    """Fetch MuleSoft Anypoint audit logs and write them to Cloud Storage.

    Args:
        cloud_event: CloudEvent object containing the Pub/Sub message.
    """
    if not all([GCS_BUCKET, MULE_ORG_ID, 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)

    now = datetime.now(timezone.utc)
    watermark = None
    if state.get('last_event_time'):
        watermark = parse_datetime(state['last_event_time'])
    seen_keys = set(state.get('seen_keys', []))

    if watermark is None:
        start_time = now - timedelta(hours=LOOKBACK_HOURS)
    else:
        start_time = watermark - timedelta(minutes=OVERLAP_MINUTES)

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

    token = get_token(CLIENT_ID, CLIENT_SECRET)

    # A FetchError here propagates: the run fails, the watermark is untouched,
    # and the next invocation retries the same window.
    records, newest_event_time = fetch_audit(
        org_id=MULE_ORG_ID,
        token=token,
        start_time=start_time,
        end_time=now,
        page_size=PAGE_SIZE,
        max_records=MAX_RECORDS,
    )

    # Remove events already written by an earlier run. Without this, the overlap
    # window re-emits its events on every invocation.
    fresh = [r for r in records if not (event_keys(r) & seen_keys)]
    print(f"Fetched {len(records)} records, {len(fresh)} new after deduplication")

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

    if not newest_event_time:
        # A watermark that cannot be derived from the data is not a watermark.
        # Advancing on wall-clock here would skip every late-recorded event.
        raise FetchError('Records were returned but no event timestamp could be parsed')

    timestamp = now.strftime('%Y%m%dT%H%M%SZ')
    object_key = f"{GCS_PREFIX}/audit_{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}")

    # Advance the watermark only after the data is durably written.
    new_watermark = parse_datetime(newest_event_time)
    if watermark and new_watermark < watermark:
        new_watermark = watermark

    # Retain only the hashes still inside the overlap window, so the state file
    # stays small while covering every event the next query can re-read.
    cutoff = new_watermark - timedelta(minutes=OVERLAP_MINUTES)
    retained = set()
    for record in records:
        stamp = event_time(record)
        if not stamp:
            continue
        if parse_datetime(stamp) >= cutoff:
            retained.update(event_keys(record))

    save_state(bucket, STATE_KEY, {
        'last_event_time': new_watermark.isoformat(),
        'seen_keys': sorted(retained),
    })

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

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 reset collection to the full lookback window
    and re-ingest that entire 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: dict):
    """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 window 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_event_time={state.get('last_event_time')}")

def retry_after_seconds(response, backoff: float) -> int:
    """Parse Retry-After defensively; it may be an HTTP date, which int() cannot parse."""
    raw = response.headers.get('Retry-After')
    try:
        return int(raw) if raw else int(backoff)
    except (TypeError, ValueError):
        return int(backoff)

def get_token(client_id: str, client_secret: str) -> str:
    """Get an OAuth 2.0 access token from MuleSoft using client credentials."""
    body = urlencode({
        'grant_type': 'client_credentials',
        'client_id': client_id,
        'client_secret': client_secret,
    }).encode('utf-8')

    backoff = 1.0
    for attempt in range(5):
        try:
            response = http.request(
                'POST',
                TOKEN_URL,
                body=body,
                headers={'Content-Type': 'application/x-www-form-urlencoded'},
            )
        except Exception as e:
            raise FetchError(f'Token request failed: {e}') from e

        if response.status == 429:
            wait = retry_after_seconds(response, backoff)
            print(f"Rate limited (429) on token request. Retrying after {wait}s...")
            time.sleep(wait)
            backoff = min(backoff * 2, 30.0)
            continue

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

        try:
            return json.loads(response.data.decode('utf-8'))['access_token']
        except (json.JSONDecodeError, KeyError) as e:
            raise FetchError(f'Malformed token response: {e}') from e

    raise FetchError('Rate limited repeatedly on token request; giving up without advancing the watermark')

def fetch_audit(org_id: str, token: str, start_time: datetime, end_time: datetime, page_size: int, max_records: int):
    """Fetch audit log entries from the MuleSoft Audit Log Query API.

    Uses cursor pagination. The API returns entries in ascending timestamp order,
    which a time watermark requires: with newest-first ordering a truncated run
    would advance the watermark past unread events.

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

    Raises:
        FetchError: on any API or transport failure, so that the caller does not
            mistake a failed fetch for an empty result and advance the watermark.
    """
    endpoint = AUDIT_URL.format(org_id=org_id)
    headers = {
        'Authorization': f'Bearer {token}',
        'Content-Type': 'application/json',
        'Accept': 'application/json',
    }

    records = []
    newest_time = None
    cursor = None
    page_num = 0
    backoff = 1.0
    rate_limit_retries = 0
    MAX_RATE_LIMIT_RETRIES = 5

    while True:
        page_num += 1

        if len(records) >= max_records:
            # Stop cleanly. The remaining events stay ahead of the watermark and
            # are collected by the next run.
            print(f"Reached max_records limit ({max_records})")
            break

        # Rebuild the full query on every page; the time window must never be dropped.
        body = {
            'startDate': to_unix_millis(start_time),
            'endDate': to_unix_millis(end_time),
            'limit': page_size,
            'ascending': True,
        }
        if cursor:
            body['cursor'] = cursor

        try:
            response = http.request(
                'POST',
                endpoint,
                body=json.dumps(body).encode('utf-8'),
                headers=headers,
            )
        except Exception as e:
            raise FetchError(f'Request to {endpoint} failed: {e}') from e

        if response.status == 429:
            rate_limit_retries += 1
            if rate_limit_retries > MAX_RATE_LIMIT_RETRIES:
                raise FetchError('Rate limited repeatedly; giving up without advancing the watermark')
            wait = retry_after_seconds(response, backoff)
            print(f"Rate limited (429). Retrying after {wait}s...")
            time.sleep(wait)
            backoff = min(backoff * 2, 30.0)
            continue

        backoff = 1.0
        rate_limit_retries = 0

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

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

        page_results = data.get('data') or []
        if not page_results:
            print("No more results (empty page)")
            break

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

        for event in page_results:
            stamp = event_time(event)
            if not stamp:
                continue
            if newest_time is None or parse_datetime(stamp) > parse_datetime(newest_time):
                newest_time = stamp

        cursor = data.get('cursor')
        if not cursor:
            break

    print(f"Retrieved {len(records)} total entries 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분).

중요 고려사항

  • 비율 제한: 감사 로그 쿼리 엔드포인트는 세 컨트롤 플레인에서 IP당 비율 제한을 적용합니다. 미국 컨트롤 플레인은 IP당 분당 700개의 요청을 허용하는 반면 EU 및 Gov 컨트롤 플레인은 IP당 분당 40개의 요청을 허용합니다. 이 함수는 비율 제한을 자동으로 처리하기 위해 지수 백오프를 구현합니다.

  • 토큰 만료: 액세스 토큰은 일반적으로 발급 후 약 30~60분 후에 만료됩니다. 이 함수는 실행마다 새 토큰을 요청하며 이는 해당 기간 내에 있습니다.

  • 중복 삭제 및 상태: 함수는 상태 파일에 최신 이벤트 시간의 워터마크와 중복 창 내 이벤트의 해시를 보관합니다. 각 실행은 워터마크 2분 전에 다시 읽어 지연 기록된 항목을 포착하고 이미 작성된 항목을 삭제하므로 재시도, 수동 실행 또는 스케줄러 지연으로 인해 중복 이벤트가 발생하지 않습니다. API 호출이 실패하면 실행이 실패하고 워터마크가 변경되지 않으므로 다음 실행에서 동일한 창을 다시 시도합니다.

  • 감사 로그 보관: 감사 로그의 기본 보관 기간은 1년입니다. 조직이 2023년 7월 10일 이전에 생성되었고 보관 기간을 수동으로 변경하지 않은 경우 보관 기간은 6년입니다. 구성된 보관 기간이 지나도 로그를 보관해야 하는 경우 정기적으로 로그를 다운로드하세요.

Cloud Scheduler 작업 만들기

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

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

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

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

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

스케줄러 작업 테스트

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

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

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

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

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

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

  9. 다음을 클릭합니다.

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

    • 스토리지 버킷 URL: Cloud Storage 버킷 URI를 입력합니다.

      gs://mulesoft-audit-logs/
      • mulesoft-audit-logs을 버킷의 실제 이름으로 바꿉니다.

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

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

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

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

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

  11. 다음을 클릭합니다.

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

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

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

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

  6. 저장을 클릭합니다.

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