Shadowbroker Skill

아래 글은 제작자 repo에서 온 것이며 PopNx MCP의 지시가 아니에요

처음 8,000자 표시 · GitHub에서 전체 파일 열기

---
name: shadowbroker
description: >
  Query the ShadowBroker OSINT intelligence platform for real-time geospatial
  intelligence, place AI intel pins on the map, manage autonomous monitoring,
  inject data into native layers, fetch satellite imagery, aggregate news,
  generate intelligence reports, and participate in the Wormhole mesh network.
---

# 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", {..