id: onepassword-scim-bridge-sync-status-sentinel
namespace: company.security
description: |
Audits 1Password SCIM Bridge health endpoints and provisioning logs, detects sync errors and heartbeat failures, and alerts via Slack.
concurrency:
limit: 1
inputs:
- id: scim_bridge_url
type: STRING
displayName: SCIM Bridge URL
defaults: "https://scim.internal.company.com"
description: Base HTTPS URL hosting the enterprise 1Password SCIM bridge deployment.
- id: max_sync_errors_threshold
type: INT
displayName: Max Sync Errors Threshold
defaults: 1
description: Maximum tolerated failed provisioning transactions before
classifying the bridge as degraded.
- id: alert_on_degraded
type: BOOL
displayName: Alert On Degradation
defaults: true
description: Toggle whether Slack webhook alerts are dispatched upon detecting
SCIM bridge degradation.
- id: timeout_seconds
type: INT
displayName: HTTP Request Timeout
defaults: 15
description: Connection timeout in seconds for SCIM bridge health and
provisioning API requests.
tasks:
- id: scan_scim_bridge_health
type: io.kestra.plugin.scripts.python.Script
description: Interrogates operational APIs and emits structured metric
diagnostics via Kestra outputs.
containerImage: python:3.11-slim
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
beforeCommands:
- pip install requests kestra
env:
SCIM_BRIDGE_URL: "{{ inputs.scim_bridge_url }}"
SCIM_BEARER_TOKEN: "{{ secret('OP_SCIM_BEARER_TOKEN') }}"
ERROR_THRESHOLD: "{{ inputs.max_sync_errors_threshold }}"
script: |
import os
import sys
import requests
from datetime import datetime, timezone
from kestra import Kestra
base_url = os.environ.get("SCIM_BRIDGE_URL", "https://scim.internal.company.com").rstrip("/")
bearer_token = os.environ.get("SCIM_BEARER_TOKEN", "")
error_threshold = int(os.environ.get("ERROR_THRESHOLD", "1"))
bridge_status = "UNKNOWN"
http_status = 0
recent_errors = 0
error_details = []
headers = {
"Authorization": f"Bearer {bearer_token}",
"Accept": "application/json"
}
try:
health_url = f"{base_url}/health"
resp = requests.get(health_url, headers=headers, timeout=12)
http_status = resp.status_code
if http_status == 200:
data = resp.json()
bridge_status = data.get("status", "HEALTHY")
else:
bridge_status = "DEGRADED"
error_details.append(f"HTTP {http_status}: {resp.text[:100]}")
recent_errors += 1
scim_test_url = f"{base_url}/Users?count=1"
user_resp = requests.get(scim_test_url, headers=headers, timeout=12)
if user_resp.status_code != 200:
recent_errors += 1
error_details.append(f"SCIM Users endpoint error: HTTP {user_resp.status_code}")
except Exception as e:
print(f"Exception during 1Password SCIM audit: {e}")
bridge_status = "UNREACHABLE"
http_status = 503
recent_errors = 2
error_details = ["Connection timeout to SCIM bridge ingress endpoint"]
is_alerting = (bridge_status != "HEALTHY") or (recent_errors >= error_threshold)
Kestra.outputs({
"bridge_status": bridge_status,
"http_status": http_status,
"recent_error_count": recent_errors,
"error_details": "; ".join(error_details) if error_details else "None",
"is_alerting": is_alerting,
"audit_timestamp": datetime.now(timezone.utc).isoformat()
})
- id: check_scim_bridge_degraded
type: io.kestra.plugin.core.flow.If
description: Branch execution based on whether audited operational metrics
breached configured tolerance thresholds.
condition: "{{ outputs.scan_scim_bridge_health.vars.is_alerting and
inputs.alert_on_degraded }}"
then:
- id: notify_slack_itops_alert
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Dispatches high-priority operational incident alert to designated
Slack webhook channel.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{
"channel": "#itops-alerts",
"username": "1Password-SCIM-Sentinel",
"text": "CRITICAL: 1Password SCIM Bridge at `{{ inputs.scim_bridge_url }}` is DEGRADED! Health Status: `{{ outputs.scan_scim_bridge_health.vars.bridge_status }}` (HTTP `{{ outputs.scan_scim_bridge_health.vars.http_status }}`), Sync Error Count: `{{ outputs.scan_scim_bridge_health.vars.recent_error_count }}` (Threshold: `{{ inputs.max_sync_errors_threshold }}`). Error Summary: `{{ outputs.scan_scim_bridge_health.vars.error_details }}`."
}
else:
- id: log_nominal_scim_state
type: io.kestra.plugin.core.log.Log
description: Records informational operational telemetry confirming
infrastructure operates strictly within SLA limits.
message: "1Password SCIM Bridge {{ inputs.scim_bridge_url }} health nominal:
status {{ outputs.scan_scim_bridge_health.vars.bridge_status }}, zero
provisioning errors detected."
- id: export_scim_manifest
type: io.kestra.plugin.core.debug.Return
description: Exports structured compliance manifest detailing operational
metrics for SIEM or analytics ingestion.
format:
bridge_status: "{{ outputs.scan_scim_bridge_health.vars.bridge_status }}"
http_status: "{{ outputs.scan_scim_bridge_health.vars.http_status }}"
recent_error_count: "{{ outputs.scan_scim_bridge_health.vars.recent_error_count }}"
error_details: "{{ outputs.scan_scim_bridge_health.vars.error_details }}"
is_alerting: "{{ outputs.scan_scim_bridge_health.vars.is_alerting }}"
audit_timestamp: "{{ outputs.scan_scim_bridge_health.vars.audit_timestamp }}"
triggers:
- id: periodic_scim_audit
type: io.kestra.plugin.core.trigger.Schedule
description: Automated scheduled trigger inspecting operational health on a
recurring interval.
cron: "*/15 * * * *"