SIEM Integration

Stream audit events from reconciliation runs to your security monitoring infrastructure in real time.

Overview

Every action in Kanoniv (reconciliation health alerts, entity merges, field updates, manual overrides, and API mutations) generates an audit event. SIEM integration lets you stream these events to external destinations (Splunk, Datadog, PagerDuty, or any webhook-compatible endpoint) so your security and compliance teams can monitor identity resolution activity alongside the rest of your infrastructure.

This is particularly valuable for enterprises that need to:

• Detect anomalous reconciliation behavior (sudden drop in merge rate, spike in review-queue items)
• Maintain a tamper-proof audit trail in an external system of record
• Trigger PagerDuty or Slack alerts when reconciliation health degrades
• Feed identity resolution events into SOC dashboards alongside authentication and access logs

Cloud Feature

Audit stream configuration requires the Cloud tier. Basic audit events (API actions, overrides) are available on all tiers and can be queried via the API.

Supported Destinations

Type	Status	Examples
Webhook	Available	Splunk HEC, Datadog Logs, PagerDuty Events API, custom HTTP endpoints
S3/GCS	Planned	Data lake archival, batch analysis
Kafka	Planned	Real-time streaming pipelines

Webhook destinations receive a JSON POST request for each audit event. This covers the majority of SIEM platforms, which accept events via HTTP endpoints.

Configuration

Audit streams are managed through the Settings API. Each stream defines a destination type, URL, and authentication configuration.

Create an Audit Stream

Python

import httpx

client = httpx.Client(
    base_url="https://api.kanoniv.com",
    headers={"X-API-Key": "kn_..."},
)

# Create a Splunk HEC audit stream
resp = client.post("/v1/settings/audit-streams", json={
    "destination_type": "webhook",
    "destination_url": "https://splunk.company.com:8088/services/collector/event",
    "auth_config": {
        "header": "Authorization",
        "value": "Splunk <your-hec-token>"
    },
    "is_active": True,
})
stream = resp.json()
print(f"Created stream: {stream['id']}")

curl

curl -X POST https://api.kanoniv.com/v1/settings/audit-streams \
  -H "X-API-Key: kn_..." \
  -H "Content-Type: application/json" \
  -d '{
    "destination_type": "webhook",
    "destination_url": "https://splunk.company.com:8088/services/collector/event",
    "auth_config": {
      "header": "Authorization",
      "value": "Splunk <your-hec-token>"
    },
    "is_active": true
  }'

List Audit Streams

Python

resp = client.get("/v1/settings/audit-streams")
for stream in resp.json():
    status = "active" if stream["is_active"] else "paused"
    print(f"{stream['id']} -> {stream['destination_url']} ({status})")

curl

curl https://api.kanoniv.com/v1/settings/audit-streams \
  -H "X-API-Key: kn_..."

Update an Audit Stream

Python

stream_id = "b3f1a2c4-..."

resp = client.put(f"/v1/settings/audit-streams/{stream_id}", json={
    "destination_type": "webhook",
    "destination_url": "https://splunk.company.com:8088/services/collector/event",
    "auth_config": {
        "header": "Authorization",
        "value": "Splunk <new-hec-token>"
    },
    "is_active": True,
})

curl

curl -X PUT https://api.kanoniv.com/v1/settings/audit-streams/b3f1a2c4-... \
  -H "X-API-Key: kn_..." \
  -H "Content-Type: application/json" \
  -d '{
    "destination_type": "webhook",
    "destination_url": "https://splunk.company.com:8088/services/collector/event",
    "auth_config": {
      "header": "Authorization",
      "value": "Splunk <new-hec-token>"
    },
    "is_active": true
  }'

Disable or Delete a Stream

Python

# Pause a stream (set is_active to false)
client.put(f"/v1/settings/audit-streams/{stream_id}", json={
    "destination_type": "webhook",
    "destination_url": "https://splunk.company.com:8088/services/collector/event",
    "auth_config": { "header": "Authorization", "value": "Splunk ..." },
    "is_active": False,
})

