Google 会使用 AI 技术将内容翻译成您偏好的语言。AI 翻译可能包含错误。

收集 MuleSoft Anypoint 平台日志

支持：

Google SecOps SIEM

本文档介绍了如何使用 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 平台的特权访问权限

获取 MuleSoft 组织 ID

登录 Anypoint Platform。
前往访问权限管理 > 组织。
在商家群组表格中，点击您组织的名称。
复制组织 ID（例如 0a12b3c4-d5e6-789f-1021-1a2b34cd5e6f）。

或者，前往 MuleSoft 业务群组，然后从网址中复制 ID。

创建 MuleSoft 关联的应用

登录 Anypoint Platform。
依次前往访问权限管理 > 关联的应用 > 创建应用。
提供以下配置详细信息：
- 应用名称：输入一个唯一名称（例如 Google SecOps export）。
- 选择应用自行操作（客户端凭证）。
依次点击添加范围 > 审核日志查看器 > 下一步。
选择您需要其日志的每个商家群组。
依次点击下一步 > 添加范围。
点击保存，然后复制客户端 ID 和客户端密钥。

注意：根据组织权限，可能需要其他范围。如果您遇到权限错误，请验证关联的应用是否已获得对所有必需商家群组的访问权限。

创建 Cloud Storage 存储桶

前往 Google Cloud 控制台。
选择您的项目或创建一个新项目。
在导航菜单中，依次前往 Cloud Storage > 存储桶。
点击创建存储分区。

提供以下配置详细信息：

设置	值
指定存储桶的名称	输入一个全局唯一的名称（例如 `mulesoft-audit-logs`）
位置类型	根据您的需求进行选择（单区域、双区域、多区域）
位置	选择位置（例如 `us-central1`）
存储类别	标准（推荐用于频繁访问的日志）
访问权限控制	均匀（推荐）
保护工具	可选：启用对象版本控制或保留政策

点击创建。

为 Cloud Run 函数创建服务账号

Cloud Run 函数需要一个具有写入 Cloud Storage 存储桶权限的服务账号。

创建服务账号

在 GCP 控制台中，依次前往 IAM 和管理 > 服务账号。
点击创建服务账号。
提供以下配置详细信息：
- 服务账号名称：输入 mulesoft-logs-collector-sa。
- 服务账号说明：输入 Service account for Cloud Run function to collect MuleSoft Anypoint logs。
点击创建并继续。
在向此服务账号授予对项目的访问权限部分中：
1. 点击选择角色。
2. 搜索并选择 Storage Object Admin。
3. 点击 + 添加其他角色。
4. 搜索并选择 Cloud Run Invoker。
5. 点击 + 添加其他角色。
6. 搜索并选择 Cloud Functions Invoker。
点击继续。
点击完成。

以下操作需要这些角色：

Storage Object Admin：将日志写入 Cloud Storage 存储桶并管理状态文件
Cloud Run Invoker：允许 Pub/Sub 调用函数
Cloud Functions Invoker：允许调用函数

授予对 Cloud Storage 存储桶的 IAM 权限

向服务账号授予 Cloud Storage 存储桶的 write 权限，如下所示：

前往 Cloud Storage > 存储桶。
点击您的存储桶名称。
前往权限标签页。
点击授予访问权限。
提供以下配置详细信息：
- 添加主账号：输入服务账号电子邮件地址（例如 mulesoft-logs-collector-sa@PROJECT_ID.iam.gserviceaccount.com）。
- 分配角色：选择 Storage Object Admin。
点击保存。

创建 Pub/Sub 主题

创建一个 Pub/Sub 主题，Cloud Scheduler 将向该主题发布消息，Cloud Run 函数将订阅该主题，具体如下所示：

在 GCP 控制台中，前往 Pub/Sub > 主题。
点击创建主题。
提供以下配置详细信息：
- 主题 ID：输入 mulesoft-audit-trigger。
- 其他设置均保留默认值。
点击创建。

创建 Cloud Run 函数以收集日志

Cloud Run 函数将由来自 Cloud Scheduler 的 Pub/Sub 消息触发，以从 MuleSoft Anypoint API 提取日志并将其写入 Cloud Storage。

在 GCP 控制台中，前往 Cloud Run。
点击编写函数。
在配置部分中，提供以下配置详细信息：

设置值

Service 名称 mulesoft-audit-collector

区域选择与您的 Cloud Storage 存储桶匹配的区域（例如 us-central1）

运行时 选择 Python 3.12 或更高版本
在触发器部分中，点击添加触发器，然后选择 Pub/Sub 触发器。
在 Eventarc 触发器窗格中，提供以下配置详细信息：
- 触发器名称：保留生成的名称或输入触发器的名称。
- 触发器类型：选择 Google 来源。
- 事件提供方：选择 Pub/Sub。
- 事件类型：选择 google.cloud.pubsub.topic.v1.messagePublished。
- 选择 Cloud Pub/Sub 主题：选择主题 mulesoft-audit-trigger。
- 区域：选择与函数相同的区域。
- 服务账号：选择服务账号 mulesoft-logs-collector-sa。
点击保存触发器。

注意：触发器选项来自 Eventarc。如果未列出 Pub/Sub 触发器，请为项目启用 Eventarc API，然后重新加载页面。如果该窗格报告服务账号缺少 Eventarc Event Receiver 或 Cloud Run Invoker 角色，请点击授予。
在身份验证部分中：
1. 选择需要进行身份验证。
2. Identity and Access Management (IAM)。
注意：调用函数时，Pub/Sub 会自动处理身份验证。
找到并展开容器、网络、安全性。
前往安全标签页：
- 服务账号：选择 mulesoft-logs-collector-sa 服务账号。

