收集 Atlassian Confluence 日志
解析器版本：5.0
支持的平台：
Google SecOps
SIEM
本文档介绍了如何将 Atlassian Confluence 日志注入到 Google Security Operations。解析器首先尝试使用专为 Atlassian Confluence 日志设计的正则表达式（Grok 模式）从原始日志消息中提取字段。如果 grok 解析失败或日志采用 JSON 格式，则代码会尝试将消息解析为 JSON。最后，提取的字段会映射到 Google SecOps UDM 架构，并添加更多上下文信息。
准备工作
请确保满足以下前提条件：
- Google SecOps 实例
- 具有审核日志访问权限的 Atlassian Confluence Cloud 账号，或者具有管理员权限的 Confluence Data Center/Server 账号
- 对于基于 GCP 的方法：对 GCP（GCS、IAM、Cloud Run、Pub/Sub、Cloud Scheduler）的特权访问权限
- 对于 Bindplane 方法：Windows Server 2016 或更高版本，或者具有 systemd 的 Linux 主机
集成选项概览
本指南提供了两种集成途径：
- 选项 1：通过 Bindplane + Syslog 收集 Confluence Data Center/Server 数据
- 方式 2：通过 GCP Cloud Run 函数 + GCS 获取 Confluence Cloud 审核日志（JSON 格式）
请选择最适合您的 Confluence 部署类型和基础架构的选项。
选项 1：通过 Bindplane + Syslog 收集 Confluence Data Center/Server 日志
此选项用于将 Confluence Data Center 或 Server 配置为通过 syslog 将日志发送到 Bindplane 代理，然后由该代理将日志转发到 Google SecOps。
获取 Google SecOps 注入身份验证文件
- 登录 Google SecOps 控制台。
- 依次前往 SIEM 设置 > 收集代理。
- 点击下载以下载数据注入身份验证文件。
将文件安全地保存在将要安装 Bindplane 代理的系统上。
获取 Google SecOps 客户 ID
- 登录 Google SecOps 控制台。
- 依次前往 SIEM 设置 > 配置文件。
复制并保存组织详细信息部分中的客户 ID。
安装 Bindplane 代理
按照以下说明在 Windows 或 Linux 操作系统上安装 Bindplane 代理。
Windows 安装
- 以管理员身份打开命令提示符或 PowerShell。
运行以下命令：
msiexec /i "https://github.com/observIQ/bindplane-otel-collector/releases/latest/download/observiq-otel-collector.msi" /quiet等待安装完成。
运行以下命令，验证安装是否成功：
sc query observiq-otel-collector
该服务应显示为 正在运行。
Linux 安装
- 打开具有 root 或 sudo 权限的终端。
运行以下命令：
sudo sh -c "$(curl -fsSlL https://github.com/observIQ/bindplane-otel-collector/releases/latest/download/install_unix.sh)" install_unix.sh等待安装完成。
运行以下命令，验证安装是否成功：
sudo systemctl status observiq-otel-collector
该服务应显示为有效（正在运行）。
其他安装资源
如需了解其他安装选项和问题排查信息，请参阅 Bindplane 代理安装指南。
配置 Bindplane 代理以注入 syslog 并将其发送到 Google SecOps
找到配置文件
Linux：
sudo systemctl status observiq-otel-collectorWindows：
notepad "C:\Program Files\observIQ OpenTelemetry Collector\config.yaml"
修改配置文件
将
config.yaml的全部内容替换为以下配置：receivers: udplog: listen_address: "0.0.0.0:514" exporters: chronicle/confluence_logs: compression: gzip creds_file_path: '/etc/bindplane-agent/ingestion-auth.json' customer_id: 'YOUR_CUSTOMER_ID' endpoint: malachiteingestion-pa.googleapis.com log_type: ATLASSIAN_CONFLUENCE raw_log_field: body ingestion_labels: service: confluence service: pipelines: logs/confluence: receivers: - udplog exporters: - chronicle/confluence_logs
配置参数
替换以下占位符：
listen_address：根据您的基础架构需要替换端口和 IP 地址。使用0.0.0.0:514在端口 514 上监听所有接口。creds_file_path：更新为身份验证文件的保存路径：- Linux：
/etc/bindplane-agent/ingestion-auth.json - Windows：
C:\Program Files\observIQ OpenTelemetry Collector\ingestion-auth.json
- Linux：
customer_id：将YOUR_CUSTOMER_ID替换为上一步中的实际客户 ID。endpoint：区域端点网址：- 美国：
malachiteingestion-pa.googleapis.com - 欧洲：
europe-malachiteingestion-pa.googleapis.com - 亚洲：
asia-southeast1-malachiteingestion-pa.googleapis.com
- 美国：
保存配置文件
修改后，保存文件：
- Linux：依次按
Ctrl+O、Enter和Ctrl+X - Windows：依次点击文件 > 保存
重启 Bindplane 代理以应用更改
如需在 Linux 中重启 Bindplane 代理，请执行以下操作：
运行以下命令：
sudo systemctl restart observiq-otel-collector验证服务是否正在运行：
sudo systemctl status observiq-otel-collector检查日志是否存在错误：
sudo journalctl -u observiq-otel-collector -f
如需在 Windows 中重启 Bindplane 代理，请执行以下操作：
请从下列选项中选择一项：
- 以管理员身份运行命令提示符或 PowerShell：
net stop observiq-otel-collector && net start observiq-otel-collector- 服务控制台：
- 按
Win+R，输入services.msc，然后按 Enter 键。 - 找到 observIQ OpenTelemetry 收集器。
- 右键点击并选择重新启动。
- 按
验证服务是否正在运行：
sc query observiq-otel-collector检查日志是否存在错误：
type "C:\Program Files\observIQ OpenTelemetry Collector\log\collector.log"
在 Confluence Data Center/Server 上配置 Syslog 转发
方案 A：配置 rsyslog 以转发本地日志文件（推荐）
- 将 Confluence 配置为将日志写入文件（默认行为）。
如果未安装 rsyslog，请进行安装：
sudo apt-get install rsyslog # Debian/Ubuntu sudo yum install rsyslog # RHEL/CentOS创建 rsyslog 配置文件
/etc/rsyslog.d/confluence.conf：# Forward Confluence logs to Bindplane $ModLoad imfile # Application logs $InputFileName /opt/atlassian/confluence/logs/atlassian-confluence.log $InputFileTag confluence-app: $InputFileStateFile stat-confluence-app $InputFileSeverity info $InputFileFacility local0 $InputRunFileMonitor # Audit logs (JSON format in DC/Server) $InputFileName <confluence-home-directory>/log/audit/audit.log $InputFileTag confluence-audit: $InputFileStateFile stat-confluence-audit $InputFileSeverity info $InputFileFacility local1 $InputRunFileMonitor # Forward to Bindplane agent *.* @@BINDPLANE_AGENT_IP:514将
BINDPLANE_AGENT_IP替换为 Bindplane 代理的 IP 地址（例如192.168.1.100）。根据您的 Confluence 安装情况调整日志文件路径：
- 应用日志通常：
<confluence-install>/logs/或<local-home>/logs/ - 审核日志：
<confluence-home-directory>/log/audit/（JSON 格式） - 如需查找 Confluence 主目录，请依次前往设置 > 常规配置 > 系统信息，然后查找 Confluence 主目录或本地主目录。
- 应用日志通常：
重启 rsyslog：
sudo systemctl restart rsyslog
选项 B：配置 Log4j2 Syslog 转发
- 通过 SSH 或 RDP 登录您的 Confluence 服务器。
找到以下位置的 Log4j2 配置文件：
<confluence-install>/confluence/WEB-INF/classes/log4j2.xml修改配置文件以添加 Syslog appender：
<Configuration> <Appenders> <!-- Existing appenders --> <Syslog name="SyslogAppender" host="BINDPLANE_AGENT_IP" port="514" protocol="UDP" format="RFC 5424" facility="LOCAL0"> <PatternLayout pattern="%d{ISO8601} %p [%t] [%c{1}] %m%n"/> </Syslog> </Appenders> <Loggers> <Root level="info"> <AppenderRef ref="SyslogAppender"/> <!-- Other appender refs --> </Root> <!-- Audit logger --> <Logger name="com.atlassian.confluence.event.events.security.AuditEvent" level="info" additivity="false"> <AppenderRef ref="SyslogAppender"/> </Logger> </Loggers> </Configuration>将
BINDPLANE_AGENT_IP替换为 Bindplane 代理的 IP 地址（例如192.168.1.100）。重启 Confluence 以应用更改：
sudo systemctl restart confluence
方案 2：通过 GCP Cloud Run 函数和 GCS 获取 Confluence Cloud 审核日志
收集 Confluence Cloud API 凭据
- 登录您的 Atlassian 账号。
- 前往 https://id.atlassian.com/manage-profile/security/api-tokens。
- 点击创建 API 令牌。
- 为令牌输入标签（例如，
Google Security Operations Integration）。 - 点击创建。
- 复制并妥善保存 API 令牌。
- 记下您的 Confluence Cloud 网站网址（例如
https://yoursite.atlassian.net）。 记下您的 Atlassian 账号电子邮件地址（用于身份验证）。
验证权限
如需验证账号是否具有所需权限，请执行以下操作：
- 登录 Confluence Cloud。
- 点击右上角的设置图标。
- 如果您可以在左侧导航栏中看到监控 > 审核日志，则说明您拥有所需的权限。
- 如果您看不到此选项，请与您的管理员联系，以获取 Confluence 管理员权限。
测试 API 访问权限
在继续进行集成之前，请先测试您的凭据：
# Replace with your actual credentials CONFLUENCE_EMAIL="your-email@example.com" CONFLUENCE_API_TOKEN="your-api-token" CONFLUENCE_URL="https://yoursite.atlassian.net" # Test API access curl -u "${CONFLUENCE_EMAIL}:${CONFLUENCE_API_TOKEN}" \ -H "Accept: application/json" \ "${CONFLUENCE_URL}/wiki/rest/api/audit"
创建 Google Cloud Storage 存储桶
- 前往 Google Cloud 控制台。
- 选择您的项目或创建新项目。
- 在导航菜单中，依次前往 Cloud Storage > 存储分区。
- 点击创建存储分区。
提供以下配置详细信息：
设置 值 为存储桶命名 输入一个全局唯一的名称（例如 confluence-audit-logs）位置类型 根据您的需求进行选择（区域级、双区域、多区域） 位置 选择营业地点（例如 us-central1）存储类别 标准（建议用于经常访问的日志） 访问权限控制 均匀（推荐） 保护工具 可选：启用对象版本控制或保留政策 点击创建。
为 Cloud Run 函数创建服务账号
Cloud Run 函数需要一个服务账号，该账号具有写入 GCS 存储桶的权限，并且可以由 Pub/Sub 调用。
创建服务账号
- 在 GCP 控制台中，依次前往 IAM 和管理 > 服务账号。
- 点击创建服务账号。
- 提供以下配置详细信息：
- 服务账号名称：输入
confluence-audit-collector-sa。 - 服务账号说明：输入
Service account for Cloud Run function to collect Confluence Cloud audit logs。
- 服务账号名称：输入
- 点击创建并继续。
- 在向此服务账号授予对项目的访问权限部分中，添加以下角色：
- 点击选择角色。
- 搜索并选择 Storage Object Admin。
- 点击 + 添加其他角色。
- 搜索并选择 Cloud Run Invoker。
- 点击 + 添加其他角色。
- 搜索并选择 Cloud Functions Invoker。
- 点击继续。
- 点击完成。
必须拥有这些角色，才能：
- Storage Object Admin：将日志写入 GCS 存储桶并管理状态文件
- Cloud Run Invoker：允许 Pub/Sub 调用函数
- Cloud Functions Invoker：允许调用函数
授予对 GCS 存储桶的 IAM 权限
向服务账号授予对 GCS 存储桶的写入权限：
- 前往 Cloud Storage > 存储分区。
- 点击您的存储桶名称。
- 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息：
- 添加主账号：输入服务账号电子邮件地址（例如
confluence-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com）。 - 分配角色：选择 Storage Object Admin。
- 添加主账号：输入服务账号电子邮件地址（例如
- 点击保存。
创建 Pub/Sub 主题
创建一个 Pub/Sub 主题，Cloud Scheduler 将向该主题发布消息，而 Cloud Run 函数将订阅该主题。
- 在 GCP 控制台中，前往 Pub/Sub > 主题。
- 点击创建主题。
- 提供以下配置详细信息：
- 主题 ID：输入
confluence-audit-trigger。 - 将其他设置保留为默认值。
- 主题 ID：输入
- 点击创建。
创建 Cloud Run 函数以收集日志
Cloud Run 函数将由来自 Cloud Scheduler 的 Pub/Sub 消息触发，以从 Confluence Cloud Audit API 中提取日志并将其写入 GCS。
- 在 GCP 控制台中，前往 Cloud Run。
- 点击创建服务。
- 选择函数（使用内嵌编辑器创建函数）。
在配置部分中，提供以下配置详细信息：
设置 值 Service 名称 confluence-audit-collector区域 选择与您的 GCS 存储桶匹配的区域（例如 us-central1）运行时 选择 Python 3.12 或更高版本 在触发器（可选）部分中：
- 点击 + 添加触发器。
- 选择 Cloud Pub/Sub。
- 在选择 Cloud Pub/Sub 主题中，选择
confluence-audit-trigger。 - 点击保存。
在身份验证部分中：
- 选择需要进行身份验证。
- 检查 Identity and Access Management (IAM)。
向下滚动并展开容器、网络、安全性。
前往安全性标签页：
- 服务账号：选择
confluence-audit-collector-sa。
- 服务账号：选择
前往容器标签页：
- 点击变量和密钥。
- 为每个环境变量点击+ 添加变量：
变量名称 示例值 说明 GCS_BUCKETconfluence-audit-logsGCS 存储桶名称 GCS_PREFIXconfluence-audit日志文件的前缀 STATE_KEYconfluence-audit/state.json状态文件路径 CONFLUENCE_URLhttps://yoursite.atlassian.netConfluence 网站网址 CONFLUENCE_EMAILyour-email@example.comAtlassian 账号电子邮件地址 CONFLUENCE_API_TOKENyour-api-token-hereAPI 令牌 MAX_RECORDS1000每次运行的记录数上限 在变量和 Secret 部分中，向下滚动到请求：
- 请求超时：输入
600秒（10 分钟）。
- 请求超时：输入
前往设置标签页：
- 在资源部分中：
- 内存：选择 512 MiB 或更高值。
- CPU：选择 1。
- 在资源部分中：
在修订版本伸缩部分中：
- 实例数下限：输入
0。 - 实例数上限：输入
100（或根据预期负载进行调整）。
- 实例数下限：输入
点击创建。
等待服务创建完成（1-2 分钟）。
创建服务后，系统会自动打开内嵌代码编辑器。
添加函数代码
- 在函数入口点中输入 main
在内嵌代码编辑器中，创建两个文件：
- main.py:
import functions_framework from google.cloud import storage import json import os import urllib3 from datetime import datetime, timezone, timedelta import time import base64 # 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', 'confluence-audit/') STATE_KEY = os.environ.get('STATE_KEY', 'confluence-audit/state.json') CONFLUENCE_URL = os.environ.get('CONFLUENCE_URL') CONFLUENCE_EMAIL = os.environ.get('CONFLUENCE_EMAIL') CONFLUENCE_API_TOKEN = os.environ.get('CONFLUENCE_API_TOKEN') MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '1000')) def to_unix_millis(dt: datetime) -> int: """Convert datetime to Unix epoch milliseconds.""" if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) dt = dt.astimezone(timezone.utc) return int(dt.timestamp() * 1000) 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) @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch Confluence Cloud audit logs and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ if not all([GCS_BUCKET, CONFLUENCE_URL, CONFLUENCE_EMAIL, CONFLUENCE_API_TOKEN]): print('Error: Missing required environment variables') return try: # Get GCS bucket bucket = storage_client.bucket(GCS_BUCKET) # Load state state = load_state(bucket, STATE_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"]) # Overlap by 2 minutes to catch any delayed events 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=24) print(f"Fetching logs from {last_time.isoformat()} to {now.isoformat()}") # Convert to Unix milliseconds start_millis = to_unix_millis(last_time) end_millis = to_unix_millis(now) # Fetch logs records, newest_event_time = fetch_logs( api_base=CONFLUENCE_URL, email=CONFLUENCE_EMAIL, api_token=CONFLUENCE_API_TOKEN, start_time_ms=start_millis, end_time_ms=end_millis, max_records=MAX_RECORDS, ) if not records: print("No new log records found.") save_state(bucket, STATE_KEY, now.isoformat()) return # Write to GCS as NDJSON timestamp = now.strftime('%Y%m%d_%H%M%S') object_key = f"{GCS_PREFIX}/logs_{timestamp}.ndjson" blob = bucket.blob(object_key) ndjson = '\n'.join([json.dumps(record, ensure_ascii=False) for record in records]) + '\n' blob.upload_from_string(ndjson, content_type='application/x-ndjson') print(f"Wrote {len(records)} records to gs://{GCS_BUCKET}/{object_key}") # Update state with newest event time if newest_event_time: save_state(bucket, STATE_KEY, newest_event_time) else: save_state(bucket, STATE_KEY, now.isoformat()) print(f"Successfully processed {len(records)} records") except Exception as e: print(f'Error processing 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}") 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}") def fetch_logs(api_base: str, email: str, api_token: str, start_time_ms: int, end_time_ms: int, max_records: int): """ Fetch logs from Confluence Cloud Audit API with pagination and rate limiting. Args: api_base: Confluence site URL email: Atlassian account email api_token: API token start_time_ms: Start time in Unix milliseconds end_time_ms: End time in Unix milliseconds max_records: Maximum total records to fetch Returns: Tuple of (records list, newest_event_time ISO string) """ # Clean up URL base_url = api_base.rstrip('/') # Build authentication header auth_string = f"{email}:{api_token}" auth_bytes = auth_string.encode('utf-8') auth_b64 = base64.b64encode(auth_bytes).decode('utf-8') headers = { 'Authorization': f'Basic {auth_b64}', 'Accept': 'application/json', 'User-Agent': 'GoogleSecOps-ConfluenceCollector/1.0' } records = [] newest_time = None page_num = 0 backoff = 1.0 start_index = 0 while True: page_num += 1 if len(records) >= max_records: print(f"Reached max_records limit ({max_records})") break # Build request URL url = f"{base_url}/wiki/rest/api/audit?startDate={start_time_ms}&endDate={end_time_ms}&start={start_index}&limit=100" try: response = http.request('GET', url, headers=headers) # Handle rate limiting with exponential backoff if response.status == 429: retry_after = int(response.headers.get('Retry-After', str(int(backoff)))) print(f"Rate limited (429). Retrying after {retry_after}s...") time.sleep(retry_after) backoff = min(backoff * 2, 30.0) continue backoff = 1.0 if response.status != 200: print(f"HTTP Error: {response.status}") response_text = response.data.decode('utf-8') print(f"Response body: {response_text}") return [], None data = json.loads(response.data.decode('utf-8')) page_results = data.get('results', []) if not page_results: print(f"No more results (empty page)") break print(f"Page {page_num}: Retrieved {len(page_results)} events") records.extend(page_results) # Track newest event time for event in page_results: try: # creationDate is in Unix milliseconds event_time_ms = event.get('creationDate') if event_time_ms: event_dt = datetime.fromtimestamp(event_time_ms / 1000, tz=timezone.utc) event_time = event_dt.isoformat() if newest_time is None or parse_datetime(event_time) > parse_datetime(newest_time): newest_time = event_time except Exception as e: print(f"Warning: Could not parse event time: {e}") # Check for more results current_size = data.get('size', 0) if current_size < 100: print(f"Reached last page (size={current_size} < limit=100)") break start_index += current_size except Exception as e: print(f"Error fetching logs: {e}") return [], None print(f"Retrieved {len(records)} total records from {page_num} pages") return records[:max_records], newest_time- requirements.txt:
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0点击部署以保存并部署该函数。
等待部署完成（2-3 分钟）。
创建 Cloud Scheduler 作业
Cloud Scheduler 会定期向 Pub/Sub 主题发布消息，从而触发 Cloud Run 函数。
- 在 GCP Console 中，前往 Cloud Scheduler。
- 点击创建作业。
提供以下配置详细信息：
设置 值 名称 confluence-audit-collector-hourly区域 选择与 Cloud Run 函数相同的区域 频率 0 * * * *（每小时一次，整点时）时区 选择时区（建议选择 UTC） 目标类型 Pub/Sub 主题 选择 confluence-audit-trigger消息正文 {}（空 JSON 对象）点击创建。
时间表频率选项
根据日志量和延迟时间要求选择频次：
| 频率 | Cron 表达式 | 使用场景 |
|---|---|---|
| 每隔 5 分钟 | */5 * * * * |
大批量、低延迟 |
| 每隔 15 分钟 | */15 * * * * |
搜索量中等 |
| 每小时 | 0 * * * * |
标准（推荐） |
| 每 6 小时 | 0 */6 * * * |
低成交量、批处理 |
| 每天 | 0 0 * * * |
历史数据收集 |
测试集成
- 在 Cloud Scheduler 控制台中，找到您的作业。
- 点击强制运行以手动触发作业。
- 等待几秒钟。
- 前往 Cloud Run > 服务。
- 点击
confluence-audit-collector。 - 点击日志标签页。
验证函数是否已成功执行。查找：
Fetching logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00 Page 1: Retrieved X events Wrote X records to gs://bucket-name/prefix/logs_YYYYMMDD_HHMMSS.ndjson Successfully processed X records前往 Cloud Storage > 存储分区。
点击您的存储桶名称。
转到
confluence-audit/文件夹。验证是否已创建具有当前时间戳的新
.ndjson文件。
如果您在日志中看到错误，请执行以下操作：
- HTTP 401：检查环境变量中的 API 凭据
- HTTP 403：验证账号是否具有 Confluence 管理员权限
- HTTP 429：速率限制 - 函数将自动重试并进行退避
- 缺少环境变量：检查是否已设置所有必需的变量
在 Google SecOps 中配置 Feed 以注入 Confluence 日志
- 依次前往 SIEM 设置 > Feed。
- 点击添加新 Feed。
- 点击配置单个 Feed。
- 在 Feed 名称字段中，输入 Feed 的名称（例如
Confluence Cloud Audit Logs）。 - 选择 Google Cloud Storage V2 作为来源类型。
- 选择 Atlassian Confluence 作为日志类型。
点击获取服务账号。系统会显示一个唯一的服务账号电子邮件地址，例如：
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com复制此电子邮件地址。您将在下一步骤中用到它。
点击下一步。
为以下输入参数指定值：
- 存储桶网址：输入带有前缀路径的 GCS 存储桶 URI：
gs://confluence-audit-logs/confluence-audit/- 将
confluence-audit-logs：您的 GCS 存储桶名称。confluence-audit：存储日志的可选前缀/文件夹路径（留空表示根目录）。
示例：
- 根存储桶：
gs://company-logs/ - 带前缀：
gs://company-logs/confluence-audit/ - 使用子文件夹：
gs://company-logs/confluence/audit/
- 来源删除选项：根据您的偏好选择删除选项：
- 永不：转移后永不删除任何文件（建议用于测试）。
- 删除已转移的文件：在成功转移后删除文件。
- 删除已转移的文件和空目录：成功转移后删除文件和空目录。
- 文件存在时间上限：包含在过去指定天数内修改的文件。默认值为 180 天。
- 资产命名空间：资产命名空间。
- 注入标签：要应用于此 Feed 中事件的标签。
点击下一步。
在最终确定界面中查看新的 Feed 配置，然后点击提交。
向 Google SecOps 服务账号授予 IAM 权限
Google SecOps 服务账号需要您的 GCS 存储桶的 Storage Object Viewer 角色。
- 前往 Cloud Storage > 存储分区。
- 点击您的存储桶名称。
- 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息：
- 添加主账号：粘贴 Google SecOps 服务账号电子邮件地址。
- 分配角色：选择 Storage Object Viewer。
点击保存。
UDM 映射表
| 日志字段 | UDM 映射 | 逻辑 |
|---|---|---|
| 代理 | read_only_udm.network.http.user_agent | 从“代理”字段中获取的值。 |
| app_protocol | read_only_udm.network.application_protocol | 派生自“app_protocol”字段。如果“app_protocol”包含“HTTPS”“HTTP”“SSH”或“RDP”，则使用相应协议。否则，默认值为“UNKNOWN_APPLICATION_PROTOCOL”。 |
| app_protocol | read_only_udm.network.application_protocol_version | 值取自“app_protocol”字段。 |
| auditType.action | read_only_udm.security_result.action | 派生自“auditType.action”字段。如果“auditType.action”包含“successful”，则该值设置为“ALLOW”。如果包含“restricted”，则将值设置为“BLOCK”。 |
| auditType.action | read_only_udm.security_result.summary | 当“auditType”不为空且“auditType_area”为“SECURITY”时，从“auditType.action”字段中获取的值。 |
| auditType.actionI18nKey | read_only_udm.metadata.product_event_type | 当“auditType”不为空时，从“auditType.actionI18nKey”字段中获取的值。 |
| auditType.area | read_only_udm.security_result.detection_fields.value | 从“auditType.area”字段中获取值，并将其分配给检测字段的“value”字段，同时将“key”字段设置为“auditType area”。当“auditType”不为空时，系统会执行此映射。 |
| auditType.category | read_only_udm.security_result.category_details | 当“auditType”不为空时，从“auditType.category”字段中获取的值。 |
| auditType.categoryI18nKey | read_only_udm.security_result.detection_fields.value | 从“auditType.categoryI18nKey”字段中获取值，并将其分配给检测字段的“value”字段，同时将“key”字段设置为“auditType categoryI18nKey”。当“auditType”不为空时，系统会执行此映射。 |
| auditType.level | read_only_udm.security_result.detection_fields.value | 从“auditType.level”字段中获取值，并将其分配给检测字段的“value”字段，同时将“key”字段设置为“auditType level”。当“auditType”不为空时，系统会执行此映射。 |
| author.displayName | read_only_udm.principal.user.user_display_name | 取自“author.displayName”字段的值。 |
| author.externalCollaborator | read_only_udm.security_result.about.resource.attribute.labels.value | 从“author.externalCollaborator”字段中提取值，并将其分配给“key”字段设置为“externalCollaborator”的标签的“value”字段。 |
| author.id | read_only_udm.principal.user.userid | 当“author.type”为“user”且“principal_user_present”为“false”时，从“author.id”字段中获取的值。 |
| author.isExternalCollaborator | read_only_udm.security_result.about.resource.attribute.labels.value | 从“author.isExternalCollaborator”字段中提取值，并将其分配给“key”字段设置为“isExternalCollaborator”的标签的“value”字段。 |
| author.name | read_only_udm.principal.user.user_display_name | 当“author.type”为“user”且“principal_user_present”为“false”时，从“author.name”字段中获取的值。 |
| bytes_in | read_only_udm.network.received_bytes | 如果“bytes_in”字段包含数字，则取自该字段的值。否则，默认值为 0。 |
| 类别 | read_only_udm.security_result.category_details | 值取自“category”字段。 |
| changedValues | read_only_udm.principal.resource.attribute.labels | 遍历“changedValues”中的每个元素，并创建键为“changedValue [index] [key]”、值为“changedValues”数组中相应值的标签。 |
| 创建日期 | read_only_udm.metadata.event_timestamp | 从“creationDate”字段中获取的值，解析为 UNIX 或 UNIX_MS 时间戳。 |
| extraAttributes | read_only_udm.principal.resource.attribute.labels | 遍历“extraAttributes”中的每个元素，并根据“name”和“nameI18nKey”字段创建键，根据相应的“value”字段创建值。 |
| http_verb | read_only_udm.network.http.method | 值取自“http_verb”字段。 |
| ip | read_only_udm.target.ip | 从“ip”字段中获取的值。 |
| principal_host | read_only_udm.principal.hostname | 从“principal_host”字段中获取的值。 |
| referral_url | read_only_udm.network.http.referral_url | 从“referral_url”字段中获取的值。 |
| remoteAddress | read_only_udm.principal.ip | 从“remoteAddress”字段中获取的值，解析为 IP 地址。 |
| response_code | read_only_udm.network.http.response_code | 从“response_code”字段中获取的值。 |
| session_duration | read_only_udm.additional.fields.value.string_value | 从“session_duration”字段中提取的值，并分配给“key”字段设置为“Session Duration”的标签的“string_value”字段。 |
| 来源 | read_only_udm.principal.ip | 从“来源”字段中获取的值，解析为 IP 地址。 |
| src_ip | read_only_udm.principal.ip | 如果“remoteAddress”为空，则取自“src_ip”字段的值。 |
| 摘要 | read_only_udm.security_result.summary | 值取自“摘要”字段。 |
| sysAdmin | read_only_udm.security_result.about.resource.attribute.labels.value | 从“sysAdmin”字段中获取值，并将其分配给“key”字段设置为“sysAdmin”的标签的“value”字段。 |
| superAdmin | read_only_udm.security_result.about.resource.attribute.labels.value | 从“superAdmin”字段中获取值，并将其分配给“key”字段设置为“superAdmin”的标签的“value”字段。 |
| target_url | read_only_udm.target.url | 从“target_url”字段中获取的值。 |
| 时间戳 | read_only_udm.metadata.event_timestamp | 从“时间戳”字段中获取的值，解析为日期和时间字符串。 |
| user_id | read_only_udm.principal.user.userid | 值取自“user_id”字段。 |
| read_only_udm.metadata.event_type | 此字段的值由一系列检查确定，默认值为“GENERIC_EVENT”。系统会根据“principal_host”“user_id”“has_principal”和“author.type”等其他字段的存在情况和内容，将其设置为特定值，例如“NETWORK_HTTP”“USER_UNCATEGORIZED”或“STATUS_UPDATE”。 | |
| read_only_udm.metadata.vendor_name | 设置为“ATLASSIAN”。 | |
| read_only_udm.metadata.product_name | 设置为“CONFLUENCE”。 | |
| read_only_udm.metadata.log_type | 设置为“ATLASSIAN_CONFLUENCE”。 | |
| read_only_udm.principal.user.user_display_name | 此字段的值可能来自“author.displayName”或“affectedObject.name”，具体取决于上下文。 | |
| read_only_udm.target.process.pid | 此字段的值可以来自“principal_host”或“pid”，具体取决于上下文。 | |
| read_only_udm.principal.resource.attribute.labels | 此字段会填充从“affectedObjects”“changedValues”和“extraAttributes”等字段派生的各种标签。这些标签的键和值会根据这些字段的具体内容动态生成。 | |
affectedObjects.id", "affectedObjects.name", "affectedObjects.type |
principal.resource.attribute.labels |
从变更日志映射 |
affectedObjects.id |
principal.user.userid |
从变更日志映射 |
affectedObjects.name |
principal.user.user_display_name |
从变更日志映射 |
changedValues.from", "changedValues.i18nKey", "changedValues.key" and "changedValues.to |
principal.resource.attribute.labels |
从变更日志映射 |
extraAttributes.name", "extraAttributes.nameI18nKey" and "extraAttributes.value |
principal.resource.attribute.labels |
从变更日志映射 |
system |
principal.hostname |
从变更日志映射 |
session_duration |
additional fields |
从变更日志映射 |
更新日志
需要更多帮助？获得社区成员和 Google SecOps 专业人士的解答。