# Or delete it entirely
client.delete(f"/v1/settings/audit-streams/{stream_id}")

curl

# Delete a stream
curl -X DELETE https://api.kanoniv.com/v1/settings/audit-streams/b3f1a2c4-... \
  -H "X-API-Key: kn_..."

Destination Examples

Datadog Logs:

{
  "destination_type": "webhook",
  "destination_url": "https://http-intake.logs.datadoghq.com/api/v2/logs",
  "auth_config": {
    "header": "DD-API-KEY",
    "value": "<your-datadog-api-key>"
  },
  "is_active": true
}

PagerDuty Events API v2:

{
  "destination_type": "webhook",
  "destination_url": "https://events.pagerduty.com/v2/enqueue",
  "auth_config": {
    "header": "X-Routing-Key",
    "value": "<your-pagerduty-integration-key>"
  },
  "is_active": true
}

Custom endpoint with Bearer token:

{
  "destination_type": "webhook",
  "destination_url": "https://siem.internal.company.com/events/kanoniv",
  "auth_config": {
    "header": "Authorization",
    "value": "Bearer <your-token>"
  },
  "is_active": true
}

Event Format

Every audit event dispatched to SIEM destinations follows this schema:

{
  "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "tenant_id": "t9f8e7d6-c5b4-a312-1098-765432fedcba",
  "actor_id": "u1234567-89ab-cdef-0123-456789abcdef",
  "actor_type": "system",
  "action": "reconciliation.health_alert",
  "resource_type": "batch_run",
  "resource_id": "r1a2b3c4-d5e6-f789-0123-456789abcdef",
  "before_state": null,
  "after_state": {
    "status": "degraded",
    "merge_rate": 0.42,
    "singleton_rate": 0.58,
    "canonical_count": 12400
  },
  "reason": "Merge rate dropped below threshold (0.42 < 0.60)",
  "timestamp": "2026-02-05T10:00:00Z"
}

Field Reference

Field	Type	Description
id	uuid	Unique event identifier
tenant_id	uuid	Tenant that owns this event
actor_id	uuid	Who performed the action. For system-generated events (e.g., reconciliation), this is the tenant ID.
actor_type	string	One of: user, worker, system, api_key
action	string	Event type. See Action Types below.
resource_type	string	The kind of resource affected: canonical_entity, batch_run, override, etc.
resource_id	uuid	ID of the affected resource
before_state	json | null	Resource state before the action (null for creates and alerts)
after_state	json | null	Resource state after the action
reason	string | null	Human-readable explanation, when available
timestamp	datetime	ISO 8601 timestamp (UTC)

Action Types

Action	Trigger
reconciliation.health_alert	Reconciliation run completed with degraded or unhealthy status
field_update	A field value changed on a canonical entity
merge	Two or more entities were merged into a single canonical
split	A canonical entity was split apart
revert	An entity was reverted to a previous state
override.create	A manual override was applied
override.delete	A manual override was removed

Querying Audit Events

You can query the same audit events that stream to SIEM destinations directly via the API. This is useful for ad-hoc investigation without leaving your terminal.

Recent Events

Returns the 100 most recent audit events for your tenant:

Python

import httpx

client = httpx.Client(
    base_url="https://api.kanoniv.com",
    headers={"X-API-Key": "kn_..."},
)

resp = client.get("/v1/audit")
events = resp.json()

for event in events:
    print(f"{event['timestamp']} | {event['action']} | {event['resource_type']}:{event['resource_id']}")

curl

curl https://api.kanoniv.com/v1/audit \
  -H "X-API-Key: kn_..."

Entity Audit Trail

Get the full audit history for a specific entity:

Python

entity_id = "ent_a1b2c3d4-..."

resp = client.get(f"/v1/audit/entity/{entity_id}")
trail = resp.json()

for event in trail:
    print(f"{event['timestamp']} | {event['action']}")
    if event.get("before_state") and event.get("after_state"):
        print(f"  before: {event['before_state']}")
        print(f"  after:  {event['after_state']}")

curl

