Skip to main content

Send security events

Use the Ingest API to feed events from your own tools (an EDR export, an API gateway, an MCP gateway) into the Security area, where Bairav raises findings and shadow-agent candidates.

What you need​

  • A user with the manage_observability permission, to create an ingest token.
  • An ingest token, obt_xxxxxxxx.xxxxxxxx, bound to one telemetry source.

1. Create an ingest token​

In the console open Settings, Security data and telemetry and create a token, or call POST /api/security/ingest-tokens.

FieldMeaning
nameA label, for example edr-feed
source_kindOne of mcp_gateway, api_gateway, otel, imported, infrastructure, sensor, test_fixture
source_nameA display name for the source

The secret is returned once. A tenant can hold at most 50 active tokens.

2. Send a batch​

samples/ingest/security_events.py
"""Send one security event to the Bairav Ingest API. Standard library only.

    export BAIRAV_HOST=your-tenant.example.com
    export BAIRAV_INGEST_TOKEN=obt_xxxxxxxx.xxxxxxxx   # create one in Settings > Security data and telemetry
"""
import datetime as dt
import json
import os
import urllib.error
import urllib.request
import uuid

host = os.environ["BAIRAV_HOST"]
token = os.environ["BAIRAV_INGEST_TOKEN"]

batch = {
    "schema_version": "1.0",
    "events": [
        {
            # event_uid makes the call idempotent: sending the same one again is reported as a duplicate.
            "event_uid": f"example-{uuid.uuid4()}",
            "occurred_at": dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
            "event_type": "llm.call",
            "severity": "info",
            "observed_identity": "host:build-agent-04/proc:python",
            "summary": "POST api.openai.com /v1/chat/completions",
            "fields": {"host": "api.openai.com", "port": 443, "sdk": "openai-python/1.40"},
        }
    ],
}

req = urllib.request.Request(
    f"https://{host}/api/ingest/security-events",
    data=json.dumps(batch).encode(),
    method="POST",
    headers={"Content-Type": "application/json", "X-Ingest-Token": token},
)
try:
    with urllib.request.urlopen(req, timeout=15) as resp:
        print(resp.status, json.load(resp))  # 202 {"accepted": 1, "duplicates": 0, "rejected": [], "dead_lettered": 0}
except urllib.error.HTTPError as err:
    # 401 bad token, 413 too large, 422 invalid batch, 429/503 retry after the Retry-After header
    print(err.code, err.headers.get("Retry-After"), err.read().decode())
    raise SystemExit(1)

A successful call returns 202:

{ "accepted": 1, "duplicates": 0, "rejected": [], "dead_lettered": 0 }

Behaviour to design around​

  • Idempotent. Events are keyed on (tenant, source, event_uid). Re-sending is safe; repeats come back as duplicates.
  • Per event, not per batch. An event with an unknown event_type or a bad field is returned in rejected with a reason and is not stored. The rest of the batch is.
  • Dead letters. An event that validated but failed processing is parked and retried automatically. dead_lettered counts them. An admin can inspect and replay them with GET /api/security/ingest/dead-letters and POST /api/security/ingest/dead-letters/replay.
  • Limits. At most 500 events and 256 KB per request, and 600 events a minute per token. Over the limit you get 429 with Retry-After. If the rate limiter is down you get 503 with Retry-After: the limit fails closed on purpose, and retrying is safe.
  • Masking. Content is masked at ingest. Prompts, responses and tool arguments are not stored unless your tenant data policy allows redacted content.

The catalogue of event types, severities and fields is in Security event types.

Where the events go​

They appear under Security, Events, are evaluated by the built-in detections, and can raise findings. Agents seen in events without a registered match become shadow-agent candidates; see Shadow agents.