收集 1Password 审核事件日志
解析器版本:2.0
本文档介绍了如何使用 Cloud Storage V2 将 1Password 审核事件日志注入到 Google Security Operations。
1Password 是一款密码管理平台,可帮助团队安全地存储、共享和管理凭据、Secret 和敏感信息。1Password Events API 可用于访问审核事件数据,这些数据会捕获在 1Password Business 账号中执行的管理和政策操作。
准备工作
请确保满足以下前提条件:
- Google SecOps 实例
- 启用了 Cloud Storage API 的 Google Cloud 项目
- 创建和管理 Cloud Storage 存储分区的权限
- 管理 Cloud Storage 存储分区的 Identity and Access Management (IAM) 政策的权限
- 创建 Cloud Run 服务、Pub/Sub 主题和 Cloud Scheduler 作业的权限
- 1Password Business 账号
- 您在 1Password 账号中拥有所有者或管理员角色
创建 Cloud Storage 存储桶
- 前往 Google Cloud 控制台。
- 选择您的项目或创建新项目。
- 在导航菜单中,依次前往 Cloud Storage > 存储分区。
- 点击创建存储分区。
提供以下配置详细信息:
设置 值 为存储桶命名 输入一个全局唯一的名称(例如 onepassword-audit-secops-logs)位置类型 根据您的需求进行选择(区域级、双区域、多区域) 位置 选择营业地点(例如 us-central1)存储类别 标准(建议用于经常访问的日志) 访问权限控制 均匀(推荐) 保护工具 可选:启用对象版本控制或保留政策 点击创建。
收集 1Password 凭据
设置“活动报告”集成
- 在 1Password.com 上登录您的账号。
- 在边栏中选择集成。
- 如果您之前已设置其他集成,请在集成页面上选择目录。
- 在事件报告部分,选择其他。
- 在名称字段中,输入集成的名称(例如
Google SecOps Audit Events)。 - 点击添加集成。
创建不记名令牌
1Password 会直接进入不记名令牌设置。
提供以下配置详细信息:
- 令牌名称:输入一个描述性名称(例如
SecOps GCS Collector - Audit Events) - Expires After:选择令牌的过期时间:30 天、90 天或 180 天。默认值为 Never。
- 要报告的事件:选择令牌可以访问哪些事件。默认情况下,系统会选择所有事件类型。保持选中审核事件,并清除登录尝试和商品使用情况,以便令牌仅限定于审核事件。
- 令牌名称:输入一个描述性名称(例如
点击 Issue Token。
在保存令牌页面上,点击 Save in 1Password(在 1Password 中保存),选择要存储令牌的保险库,然后点击保存。该令牌会保存为 API 凭据项。
点击查看集成详情,确认集成处于有效状态。
确定 Events API 的基本网址
基本网址取决于托管 1Password 账号的服务器:
| 如果您的账号托管在 | Events API 的基本网址为 |
|---|---|
1password.com |
https://events.1password.com |
ent.1password.com |
https://events.ent.1password.com |
1password.ca |
https://events.1password.ca |
1password.eu |
https://events.1password.eu |
测试 API 访问权限
在继续进行集成之前,请先测试您的凭据:
# Replace with your actual bearer token and base URL BEARER_TOKEN="<your-bearer-token>" API_BASE="https://events.1password.com" # Test API access using the introspect endpoint curl -v \ -H "Authorization: Bearer $BEARER_TOKEN" \ "$API_BASE/api/v2/auth/introspect"
如果请求成功,系统会返回一个 JSON 对象,其中包含 features 字段,该字段会列出您的令牌可以访问的活动类型(例如 ["auditevents"])。
为 Cloud Run 函数创建服务账号
Cloud Run 函数需要一个服务账号,该账号具有写入 Cloud Storage 存储桶的权限,并且可以由 Pub/Sub 调用。
创建服务账号
- 在 GCP 控制台中,依次前往 IAM 和管理 > 服务账号。
- 点击创建服务账号。
- 提供以下配置详细信息:
- 服务账号名称:输入
onepassword-audit-collector-sa - 服务账号说明:输入
Service account for Cloud Run function to collect 1Password Audit Events logs
- 服务账号名称:输入
- 点击创建并继续。
- 在向此服务账号授予对项目的访问权限部分中,添加以下角色:
- 点击选择角色。
- 搜索并选择 Storage Object Admin。
- 点击 + 添加其他角色。
- 搜索并选择 Cloud Run Invoker。
- 点击 + 添加其他角色。
- 搜索并选择 Cloud Functions Invoker。
- 点击继续。
- 点击完成。
必须拥有这些角色,才能:
- Storage Object Admin:将日志写入 Cloud Storage 存储桶并管理状态文件
- Cloud Run Invoker:允许 Pub/Sub 调用该函数
- Cloud Functions Invoker:允许调用函数
授予对 Cloud Storage 存储桶的 IAM 权限
向服务账号授予对 Cloud Storage 存储桶的写入权限:
- 前往 Cloud Storage > 存储分区。
- 点击您的存储桶 (
onepassword-audit-secops-logs)。 - 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息:
- 添加主账号:输入
onepassword-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com - 分配角色:选择 Storage Object Admin
- 添加主账号:输入
- 点击保存。
创建 Pub/Sub 主题
创建一个 Pub/Sub 主题,Cloud Scheduler 将向该主题发布消息,而 Cloud Run 函数将订阅该主题。
- 在 GCP 控制台中,前往 Pub/Sub > 主题。
- 点击创建主题。
- 提供以下配置详细信息:
- 主题 ID:输入
onepassword-audit-trigger - 将其他设置保留为默认值
- 主题 ID:输入
- 点击创建。
创建 Cloud Run 函数以收集日志
Cloud Run 函数将由 Cloud Scheduler 中的 Pub/Sub 消息触发,以从 1Password Events API 中提取审核事件日志并将其写入 Cloud Storage。
- 在 GCP 控制台中,前往 Cloud Run。
- 点击创建服务。
- 选择函数(使用内嵌编辑器创建函数)。
在配置部分中,提供以下配置详细信息:
设置 值 Service 名称 onepassword-audit-collector区域 选择与您的 Cloud Storage 存储桶匹配的区域(例如 us-central1)运行时 选择 Python 3.12 或更高版本 在触发器(可选)部分中:
- 点击 + 添加触发器。
- 选择 Cloud Pub/Sub。
- 在选择 Cloud Pub/Sub 主题中,选择
onepassword-audit-trigger。 - 点击保存。
在身份验证部分中:
- 选择需要进行身份验证。
- 检查 Identity and Access Management (IAM)。
滚动到容器、网络、安全性并展开该部分。
前往安全性标签页:
- 服务账号:选择
onepassword-audit-collector-sa
- 服务账号:选择
前往容器标签页:
- 点击变量和密钥。
- 为每个环境变量点击+ 添加变量:
变量名称 示例值 说明 GCS_BUCKETonepassword-audit-secops-logsCloud Storage 存储桶名称 GCS_PREFIXonepassword-audit日志文件的前缀 STATE_KEYonepassword-audit-state.json状态文件路径,位于日志前缀之外 OP_API_BASEhttps://events.1password.com1Password Events API 基本网址 OP_BEARER_TOKEN<your-bearer-token>1Password Events API 令牌 MAX_RECORDS10000每次运行的记录数上限 PAGE_SIZE1000每页记录数(最多 1000 条) LOOKBACK_HOURS24初始回溯期(以小时为单位) 在变量和密钥部分中,前往请求:
- 请求超时:输入
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 json import os import re import urllib3 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() # Environment variables GCS_BUCKET = os.environ.get('GCS_BUCKET') GCS_PREFIX = os.environ.get('GCS_PREFIX', 'onepassword-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 is ingested as log data and then # deleted, which drops the cursor and replays the whole lookback window. STATE_KEY = os.environ.get('STATE_KEY', 'onepassword-audit-state.json') API_BASE = os.environ.get('OP_API_BASE') BEARER_TOKEN = os.environ.get('OP_BEARER_TOKEN') MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000')) # The Events API accepts a limit from 1 to 1000. PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '1000')) LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24')) # 1Password Events API v2 audit events endpoint AUDIT_EVENTS_PATH = '/api/v2/auditevents' class FetchError(Exception): """Raised when the 1Password Events API call fails. State must never be written on a failed fetch. Saving a rejected cursor back would wedge the collector: every later run would fail on page 1, ingest nothing, and still report success. """ def parse_datetime(value: str) -> datetime: """Parse a 1Password `timestamp` into an aware datetime. The API returns RFC 3339 with either a numeric offset and whole seconds (2023-03-15T16:33:50-03:00) or Z with nanosecond precision (2025-10-31T13:45:49.203617068Z). Fractional digits are trimmed to microseconds because fromisoformat accepts at most six. """ text = str(value).strip() if text.endswith('Z'): text = text[:-1] + '+00:00' text = re.sub(r'\.(\d{6})\d+', r'.\1', text) dt = datetime.fromisoformat(text) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt.astimezone(timezone.utc) @functions_framework.cloud_event def main(cloud_event): """Fetch 1Password audit events and write them to Cloud Storage as NDJSON.""" if not all([GCS_BUCKET, API_BASE, BEARER_TOKEN]): # 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 ' '(GCS_BUCKET, OP_API_BASE, OP_BEARER_TOKEN)') bucket = storage_client.bucket(GCS_BUCKET) state = load_state(bucket, STATE_KEY) now = datetime.now(timezone.utc) print('--- Processing endpoint: auditevents ---') saved_cursor = state.get('cursor_auditevents') seen_uuids = set(state.get('seen_uuids', [])) # If the cursor is gone, resume from the last event actually written instead # of replaying a blind 24 hours. LOOKBACK_HOURS applies only on a cold start. if saved_cursor: start_time = None elif state.get('last_event_time'): start_time = parse_datetime(state['last_event_time']) print(f'No cursor found. Resuming from the last recorded event at {start_time.isoformat()}') else: start_time = now - timedelta(hours=LOOKBACK_HOURS) print(f'Cold start. Fetching from {start_time.isoformat()}') records, last_cursor = fetch_endpoint( api_base=API_BASE, path=AUDIT_EVENTS_PATH, bearer_token=BEARER_TOKEN, saved_cursor=saved_cursor, start_time=start_time, page_size=PAGE_SIZE, max_records=MAX_RECORDS, ) # The cursor path can re-deliver events, so filter by uuid before writing. fresh = [r for r in records if str(r.get('uuid', '')) not in seen_uuids] print(f'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}/auditevents_{timestamp}.ndjson' blob = bucket.blob(object_key) ndjson = '\n'.join( json.dumps(r, ensure_ascii=False) for r in fresh ) + '\n' blob.upload_from_string( ndjson, content_type='application/x-ndjson' ) print(f'Wrote {len(fresh)} auditevents records ' f'to gs://{GCS_BUCKET}/{object_key}') # State is written only after any data is durably uploaded. new_state = dict(state) if last_cursor: new_state['cursor_auditevents'] = last_cursor newest = None for record in records: stamp = record.get('timestamp') if not stamp: continue try: parsed = parse_datetime(stamp) except ValueError: continue if newest is None or parsed > newest: newest = parsed if newest: new_state['last_event_time'] = newest.isoformat() # Retain the uuids of the newest batch so a cursor reset does not # re-ingest the events that straddle the resume point. new_state['seen_uuids'] = sorted( {str(r.get('uuid', '')) for r in records if r.get('uuid')} ) new_state['last_run'] = now.isoformat() save_state(bucket, STATE_KEY, new_state) 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 drop the cursor and replay the lookback window. """ 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 resumes from the stale cursor and duplicates it. """ blob = bucket.blob(key) blob.upload_from_string( json.dumps(state, indent=2), content_type='application/json', ) print(f"Saved state: cursor={state.get('cursor_auditevents')!r} " f"last_event_time={state.get('last_event_time')}") def fetch_endpoint(api_base, path, bearer_token, saved_cursor, start_time, page_size, max_records): """Fetch events from the 1Password Events API v2 audit events endpoint. The API uses cursor-based pagination with POST requests. The first request sends a ResetCursor object with an optional start_time and limit. Subsequent requests send the cursor returned by the previous response. The cursor is a persistent checkpoint and stays valid across runs. Args: api_base: Events API base URL path: Endpoint path bearer_token: JWT bearer token saved_cursor: Cursor from the previous run, or None start_time: Lower bound used only when there is no cursor page_size: Max events per page (1 to 1000) max_records: Max total events per run Returns: Tuple of (records list, cursor to persist). Raises: FetchError: on any API or transport failure, so that a rejected cursor is never written back to the state file. """ url = f'{api_base.rstrip("/")}{path}' headers = { 'Authorization': f'Bearer {bearer_token}', 'Content-Type': 'application/json', 'Accept': 'application/json', } records = [] cursor = saved_cursor 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: print(f'Reached max_records limit ({max_records})') break if cursor: # Continuing cursor: resume from the last position. body = json.dumps({'cursor': cursor}) else: # ResetCursor: first request, or the cursor was lost. reset = {'limit': page_size} if start_time: reset['start_time'] = start_time.strftime('%Y-%m-%dT%H:%M:%SZ') body = json.dumps(reset) try: response = http.request( 'POST', url, body=body, headers=headers, ) except Exception as e: raise FetchError(f'Request to {url} failed: {e}') from e # Documented limits are 600 requests per minute and 30,000 per hour. if response.status == 429: rate_limit_retries += 1 if rate_limit_retries > MAX_RATE_LIMIT_RETRIES: raise FetchError('Rate limited repeatedly; giving up without saving state') raw_retry_after = response.headers.get('Retry-After') try: retry_after = int(raw_retry_after) if raw_retry_after else int(backoff) except (TypeError, ValueError): retry_after = int(backoff) print(f'Rate limited (429). Retrying after {retry_after}s...') time.sleep(retry_after) backoff = min(backoff * 2, 60.0) continue backoff = 1.0 rate_limit_retries = 0 if response.status != 200: body_text = response.data.decode('utf-8') # A 400 or 401 here may mean the saved cursor was rejected. Raising # leaves the previous state intact so the run can be retried, rather # than persisting a cursor that stalls every future run. raise FetchError(f'HTTP {response.status} from the Events API: {body_text}') try: data = json.loads(response.data.decode('utf-8')) except json.JSONDecodeError as e: raise FetchError(f'Malformed JSON response from the Events API: {e}') from e page_items = data.get('items', []) cursor = data.get('cursor') has_more = data.get('has_more', False) if page_items: print(f'Page {page_num}: Retrieved ' f'{len(page_items)} events') records.extend(page_items) if not has_more: print('No more pages (has_more=false)') break if not cursor: print('No cursor returned, stopping') break if not page_items: # has_more can be true while a page is empty. Stop rather than spin. print('Empty page with has_more=true; resuming on the next run') break print(f'Retrieved {len(records)} total records ' f'from {page_num} pages') return records, cursor第二个文件 - 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。
- 点击创建作业。
提供以下配置详细信息:
设置 值 名称 onepassword-audit-collector-hourly区域 选择与 Cloud Run 函数相同的区域 频率 0 * * * *(每小时一次,整点时)时区 选择时区(建议选择 UTC) 目标类型 Pub/Sub 主题 选择 onepassword-audit-trigger消息正文 {}(空 JSON 对象)点击创建。
时间表频率选项
根据日志量和延迟时间要求选择频次:
| 频率 | Cron 表达式 | 使用场景 |
|---|---|---|
| 每隔 5 分钟 | */5 * * * * |
大批量、低延迟 |
| 每隔 15 分钟 | */15 * * * * |
搜索量中等 |
| 每小时 | 0 * * * * |
标准(推荐) |
| 每 6 小时 | 0 */6 * * * |
低成交量、批处理 |
| 每天 | 0 0 * * * |
历史数据收集 |
测试集成
- 在 Cloud Scheduler 控制台中,找到
onepassword-audit-collector-hourly。 - 点击强制运行以手动触发作业。
- 等待几秒钟。
- 前往 Cloud Run > 服务。
- 点击
onepassword-audit-collector。 - 点击日志标签页。
验证函数是否已成功执行。查找:
--- Processing endpoint: auditevents --- Page 1: Retrieved X events Fetched X records, Y new after deduplication Wrote Y auditevents records to gs://onepassword-audit-secops-logs/onepassword-audit/auditevents_YYYYMMDDTHHMMSSZ.ndjson Saved state: cursor='...' last_event_time=YYYY-MM-DDTHH:MM:SS+00:00 Successfully processed Y records前往 Cloud Storage > 存储分区。
点击
onepassword-audit-secops-logs。转到
onepassword-audit/文件夹。验证是否已创建具有当前时间戳的新
.ndjson文件。
如果您在日志中看到错误,请执行以下操作:
- HTTP 401:检查
OP_BEARER_TOKEN环境变量中的不记名令牌。令牌可能已过期。 - HTTP 429:速率限制。该函数会自动重试,并采用退避机制。1Password Events API 允许每分钟 600 个请求,每小时 30,000 个请求。
- 缺少环境变量:检查是否已在 Cloud Run 函数配置中设置所有必需的变量。
- 未返回任何记录:使用自检端点验证您的不记名令牌是否具有访问审核事件的权限。
- 同一事件反复到达:确认
STATE_KEY不在 Feed 读取的前缀内。如果状态文件被 Feed 删除,游标会丢失,收集器会从上次记录的事件重新读取。
检索 Google SecOps 服务账号
Google SecOps 使用唯一的服务账号从您的 Cloud Storage 存储桶中读取数据。您必须授予此服务账号对您的存储桶的访问权限。
获取服务账号电子邮件地址
- 依次前往 SIEM 设置 > Feed。
- 点击添加新 Feed。
- 点击配置单个 Feed。
- 在 Feed 名称字段中,输入 Feed 的名称(例如
1Password Audit Events Logs)。 - 选择 Google Cloud Storage V2 作为来源类型。
- 选择 1Password 审核事件作为日志类型。
- 点击获取服务账号。
系统会显示一个唯一的服务账号电子邮件地址,例如:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com复制此电子邮件地址,以便在下一步中使用。
点击下一步。
为以下输入参数指定值:
存储桶网址:输入带有前缀路径的 Cloud Storage 存储桶 URI:
gs://onepassword-audit-secops-logs/onepassword-audit/来源删除选项:根据您的偏好选择删除选项:
- 永不删除文件:永不从源中删除文件(建议用于测试)。
删除已转移的文件和空目录:在成功提取完成后,从来源中删除文件和空目录。
文件存在时间上限:包含在过去指定天数内修改过的文件(默认值为 180 天)
资产命名空间:资产命名空间
注入标签:要应用于此 Feed 中事件的标签
点击下一步。
在最终确定界面中查看新的 Feed 配置,然后点击提交。
向 Google SecOps 服务账号授予 IAM 权限
Google SecOps 服务账号需要您的 Cloud Storage 存储桶具备两个角色:用于读取日志对象的 Storage Object Viewer,以及用于读取存储桶元数据的存储桶级角色。
- 前往 Cloud Storage > 存储分区。
- 点击
onepassword-audit-secops-logs。 - 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息:
- 添加主账号:粘贴 Google SecOps 服务账号电子邮件地址
- 分配角色:选择以下两个角色:
- Storage Object Viewer:读取日志对象。
- Storage Legacy Bucket Reader:读取存储桶元数据。如果您选择了删除已转移的文件和空目录删除选项,请改为选择存储空间旧版存储分区写入者,该角色也会授予删除权限。
点击保存。
UDM 映射表
| 日志字段 | UDM 映射 | 逻辑 |
|---|---|---|
additional_action |
additional.fields |
已合并 |
aux_id_label |
additional.fields |
已合并 |
aux_info_list |
additional.fields |
已合并 |
object_type_label |
additional.fields |
已合并 |
EventTime |
metadata.event_timestamp |
解析为 yyyy-MM-dd HH:mm:ss |
expiry |
metadata.event_timestamp |
解析为 yyyy-MM-ddTHH:mm:ssZZZZZZZZZZ |
item.session.login_time |
metadata.event_timestamp |
解析为 yyyy-MM-ddTHH:mm:ssZZZZZZZZZZ |
item.timestamp |
metadata.event_timestamp |
解析为 ISO8601 |
has_principal |
metadata.event_type |
已映射:true → USER_UNCATEGORIZED、true → STATUS_UPDATE |
item.session.uuid |
network.session_id |
直接映射 |
deviceUuid |
principal.asset.asset_id |
直接映射 |
session_ip |
principal.asset.ip |
已合并 |
item.actor_details.email |
principal.email |
直接映射 |
session_ip |
principal.ip |
已合并 |
item.location.city |
principal.location.city |
直接映射 |
item.location.country |
principal.location.country_or_region |
直接映射 |
item.location.region |
principal.location.name |
直接映射 |
item.location.latitude |
principal.location.region_latitude |
直接映射 |
item.location.longitude |
principal.location.region_longitude |
直接映射 |
uuid_label |
principal.resource.attribute.labels |
已合并 |
item.actor_details.name |
principal.user.user_display_name |
直接映射 |
item.actor_details.uuid |
principal.user.userid |
直接映射 |
actor_uuid_label |
security_result.about.resource.attribute.labels |
已合并 |
aux_uid_label |
security_result.about.resource.attribute.labels |
已合并 |
object_id_label |
security_result.about.resource.attribute.labels |
已合并 |
item.session.device_uuid |
target.resource.product_object_id |
直接映射 |
| 不适用 | metadata.event_type |
常量:USER_UNCATEGORIZED |
| 不适用 | metadata.product_name |
常量:ONEPASSWORD_AUDIT_EVENTS |
| 不适用 | metadata.vendor_name |
常量:ONEPASSWORD |
更新日志
需要更多帮助?获得社区成员和 Google SecOps 专业人士的解答。