设置	值
Service 名称	`mulesoft-audit-collector`
区域	选择与您的 Cloud Storage 存储桶匹配的区域（例如 `us-central1`）
运行时	选择 Python 3.12 或更高版本

前往容器标签页：

点击变量和密钥。
为每个环境变量点击 + 添加变量：

变量名称	示例值	说明
`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`	首次运行时的初始回溯期

前往变量和 Secret 标签页，然后前往请求：
- 请求超时：输入 600 秒（10 分钟）。
前往容器中的设置标签页：
- 在资源部分中：
  - 内存：选择 512 MiB 或更高值。
  - CPU：选择 1。
- 点击完成。
前往执行环境：
- 选择默认（推荐）。
在修订版本伸缩部分中：
- 实例数下限：输入 0。
- 实例数上限：输入 100（或根据预期负载进行调整）。
点击创建。
等待服务创建完成（1-2 分钟）。
创建服务后，内嵌代码编辑器会自动打开。

添加函数代码

在函数入口点中输入 main

在内嵌代码编辑器中，创建两个文件：

第一个文件 - 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

点击部署以保存并部署函数。
等待部署完成（2-3 分钟）。

重要注意事项

速率限制：审核日志查询端点在三个控制平面中对每个 IP 应用速率限制。美国控制平面允许每个 IP 地址每分钟 700 个请求，而欧盟和政府控制平面允许每个 IP 地址每分钟 40 个请求。该函数实现了指数退避算法，可自动处理速率限制。
令牌过期：访问令牌通常会在签发后约 30 到 60 分钟过期。该函数每次执行都会请求一个新令牌，这完全在上述时间范围内。
去重和状态：该函数会在状态文件中保留最新事件时间的水印和重叠窗口内事件的哈希值。每次运行都会重新读取水印前两分钟的数据，以捕获延迟记录的条目，并舍弃已写入的条目，因此重试、手动运行或调度程序延迟不会产生重复的事件。如果 API 调用失败，运行也会失败，并且水位保持不变，因此下一次运行会重试同一时间窗口。
审核日志保留：审核日志的默认保留期限为一年。如果您的组织是在 2023 年 7 月 10 日之前创建的，并且您未手动更改保留期限，则保留期限为 6 年。如果您需要将日志保留到配置的保留期限之后，请定期下载日志。

创建 Cloud Scheduler 作业

Cloud Scheduler 将定期向 Pub/Sub 主题发布消息，从而触发 Cloud Run 函数。

在 GCP 控制台中，前往 Cloud Scheduler。
点击创建作业。

提供以下配置详细信息：

设置	值
名称	`mulesoft-audit-collector-hourly`
区域	选择与 Cloud Run 函数相同的区域
频率	`0 * * * *`（每小时整点）
时区	选择时区（建议选择世界协调时间 [UTC]）
目标类型	Pub/Sub
主题	选择主题 `mulesoft-audit-trigger`
消息正文	`{}`（空 JSON 对象）

点击创建。

测试调度器作业

在 Cloud Scheduler 控制台中，找到您的作业。
点击强制运行以手动触发。
等待几秒钟，然后前往 Cloud Run > 服务 > mulesoft-audit-collector > 日志。
验证函数是否已成功执行。
检查 Cloud Storage 存储桶，确认日志已写入。

在 Google SecOps 中配置 Feed 以注入 MuleSoft 日志

前往 SIEM 设置 > Feed。
点击添加新 Feed。
点击配置单个 Feed。
在 Feed 名称字段中，输入 Feed 的名称（例如 MuleSoft Logs）。
选择 Google Cloud Storage V2 作为来源类型。
选择 Mulesoft 作为日志类型。
点击获取服务账号。系统会显示一个专属的服务账号邮箱，例如：
```
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
```
复制此电子邮件地址。您将在下一步骤中用到它。

注意：每个 Google SecOps 实例都有一个唯一服务账号。请勿使用其他文档或示例中的服务账号。
点击下一步。
为以下输入参数指定值：
- 存储桶网址：输入 Cloud Storage 存储桶 URI：
```
gs://mulesoft-audit-logs/
```
  - 将 mulesoft-audit-logs 替换为存储桶的实际名称。
  注意：请务必在 URI 末尾添加尾随斜杠 (/)。
- 来源删除选项：根据您的偏好选择删除选项：
  - 永不删除文件：永不从来源中删除文件（建议在测试时使用）。
  - 删除已转移的文件和空目录：在成功完成提取后，从来源位置删除文件和空目录。
- 文件存在时间上限：包含过去指定天数内修改过的文件。默认值为 180 天。
- 资产命名空间：资产命名空间。
- 注入标签：要应用于此 Feed 中事件的标签。
点击下一步。
在最终确定界面中查看新的新闻信息流配置，然后点击提交。

向 Google SecOps 服务账号授予 IAM 权限

Google SecOps 服务账号需要对您的 Cloud Storage 存储桶拥有两个角色：用于读取日志对象的 Storage Object Viewer，以及用于读取存储桶元数据的存储桶级角色。

前往 Cloud Storage > 存储桶。
点击您的存储桶名称。
前往权限标签页。
点击授予访问权限。
提供以下配置详细信息：
- 添加主账号：粘贴 Google SecOps 服务账号电子邮件地址。
- 分配角色：选择以下两个角色：
- Storage Object Viewer：读取日志对象。
- Storage Legacy Bucket Reader：读取存储桶元数据。如果您选择了删除已转移的文件和空目录删除选项，请改为选择存储旧版存储桶写入者，该角色也会授予删除权限。
点击保存。

注意：请同时授予这两个角色。Storage Object Viewer 不会授予 Feed 所需的存储桶级层权限，而 Storage Legacy Bucket 角色不会授予读取对象内容的权限。