curl https://api.kanoniv.com/v1/audit/entity/ent_a1b2c3d4-... \
  -H "X-API-Key: kn_..."

End-to-End: SDK Reconciliation to SIEM

This section walks through a complete workflow: configuring a SIEM destination, running a reconciliation with the Python SDK, and observing the resulting audit events in Splunk.

Step 1: Configure a Splunk HEC Destination

First, create an audit stream that points to your Splunk HTTP Event Collector:

Python

import httpx

api = httpx.Client(
    base_url="https://api.kanoniv.com",
    headers={"X-API-Key": "kn_..."},
)

# Create the Splunk audit stream
resp = api.post("/v1/settings/audit-streams", json={
    "destination_type": "webhook",
    "destination_url": "https://splunk.company.com:8088/services/collector/event",
    "auth_config": {
        "header": "Authorization",
        "value": "Splunk a1b2c3d4-e5f6-7890-abcd-ef1234567890"
    },
    "is_active": True,
})
resp.raise_for_status()
print(f"Audit stream created: {resp.json()['id']}")

curl

curl -X POST https://api.kanoniv.com/v1/settings/audit-streams \
  -H "X-API-Key: kn_..." \
  -H "Content-Type: application/json" \
  -d '{
    "destination_type": "webhook",
    "destination_url": "https://splunk.company.com:8088/services/collector/event",
    "auth_config": {
      "header": "Authorization",
      "value": "Splunk a1b2c3d4-e5f6-7890-abcd-ef1234567890"
    },
    "is_active": true
  }'

Step 2: Load Data with Python SDK Source Adapters

Use the Kanoniv Python SDK to pull records from multiple systems:

from kanoniv import Source, Spec, reconcile

# Load CRM contacts from a CSV export
crm = Source.from_csv("crm", "crm_contacts.csv", primary_key="customer_id")

# Load Stripe billing data from a Pandas DataFrame
import pandas as pd
stripe_df = pd.read_csv("stripe_export.csv")
stripe = Source.from_pandas("stripe", stripe_df, primary_key="stripe_id")

# Load warehouse data directly from PostgreSQL
warehouse = Source.from_warehouse(
    "pg_customers",
    table="public.customers",
    connection_string="postgresql://user:pass@host/db",
    primary_key="customer_id",
)

# Or pull from a dbt model in Snowflake
dbt_src = Source.from_dbt(
    "dbt_orders",
    model="ref('stg_orders')",
    manifest_path="target/manifest.json",
    connection_string="snowflake://user:pass@account/db/schema",
    primary_key="order_id",
)

Step 3: Run Reconciliation

Define a spec and run reconciliation. The tier="enterprise" flag enables audit event generation for all match decisions:

spec = Spec.from_file("customer_spec.yml")

result = reconcile(
    sources=[crm, stripe, warehouse],
    spec=spec,
    tier="cloud",
)

print(f"Clusters: {result.cluster_count}")
print(f"Merge rate: {result.merge_rate:.2%}")
print(f"Singletons: {result.singleton_count}")

# Export golden records to a DataFrame for downstream use
golden_df = result.to_pandas()

Step 4: Audit Events Flow Automatically

When reconciliation runs through the Kanoniv cloud API, the pipeline generates audit events at each stage:

1. Health alerts. If the reconciliation run produces degraded health metrics (low merge rate, high singleton rate), a reconciliation.health_alert event is published.
2. Entity mutations. Merges, splits, and field updates each produce their own audit events with before_state and after_state snapshots.
3. Override applications. Any manual overrides (merge, split, lock, field) applied during the run generate override.create events.

These events are:

• Persisted to the audit_events table (queryable via GET /v1/audit)
• Published to Redis pubsub on the audit_events:{tenant_id} channel
• Picked up by the SIEM dispatcher, which forwards them to all active audit streams for the tenant

SDK reconcile() --> Cloud API --> Pipeline Worker
                                       |
                                       +--> audit_events table (API queryable)
                                       |
                                       +--> Redis pubsub audit_events:{tenant_id}
                                               |
                                               +--> SIEM Dispatcher
                                                       |
                                                       +--> Splunk HEC (webhook)
                                                       +--> Datadog (webhook)
                                                       +--> PagerDuty (webhook)

