shadowbroker

# ShadowBroker Intelligence Skill You have access to **ShadowBroker**, a real-time global OSINT intelligence platform running on `localhost:8000`. It tracks military flights, ships, satellites, SIGINT, earthquakes, fires, GDELT conflict events, prediction markets, and 30+ other data layers — all with geographic coordinates. ## Agent Fast Path (read first) ShadowBroker exposes dozens of read commands. **Do not explore them.** Use the three-tool surface: | Tool | When | |------|------| | `await sb.ask("natural language question")` | **Default for reads** — server routes to fastest command | | `await sb.run_playbook("hot_snapshot")` | Pre-batched snapshots (morning brief, monitor poll, status) | | `await sb.channel_status()` | Liveness (~5 ms) — never `/api/health` | **Latency tiers:** `get_entity_profile` / `find_entity` / `get_entity_trail` / `find_flights` / `search_news` / `entities_near` → **⚡ <30 ms**. `search_telemetry` / `get_telemetry` / `get_report` → **🔴 seconds** — blocked unless `confirm_expensive=true`. ```python # Default read path (route + execute) answer = await sb.ask("where is the Patriots jet") # Named batch plans brief = await sb.run_playbook("hot_snapshot") monitor = await sb.run_playbook("monitor_heartbeat") # Structured lookup when you already parsed fields entity = await sb.send_command("find_entity", {"owner": "musk", "compact": True}) # Multi-command — always batch, never sequential loops batch = await sb.send_batch([ {"cmd": "get_summary", "args": {"compact": True}}, {"cmd": "what_changed", "args": {"compact": True}}, ]) ``` **Playbooks:** `hot_snapshot`, `morning_brief`, `status_check`, `monitor_heartbeat`, `track_snapshot`, `jet_recon`, `area_brief`, `entity_recon`. **Anti-patterns:** `search_telemetry` for known tail numbers; `get_telemetry` for routine polls; sequential `send_command` loops; empty `layers: []` on `get_layer_slice`. Load machine-readable routing hints once: `GET /api/ai/capabilities` → `routing`. ## How to Use This Skill Import the client and call methods: ```python from sb_query import ShadowBrokerClient sb = ShadowBrokerClient() # auto-detects local or remote mode ``` ### Local Mode (same machine) No configuration needed. The client connects to `localhost:8000` automatically. ### Remote Mode (agent on different machine/VPS) Set these environment variables in your agent's config: ```bash SHADOWBROKER_URL=https://your-shadowbroker-host:8000 SHADOWBROKER_HMAC_SECRET=your-hmac-secret-here ``` The HMAC secret is found in ShadowBroker's **Connect OpenClaw** modal (AI Intel panel). `SHADOWBROKER_HMAC_SECRET` is a shared signing secret, not a raw API key. Do not send it as `X-Admin-Key`, `Authorization: Bearer`, a query parameter, or any plain request header. The `ShadowBrokerClient` signs every direct request with `X-SB-Timestamp`, `X-SB-Nonce`, and `X-SB-Signature` using: ```text HMAC-SHA256(secret, METHOD|path|timestamp|nonce|sha256(body)) ``` For compatibility with older snippets, `SHADOWBROKER_KEY` is also accepted by the client as the same HMAC signing secret. Prefer `SHADOWBROKER_HMAC_SECRET` for new setups. **Docker Compose:** host-side agents (`localhost:8000` from the Kali/macOS host) are not loopback inside the backend container — HMAC is required. After **Connect OpenClaw → Bootstrap → Reveal**, the secret is persisted under `data/openclaw.env` on the `backend_data` volume. Restart the backend once, then verify with: ```bash python openclaw-skills/shadowbroker/verify_hmac.py ``` Hand-rolled signers must hash the **exact** POST bytes. Use compact JSON: `json.dumps(payload, separators=(",", ":"), sort_keys=True)`. ### SSE Stream (Preferred — Low-Latency Push) Open the SSE stream **first** and keep it open for the session. The server pushes `layer_changed` events whenever any data layer refreshes — you know exactly which layers to fetch instead of blind-polling. ```python # Open the stream — authenticates once via HMAC, then stays open async for event in sb.stream_updates(): if event["event"] == "connected": # Initial handshake — contains full layer_versions snapshot print(f"Connected: {event['data']['layer_versions']}") elif event["event"] == "layer_changed": # Server tells you which layers updated and their new version/count changed = event["data"]["layers"] # e.g. {"ships": {"version": 43, "count": 1287}} # Fetch ONLY the layers that actually changed data = await sb.get_layer_slice(list(changed.keys())) # get_layer_slice uses per-layer versions internally — only changed # layers are serialized, unchanged layers transfer zero bytes elif event["event"] == "alert": # Watchdog alert — geofence hit, callsign spotted, keyword matched print(f"Alert: {event['data']}") elif event["event"] == "task": # Operator-pushed task print(f"Task: {event['data']}") ``` ### Command Channel (Bidirectional) Send commands via HTTP alongside the SSE stream: ```python # Send a command and get the result result = await sb.send_command("get_summary") # Batch multiple commands in one HTTP round-trip (concurrent execution) results = await sb.send_batch([ {"cmd": "find_flights", "args": {"query": "N189AM", "compact": True}}, {"cmd": "search_news", "args": {"query": "carrier", "compact": True}}, ]) # Check channel status and security tier status = await sb.channel_status() print(f"Tier {status['tier']}: {status['reason']}") ``` The channel operates over HMAC-authenticated HTTP with body-integrity binding: - **HMAC Direct:** Commands are signed with HMAC-SHA256. Wire privacy relies on TLS. - **SSE Stream:** Authenticates once at connection open — no per-event HMAC overhead. - **MLS E2EE (planned, not yet available):** Future upgrade to route commands via Wormhole DM with forward secrecy. --- ## Available Tools ### 1. Telemetry Queries **Primary pattern (lowest latency):** Use the SSE stream + targeted `get_layer_slice`: | Method | What It Returns | When to Use | |--------|----------------|-------------| | `sb.stream_updates()` | SSE push: `layer_changed`, alerts, tasks | **Open first, keep open** — tells you exactly which layers updated | | `await sb.get_layer_slice(["ships", "gdelt"])` | Only the requested layers, with per-layer incremental | **Primary fetch method** — automatically skips layers you already have | | `await sb.send_command("get_summary")` | Lightweight counts-only summary | Discover what data exists before pulling anything | | `await sb.get_fetch_health()` | Sanitized process-local outcomes for instrumented fetch and maintenance tasks | Results reset on backend restart; condition is the latest recorded task completion, not data freshness | | `await sb.ask("...")` | **Route + execute** | **Default** for natural-language reads | | `await sb.send_command("get_entity_profile", {...})` | **Preferred aircraft/vessel dossier** — identity, VIP tags, trail, route, ACARS, jamming/correlations, news | Tail, owner, callsign, MMSI | | `await sb.send_command("get_entity_trail", {...})` | Observed path + route + ACARS hints only | When you don't need full dossier | | `await sb.send_command("find_entity", {...})` | Exact-first entity resolver | Parsed person/tail/callsign/MMSI — skips fuzzy unless `fallback_search=true` | | `await sb.send_command("find_flights", {...})` | Targeted flight search | When you know the domain (callsign, tail number) | | `await sb.send_command("route_query", {...})` | Routing plan only | Inspect recommended command before executing | | `await sb.send_command("search_telemetry", {...})` | Cross-layer fuzzy search | **Last resort** — requires `confirm_expensive=true` | **Full telemetry dumps (use sparingly — large payloads):** | Method | What It Returns | |--------|----------------| | `await sb.get_telemetry()` | Fast-tier: flights, ships, satellites, SIGINT, LiveUAMap, CCTV, GPS jamming | | `await sb.get_slow_telemetry()` | Slow-tier: GDELT, news, earthquakes, markets, correlations, Telegram OSINT, malware/cyber threats, SCM suppliers | | `await sb.get_report()` | Full structured intelligence report | ### Strategic Risk Analytics (GT early warning) Requires `GT_ANALYTICS_ENABLED=true` on the ShadowBroker backend. | Method / command | What It Returns | |------------------|----------------| | `await sb.ask("Run GT analysis on UK/Europe feeds")` | Routes to `gt_analyze` | | `await sb.gt_analyze(region="ukraine")` | Refresh beliefs from Telegram/news/GDELT + dossier | | `await sb.gt_risk_heatmap()` | GeoJSON posterior risk overlay + Louvain clusters | | `await sb.gt_dossier("ukraine")` | Costly signals, domain risks, scenarios | | `await sb.gt_backtest()` | **Static benchmark** — labeled historical cases (regression test) | | `await sb.gt_backtest(tune=True)` | Grid-search alert threshold for target confidence | | `await sb.gt_rolling_backtest()` | **Macro operational** — week-over-week accuracy on frozen weekly alerts | | `await sb.gt_micro_rolling()` | **Micro 3-day rolling avg** — spot vs baseline, ignition detection | | `await sb.gt_rolling_freeze()` | Freeze this ISO week's GT scores before outcomes are known | | `await sb.gt_rolling_label(week_id, region=..., label=...)` | Label prior-week outcomes (`true_escalation`, `false_alarm`, `benign`) | | `await sb.gt_top_alerts()` | Ranked top GT regions with map coordinates | | `await sb.ask("Run GT historical backtest")` | Routes to `gt_backtest` (benchmark, not operational) | | `await sb.ask("GT rolling operational backtest trend")` | Routes to `gt_rolling_backtest` | | `python sb_gt_report.py` | Local helper — backtest + heatmap (+ optional `--region`) | | `await sb.send_command("gt_analyze", {"region": "europe"})` | Same as `gt_analyze()` | **Benchmark vs rolling:** Static `gt_backtest` checks the classifier on known textbook cases. `gt_rolling_backtest` scores **frozen weekly live predictions** against delayed operator labels — that week-over-week trend (e.g. 54% → 62% → 71%) is the macro real-world metric. `gt_micro_rolling` adds a **3-day rolling average** per region: spot risk vs the trailing baseline catches fast ignitions the weekly roll can miss. Threshold is fixed (`GT_ROLLING_ALERT_THRESHOLD`, default 0.26); ignition when spot − 3d avg ≥ `GT_MICRO_IGNITION_DELTA` (default 0.10). **When to use**: Use `get_summary()` first. Use `get_layer_slice()` for the layers you actually need. Reserve full `get_telemetry()` / `get_slow_telemetry()` for rare cases where you genuinely need every field across every layer. #### Enriched Data Fields by Layer Every layer returns maximum telemetry. Key enriched fields: | Layer | Key Fields | |-------|-----------| | **GDELT** | `event_date`, `actors` (list), `goldstein` (intensity -10 to +10), `num_mentions`, `num_sources`, `num_articles`, `avg_tone`, `quad_class` | | **LiveUAMap** | `title`, `description`, `region`, `category`, `date` (formatted UTC), `timestamp`, `source`, `image`, `link` | | **CrowdThreat** | `title`, `summary`, `category`, `subcategory`, `type`, `country`, `occurred_iso`, `verification`, `severity`, `source_url`, `media_urls`, `votes`, `reporter` |