Recipe: Threat intel feed integration

Overview

Meridian exposes a paginated, append-only threat intel feed at /api/v1/intel/feed. Each event carries a monotonic cursor, a SHA-256 content hash, and an Ed25519 signature over the event body. Consumers poll the endpoint with their last-seen cursor and verify integrity before ingestion.

Prerequisites

• A Meridian API key with the intel:read scope.
• The Ed25519 public key from your Meridian dashboard (Settings → Intel Feed).
• A consumer that can verify Ed25519 signatures and tolerate HTTP 429 back-pressure.

Feed schema

{
  "cursor": "1711372800-a3f2b1",
  "events": [
    {
      "id": "evt_9xK2mQ",
      "timestamp": "2025-03-25T14:22:01Z",
      "type": "process_injection",
      "severity": "high",
      "source": "meridian_agent/3.2.1",
      "payload": { ... },
      "hash": "sha256:abc123...",
      "signature": "base64..."
    }
  ],
  "truncated": false
}

When truncated is true, immediately re-poll with the last cursor to drain the remaining window.

Integration steps

1. Fetch the initial cursor. Call GET /api/v1/intel/feed with no query parameters. Store the returned cursor.
2. Poll with cursor. Append ?cursor=<value> on subsequent requests. The server returns only events after that cursor.
3. Verify each event. Compute SHA-256 over the raw event JSON (excluding the signature field), then verify the Ed25519 signature against your public key. Discard any event that fails verification.
4. Handle rate limits. Respect the Retry-After header on 429 responses. Use exponential backoff with jitter (base 1 s, cap 60 s).
5. Persist cursor. Write the last successfully verified cursor to durable storage so restarts never replay or skip events.

Reference consumer (Python)

import requests, json, time
from nacl.signing import VerifyKey

API = "https://api.getnimbus.net/v1/intel/feed"
KEY = VerifyKey(bytes.fromhex("YOUR_PUBKEY_HEX"))
cursor = None

while True:
    params = {"cursor": cursor} if cursor else {}
    resp = requests.get(API, headers={
        "Authorization": "Bearer <token>"
    }, params=params)
    if resp.status_code == 429:
        time.sleep(int(resp.headers.get("Retry-After", 5)))
        continue
    resp.raise_for_status()
    data = resp.json()
    for evt in data["events"]:
        sig = bytes.fromhex(evt.pop("signature"))
        raw = json.dumps(evt, sort_keys=True).encode()
        KEY.verify(raw, sig)
        # forward to SIEM here
    cursor = data["cursor"]
    if not data["truncated"]:
        time.sleep(2)