Step 5: Query the Audit Trail

After a reconciliation run, inspect the audit events for a specific entity:

Python

# Find the audit trail for a specific canonical entity
entity_id = golden_df.iloc[0]["canonical_id"]

resp = api.get(f"/v1/audit/entity/{entity_id}")
trail = resp.json()

for event in trail:
    print(f"[{event['timestamp']}] {event['action']} ({event['actor_type']})")
    if event.get("reason"):
        print(f"  Reason: {event['reason']}")
    if event.get("after_state"):
        print(f"  State: {event['after_state']}")

curl

curl https://api.kanoniv.com/v1/audit/entity/<entity-id> \
  -H "X-API-Key: kn_..."

Example output:

[2026-02-05T10:00:12Z] merge (worker)
  Reason: Matched on email_exact (confidence: 0.97)
  State: {"merged_from": ["src_crm:C-1042", "src_stripe:cus_abc123"]}

[2026-02-05T10:00:12Z] field_update (worker)
  State: {"field": "phone", "old": "+1-555-0100", "new": "+1-555-0199", "source": "pg_customers"}

Step 6: Monitor Reconciliation Health in Splunk

Once events are flowing to Splunk, you can build alerts that trigger when reconciliation health degrades. The reconciliation.health_alert event includes the full health summary in after_state:

{
  "id": "evt_...",
  "action": "reconciliation.health_alert",
  "resource_type": "batch_run",
  "after_state": {
    "status": "degraded",
    "merge_rate": 0.42,
    "singleton_rate": 0.58,
    "canonical_count": 12400
  },
  "reason": "Merge rate dropped below threshold",
  "timestamp": "2026-02-05T10:00:00Z"
}

Create a Splunk alert on this event to notify your on-call team when reconciliation quality drops.

Reliability

The SIEM dispatcher is designed for reliable delivery:

• Buffered delivery. Events are buffered in Redis pubsub before dispatch. The dispatcher processes events asynchronously, so a slow destination does not block the reconciliation pipeline.
• Per-stream isolation. Each audit stream destination is dispatched in its own async task. A failure in one destination does not affect others.
• Retry on failure. Failed webhook deliveries are retried. Delivery failures are logged and counted via the siem_events_dispatched_total metric (with status="failure" label).
• Observability. The dispatcher emits metrics for monitoring:
  • siem_events_received_total: Total events received from Redis, by tenant
  • siem_events_dispatched_total: Total events dispatched, by tenant, destination, and status (success / failure)

WARNING

If the SIEM dispatcher is down, events published to Redis pubsub during the outage will be lost (Redis pubsub is fire-and-forget). Events are always persisted to the audit_events table regardless of dispatcher status, so you can backfill from the API if needed.

Splunk Dashboard Example

Once events are flowing into Splunk, use these queries to build a Kanoniv monitoring dashboard.

Reconciliation Health Alerts (Last 24h)

index=kanoniv action="reconciliation.health_alert"
| spath output=status path=after_state.status
| spath output=merge_rate path=after_state.merge_rate
| timechart span=1h count by status

All Entity Mutations by Action Type

index=kanoniv resource_type="canonical_entity"
| stats count by action
| sort -count

Merge Rate Over Time

index=kanoniv action="reconciliation.health_alert"
| spath output=merge_rate path=after_state.merge_rate
| timechart span=1h avg(merge_rate) as avg_merge_rate

Alert: Merge Rate Below Threshold

Create a Splunk saved search with alert action:

index=kanoniv action="reconciliation.health_alert"
| spath output=merge_rate path=after_state.merge_rate
| where merge_rate < 0.60
| table timestamp, tenant_id, merge_rate, reason

Set this as a real-time alert that triggers a PagerDuty incident or Slack notification when the merge rate drops below your acceptable threshold.

Next Steps

• Audit Logging & Match Trail: Configure what gets logged during reconciliation
• Compliance & PII Masking: Ensure audit events respect PII redaction rules