Collect 1Password Audit Events logs
Parser Version: 2.0
This document explains how to ingest 1Password Audit Events logs to Google Security Operations using Cloud Storage V2.
1Password is a password management platform that helps teams securely store, share, and manage credentials, secrets, and sensitive information. The 1Password Events API provides access to audit event data that captures administrative and policy actions performed in your 1Password Business account.
Before you begin
Make sure you have the following prerequisites:
- A Google SecOps instance
- A Google Cloud project with Cloud Storage API enabled
- Permissions to create and manage Cloud Storage buckets
- Permissions to manage Identity and Access Management (IAM) policies on Cloud Storage buckets
- Permissions to create Cloud Run services, Pub/Sub topics, and Cloud Scheduler jobs
- A 1Password Business account
- An owner or administrator role in your 1Password account
Create a Cloud Storage bucket
- Go to the Google Cloud Console.
- Select your project or create a new one.
- In the navigation menu, go to Cloud Storage > Buckets.
- Click Create bucket.
Provide the following configuration details:
Setting Value Name your bucket Enter a globally unique name (for example, onepassword-audit-secops-logs)Location type Choose based on your needs (Region, Dual-region, Multi-region) Location Select the location (for example, us-central1)Storage class Standard (recommended for frequently accessed logs) Access control Uniform (recommended) Protection tools Optional: Enable object versioning or retention policy Click Create.
Collect 1Password credentials
Set up Events Reporting integration
- Sign in to your account on 1Password.com.
- Select Integrations in the sidebar.
- If you have previously set up other integrations, select Directory on the Integrations page.
- In the Events Reporting section, select Other.
- In the Name field, enter a name for the integration (for example,
Google SecOps Audit Events). - Click Add Integration.
Create a bearer token
1Password continues straight to the bearer token setup.
Provide the following configuration details:
- Token Name: Enter a descriptive name (for example,
SecOps GCS Collector - Audit Events) - Expires After: Choose when the token expires: 30 days, 90 days, or 180 days. The default is Never.
- Events to Report: Choose which events the token can access. All event types are selected by default. Keep Audit events selected and clear Sign-in attempts and Item usages so the token is scoped to audit events only.
- Token Name: Enter a descriptive name (for example,
Click Issue Token.
On the Save your token page, click Save in 1Password, choose the vault where you want to store the token, and then click Save. The token is saved as an API Credential item.
Click View Integration Details to confirm the integration is active.
Determine your Events API base URL
The base URL depends on the server that hosts your 1Password account:
| If your account is hosted on | Your Events API base URL is |
|---|---|
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 |
Test API access
Test your credentials before proceeding with the integration:
# 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"
A successful response returns a JSON object with the features field listing the event types your token can access (for example, ["auditevents"]).
Create a service account for the Cloud Run function
The Cloud Run function needs a service account with permissions to write to Cloud Storage bucket and be invoked by Pub/Sub.
Create the service account
- In the GCP Console, go to IAM & Admin > Service Accounts.
- Click Create Service Account.
- Provide the following configuration details:
- Service account name: Enter
onepassword-audit-collector-sa - Service account description: Enter
Service account for Cloud Run function to collect 1Password Audit Events logs
- Service account name: Enter
- Click Create and Continue.
- In the Grant this service account access to project section, add the following roles:
- Click Select a role.
- Search for and select Storage Object Admin.
- Click + Add another role.
- Search for and select Cloud Run Invoker.
- Click + Add another role.
- Search for and select Cloud Functions Invoker.
- Click Continue.
- Click Done.
These roles are required for:
- Storage Object Admin: Writes logs to Cloud Storage bucket and manage state files
- Cloud Run Invoker: Allows Pub/Sub to invoke the function
- Cloud Functions Invoker: Allows function invocation
Grant IAM permissions on the Cloud Storage bucket
Grant the service account write permissions on the Cloud Storage bucket:
- Go to Cloud Storage > Buckets.
- Click your bucket (
onepassword-audit-secops-logs). - Go to the Permissions tab.
- Click Grant access.
- Provide the following configuration details:
- Add principals: Enter
onepassword-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com - Assign roles: Select Storage Object Admin
- Add principals: Enter
- Click Save.
Create a Pub/Sub topic
Create a Pub/Sub topic that Cloud Scheduler will publish to and the Cloud Run function will subscribe to.
- In the GCP Console, go to Pub/Sub > Topics.
- Click Create topic.
- Provide the following configuration details:
- Topic ID: Enter
onepassword-audit-trigger - Leave other settings as default
- Topic ID: Enter
- Click Create.
Create a Cloud Run function to collect logs
The Cloud Run function will be triggered by Pub/Sub messages from Cloud Scheduler to fetch audit event logs from the 1Password Events API and write them to Cloud Storage.
- In the GCP Console, go to Cloud Run.
- Click Create service.
- Select Function (use an inline editor to create a function).
In the Configure section, provide the following configuration details:
Setting Value Service name onepassword-audit-collectorRegion Select region matching your Cloud Storage bucket (for example, us-central1)Runtime Select Python 3.12 or later In the Trigger (optional) section:
- Click + Add trigger.
- Select Cloud Pub/Sub.
- In Select a Cloud Pub/Sub topic, choose
onepassword-audit-trigger. - Click Save.
In the Authentication section:
- Select Require authentication.
- Check Identity and Access Management (IAM).
Scroll to and expand Containers, Networking, Security.
Go to the Security tab:
- Service account: Select
onepassword-audit-collector-sa
- Service account: Select
Go to the Containers tab:
- Click Variables & Secrets.
- Click + Add variable for each environment variable:
Variable Name Example Value Description GCS_BUCKETonepassword-audit-secops-logsCloud Storage bucket name GCS_PREFIXonepassword-auditPrefix for log files STATE_KEYonepassword-audit-state.jsonState file path, outside the log prefix OP_API_BASEhttps://events.1password.com1Password Events API base URL OP_BEARER_TOKEN<your-bearer-token>1Password Events API bearer token MAX_RECORDS10000Max records per run PAGE_SIZE1000Records per page (max 1000) LOOKBACK_HOURS24Initial lookback period in hours In the Variables & Secrets section, go to to Requests:
- Request timeout: Enter
600seconds (10 minutes)
- Request timeout: Enter
Go to the Settings tab:
- In the Resources section:
- Memory: Select 512 MiB or higher
- CPU: Select 1
- In the Resources section:
In the Revision scaling section:
- Minimum number of instances: Enter
0 - Maximum number of instances: Enter
100(or adjust based on expected load)
- Minimum number of instances: Enter
Click Create.
Wait for the service to be created (1-2 minutes).
After the service is created, the inline code editor will open automatically.
Add the function code
- Enter main in the Entry point field.
In the inline code editor, create two files:
First file - 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, cursorSecond file - requirements.txt:
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0
Click Deploy to save and deploy the function.
Wait for deployment to complete (2-3 minutes).
Create a Cloud Scheduler job
Cloud Scheduler will publish messages to the Pub/Sub topic at regular intervals, triggering the Cloud Run function.
- In the GCP Console, go to Cloud Scheduler.
- Click Create Job.
Provide the following configuration details:
Setting Value Name onepassword-audit-collector-hourlyRegion Select same region as Cloud Run function Frequency 0 * * * *(every hour, on the hour)Timezone Select timezone (UTC recommended) Target type Pub/Sub Topic Select onepassword-audit-triggerMessage body {}(empty JSON object)Click Create.
Schedule frequency options
Choose frequency based on log volume and latency requirements:
| Frequency | Cron Expression | Use Case |
|---|---|---|
| Every 5 minutes | */5 * * * * |
High-volume, low-latency |
| Every 15 minutes | */15 * * * * |
Medium volume |
| Every hour | 0 * * * * |
Standard (recommended) |
| Every 6 hours | 0 */6 * * * |
Low volume, batch processing |
| Daily | 0 0 * * * |
Historical data collection |
Test the integration
- In the Cloud Scheduler console, find
onepassword-audit-collector-hourly. - Click Force run to trigger the job manually.
- Wait a few seconds.
- Go to Cloud Run > Services.
- Click
onepassword-audit-collector. - Click the Logs tab.
Verify the function executed successfully. Look for:
--- 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 recordsGo to Cloud Storage > Buckets.
Click
onepassword-audit-secops-logs.Navigate to the
onepassword-audit/folder.Verify that a new
.ndjsonfile was created with the current timestamp.
If you see errors in the logs:
- HTTP 401: Check the bearer token in the
OP_BEARER_TOKENenvironment variable. The token may have expired. - HTTP 429: Rate limiting. The function automatically retries with backoff. The 1Password Events API allows 600 requests per minute and 30,000 requests per hour.
- Missing environment variables: Check all required variables are set in the Cloud Run function configuration.
- No records returned: Verify your bearer token has access to audit events using the introspect endpoint.
- The same events arrive repeatedly: Confirm that
STATE_KEYis not inside the prefix the feed reads. If the state file is deleted by the feed, the cursor is lost and the collector re-reads from its last recorded event.
Retrieve the Google SecOps service account
Google SecOps uses a unique service account to read data from your Cloud Storage bucket. You must grant this service account access to your bucket.
Get the service account email
- Go to SIEM Settings > Feeds.
- Click Add New Feed.
- Click Configure a single feed.
- In the Feed name field, enter a name for the feed (for example,
1Password Audit Events Logs). - Select Google Cloud Storage V2 as the Source type.
- Select 1Password Audit Events as the Log type.
- Click Get Service Account.
A unique service account email will be displayed, for example:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.comCopy this email address for use in the next step.
Click Next.
Specify values for the following input parameters:
Storage bucket URL: Enter the Cloud Storage bucket URI with the prefix path:
gs://onepassword-audit-secops-logs/onepassword-audit/Source deletion option: Select the deletion option according to your preference:
- Never delete files: Never delete files from the source (recommended for testing).
Delete transferred files and empty directories: Delete files and empty directories from the source after a successful fetch completes.
Maximum File Age: Include files modified in the last number of days (default is 180 days)
Asset namespace: The asset namespace
Ingestion labels: The label to be applied to the events from this feed
Click Next.
Review your new feed configuration in the Finalize screen, and then click Submit.
Grant IAM permissions to the Google SecOps service account
The Google SecOps service account needs two roles on your Cloud Storage bucket: Storage Object Viewer to read the log objects, and a bucket-level role to read the bucket metadata.
- Go to Cloud Storage > Buckets.
- Click
onepassword-audit-secops-logs. - Go to the Permissions tab.
- Click Grant access.
- Provide the following configuration details:
- Add principals: Paste the Google SecOps service account email
- Assign roles: Select both of the following:
- Storage Object Viewer: reads the log objects.
- Storage Legacy Bucket Reader: reads the bucket metadata. If you selected the Delete transferred files and empty directories deletion option, select Storage Legacy Bucket Writer instead, which also grants the delete permission.
Click Save.
UDM mapping table
| Log field | UDM mapping | Logic |
|---|---|---|
additional_action |
additional.fields |
Merged |
aux_id_label |
additional.fields |
Merged |
aux_info_list |
additional.fields |
Merged |
object_type_label |
additional.fields |
Merged |
EventTime |
metadata.event_timestamp |
Parsed as yyyy-MM-dd HH:mm:ss |
expiry |
metadata.event_timestamp |
Parsed as yyyy-MM-ddTHH:mm:ssZZZZZZZZZZ |
item.session.login_time |
metadata.event_timestamp |
Parsed as yyyy-MM-ddTHH:mm:ssZZZZZZZZZZ |
item.timestamp |
metadata.event_timestamp |
Parsed as ISO8601 |
has_principal |
metadata.event_type |
Mapped: true → USER_UNCATEGORIZED, true → STATUS_UPDATE |
item.session.uuid |
network.session_id |
Directly mapped |
deviceUuid |
principal.asset.asset_id |
Directly mapped |
session_ip |
principal.asset.ip |
Merged |
item.actor_details.email |
principal.email |
Directly mapped |
session_ip |
principal.ip |
Merged |
item.location.city |
principal.location.city |
Directly mapped |
item.location.country |
principal.location.country_or_region |
Directly mapped |
item.location.region |
principal.location.name |
Directly mapped |
item.location.latitude |
principal.location.region_latitude |
Directly mapped |
item.location.longitude |
principal.location.region_longitude |
Directly mapped |
uuid_label |
principal.resource.attribute.labels |
Merged |
item.actor_details.name |
principal.user.user_display_name |
Directly mapped |
item.actor_details.uuid |
principal.user.userid |
Directly mapped |
actor_uuid_label |
security_result.about.resource.attribute.labels |
Merged |
aux_uid_label |
security_result.about.resource.attribute.labels |
Merged |
object_id_label |
security_result.about.resource.attribute.labels |
Merged |
item.session.device_uuid |
target.resource.product_object_id |
Directly mapped |
| N/A | metadata.event_type |
Constant: USER_UNCATEGORIZED |
| N/A | metadata.product_name |
Constant: ONEPASSWORD_AUDIT_EVENTS |
| N/A | metadata.vendor_name |
Constant: ONEPASSWORD |
Change Log
View the Change Log for this parser
Need more help? Get answers from Community members and Google SecOps professionals.