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:

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.

# 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:

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:

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:

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:

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.

# 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:

# 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:

Full telemetry dumps (use sparingly — large payloads):

Strategic Risk Analytics (GT early warning)

Requires GT_ANALYTICS_ENABLED=true on the ShadowBroker backend.

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 aliases for get_layer_slice / search_telemetry: telegram → telegram_osint, malware/botnet → malware_threats, cyber/cisa/kev → cyber_threats, scm/suppliers → scm_suppliers.

1b. Recon / OSINT Toolkit

The Recon panel lookups are available on the OpenClaw command channel — no need to hit /api/osint/* directly.

# List supported tools
await sb.send_command("osint_tools")

# IP geolocation + threat context
await sb.send_command("osint_lookup", {"tool": "ip", "ip": "8.8.8.8"})

# DNS, WHOIS, certificate transparency
await sb.send_command("osint_lookup", {"tool": "dns", "domain": "example.com"})
await sb.send_command("osint_lookup", {"tool": "whois", "domain": "example.com"})
await sb.send_command("osint_lookup", {"tool": "certs", "domain": "example.com"})

# BGP/ASN, sanctions, CVE, MAC vendor, GitHub, breach check
await sb.send_command("osint_lookup", {"tool": "bgp", "query": "AS15169"})
await sb.send_command("osint_lookup", {"tool": "sanctions", "query": "Rosneft"})
await sb.send_command("osint_lookup", {"tool": "cve", "cve": "CVE-2024-1234"})
await sb.send_command("osint_lookup", {"tool": "mac", "mac": "00:11:22:33:44:55"})
await sb.send_command("osint_lookup", {"tool": "github", "username": "octocat"})
await sb.send_command("osint_lookup", {"tool": "leaks", "email": "user@example.com"})

# Entity relationship graph (aircraft, vessel, ip, company, person, country)
await sb.send_command("entity_expand", {"type": "ip", "id": "8.8.8.8"})
await sb.send_command("entity_expand", {"type": "aircraft", "id": "N400QS", "icao24": "a0f011"})

# Subnet sweep (full tier only — active Shodan InternetDB scan)
await sb.send_command("osint_sweep", {"ip": "1.2.3.4", "cidr": 24})

2. Pin Placement (AI Intel Map Layer)

Pins appear on the user's map in a dedicated "AI Intel" layer.

# Single pin
await sb.place_pin(
    lat=34.05, lng=-118.24,
    label="UAP Sighting #1",
    category="anomaly",       # see categories below
    description="Multiple witnesses reported lights over Griffith Observatory",
    source="NUFORC Database",
    source_url="https://nuforc.org/...",
    confidence=0.8,            # 0.0 to 1.0
    ttl_hours=48,              # auto-delete after 48 hours (0 = permanent)
)

# Batch pins (up to 100 at once)
await sb.place_pins_batch([
    {"lat": 34.05, "lng": -118.24, "label": "Site A", "category": "research"},
    {"lat": 34.10, "lng": -118.30, "label": "Site B", "category": "research"},
])

# List pins
pins = await sb.get_pins(category="anomaly")

# Delete
await sb.clear_pins(category="anomaly")  # by category
await sb.clear_pins()                     # all

Pin Categories (each has a specific color on the map):

3. Geocoding

# Place name → coordinates
results = await sb.geocode("Griffith Observatory, Los Angeles")
# Returns: [{"lat": 34.1184, "lon": -118.3004, "display_name": "..."}]

# Always geocode before placing pins if you have a place name, not coordinates.

4. Satellite Imagery

# Get latest Sentinel-2 satellite scenes for any location
scenes = await sb.get_satellite_images(lat=35.68, lng=51.38, count=3)
# Returns: {"scenes": [{"scene_id", "datetime", "cloud_cover", "thumbnail_url", "fullres_url"}]}

When to use: When the user asks to "see satellite images of [place]" or wants visual intelligence of a location. Geocode first, then fetch imagery.

5. News & GDELT Near Location

# Get GDELT conflict events + news articles near a coordinate
nearby = await sb.get_news_near(lat=-15.4, lng=28.3, radius=500)
# Returns: {"gdelt": [...], "news": [...]} with headlines, source URLs, distances

When to use: When the user asks "what's happening in [country/city]" or wants news from a specific region. Geocode the place name first.

6. Near Me (Full Proximity Scan)

# Get ALL telemetry within a radius of a location
everything = await sb.get_near_me(lat=39.74, lng=-104.99, radius_miles=100)
# Returns EVERY layer within radius, each item tagged with distance_miles:
#   military_flights, commercial_flights, tracked_flights, private_jets,
#   ships, sigint, earthquakes, volcanoes, gdelt, liveuamap, crowdthreat,
#   uap_sightings, wastewater, firms_fires, weather_alerts, air_quality,
#   cctv, gps_jamming, satellites, news, correlations

When to use: When the user says "what's near me" or wants a proximity digest. This pulls from both fast and slow tiers automatically.

7. Native Layer Data Injection

Inject custom data directly into ShadowBroker's native layers (CCTV, ships, etc.):

# Add a custom CCTV camera to the CCTV layer
await sb.inject_data("cctv", [
    {"lat": 34.1, "lng": -118.3, "url": "https://stream.example.com/cam1",
     "name": "My Traffic Camera"}
])

# Remove all user-injected data
await sb.clear_injected()           # all layers
await sb.clear_injected("cctv")     # just CCTV

Injectable layers: cctv, ships, sigint, kiwisdr, military_bases, datacenters, power_plants, satnogs_stations, volcanoes, earthquakes, news, viirs_change_nodes, air_quality

When to use: When the user wants to add their own data sources to existing layers (e.g., "add this CCTV camera I found", "add this military base").

8. Wormhole / InfoNet / Mesh Network

OpenClaw agents participate in the private Infonet on behalf of the operator who configured the skill. All traffic uses the operator's wormhole persona and local node runtime (MLS gate crypto, Ed25519 signing, Tor onion transport) — the agent does not get a separate fleet identity.

Access tiers

• restricted (default): read Infonet status, list gates, read gate messages, poll DMs.
• full (OPENCLAW_ACCESS_TIER=full): also warm Tor, join the swarm, post gate messages, cast votes, and send DMs when the user commands it.

Remote agents authenticate with HMAC on /api/ai/channel/command; loopback uses the local operator lane.

# Warm Tor, enable the node, announce to fleet seed (full tier)
await sb.ensure_infonet_ready(join_swarm=True)

# Status snapshot (chain health, wormhole, runtime)
status = await sb.infonet_status()

# Read the public Infonet gate (MLS-encrypted, decrypt with operator keys)
messages = await sb.read_gate_messages("infonet", limit=20, decrypt=True)

# Post on behalf of the operator (full tier) — propagates via peer-push
await sb.post_to_gate("infonet", "Intelligence bulletin: 3 carriers underway in Med")
# Legacy alias:
await sb.post_to_infonet("same as post_to_gate on infonet gate")

# Upvote / downvote a node (full tier)
await sb.cast_vote("!sb_peer_id_or_pubkey", vote=1, gate="infonet")

# Encrypted DMs (peer_id / !sb_... recipient)
await sb.send_encrypted_dm("!sb_recipient", "Eyes only: carrier update")
dms = await sb.read_encrypted_dms(limit=20)

gates = await sb.list_gates()
await sb.join_infonet_swarm()  # re-announce + refresh manifest

# Meshtastic radio
signals = await sb.listen_mesh(region="US", limit=20)
await sb.send_mesh("US", "ShadowBroker AI: SIGINT anomaly detected in sector 7")

# Dead drops
await sb.dead_drop_leave("location_hash", "anonymous intelligence payload")
found = await sb.dead_drop_check("location_hash")

9. Alert Delivery

Send branded alerts to the user's messaging channels:

from sb_alerts import AlertDispatcher
alerts = AlertDispatcher()
alerts.add_discord("https://discord.com/api/webhooks/YOUR/WEBHOOK")
alerts.add_telegram("BOT_TOKEN", "CHAT_ID")

await alerts.send_brief("Morning intelligence digest here...")
await alerts.send_warning("Earthquake M5.2 detected 43mi from your location")
await alerts.send_threat("Threat level changed: GUARDED → ELEVATED")
await alerts.send_news("Breaking: GPS jamming detected over Baltic Sea")
await alerts.send_intel("USS Ford entered Mediterranean, heading east")

10. Intelligence Reports

# Full structured report
report = await sb.get_report()
# Contains: summary stats, top military flights, correlations, earthquakes, SIGINT, pin counts

# Lightweight summary (counts only)
summary = await sb.get_summary()

11. SAR (Synthetic Aperture Radar) Layer

ShadowBroker can ingest free SAR data in two modes:

• Mode A (default-on, no account): Sentinel-1 scene catalog from the Alaska Satellite Facility — pure metadata, no downloads, no DSP. Lets the agent answer "what radar passes have happened over this AOI in the last 36 hours and when's the next pass?"
• Mode B (opt-in, free account): Pre-processed ground-change anomalies from NASA OPERA, Copernicus EGMS, GFM, EMS, and UNOSAT — already-computed flood polygons, deformation maps, and damage assessments. Requires the user to enable Mode B in Settings → SAR (sets two env flags) and add a free Earthdata token.

# Always check status first — when Mode B is off the response includes a
# step-by-step help block with signup URLs the agent can paste to the user.
status = await sb.sar_status()
if not status["data"]["products"]["enabled"]:
    # Mode B disabled — surface the in-app links to the user
    for step in status["data"]["products"]["help"]["steps"]:
        print(f"Step {step['step']}: {step['label']} → {step['url']}")

# Recent anomalies (Mode B; empty list when disabled)
anomalies = await sb.sar_anomalies_recent(kind="flood_extent", limit=20)

# Anomalies near a coordinate
near = await sb.sar_anomalies_near(lat=50.45, lng=30.52, radius_km=50)

# Scene catalog (Mode A; always populated when AOIs exist)
scenes = await sb.sar_scene_search(aoi_id="kyiv_metro", limit=10)

# Per-AOI coverage + next-pass estimate
coverage = await sb.sar_coverage_for_aoi(aoi_id="kyiv_metro")

# AOI management
aois = await sb.sar_aoi_list()
await sb.sar_aoi_add(
    id="port_of_odesa", name="Port of Odesa",
    center_lat=46.4858, center_lon=30.7333, radius_km=15,
    category="conflict",
)

# Promote an anomaly to an AI Intel pin (writes into the dashboard)
await sb.sar_pin_from_anomaly(anomaly_id="opera-disp-...", label="OPERA deformation")

# Continuous watchdog — fire when matching anomalies appear in an AOI
await sb.sar_watch_anomaly(aoi_id="port_of_odesa", kind="surface_water_change")

# Inspect the same detail payload the operator's map popup shows for a pin
detail = await sb.sar_pin_click(anomaly_id="opera-disp-...")
# -> {"anomaly": {...}, "aoi": {...}, "recent_scenes": [...]}

# Fly the operator's map to an AOI center (useful after adding a new AOI,
# or to direct attention after a fresh anomaly arrives).  The frontend
# picks this up via useAgentActions and calls its map flyTo handler.
await sb.sar_focus_aoi(aoi_id="kyiv_metro", zoom=9.0)

SAR rules of engagement:

1. Call sar_status() first when the user asks about SAR/radar/deformation/floods.
2. If Mode B is off, paste the help.steps URLs to the user — never tell them to "search for it", the links are right there in the response.
3. SAR anomalies carry an evidence_hash — preserve it when promoting to a pin so other nodes can verify lineage.
4. Mode B writes signed mesh events only when the local node is at private_transitional or higher. Otherwise the data stays local.

12. Analysis Zones (Agent-Authored Map Notes)

The old regex-based "contradiction detector" has been removed — it pattern matched denial keywords against internet outages and produced constant false positives. It has been replaced with analysis zones: colored square overlays you drop on the map with a written assessment. Think of them as sticky notes: "I noticed X in this area, here is what I think it means."

The operator reads your assessment by clicking the zone and can delete any zone from the popup with a trash icon. Zones persist across restarts.

# List zones currently on the map
zones = await sb.list_analysis_zones()

# Drop a new zone — general assessment (cyan)
await sb.place_analysis_zone(
    lat=50.45, lng=30.52,
    title="Kyiv metro unusual quiet",
    body=(
        "Transit ridership dropped ~60% vs baseline over the last 6 hours "
        "while ADS-B shows two Russian ELINT orbits north of the city. "
        "No official advisory posted yet.  Possible pre-strike posture, "
        "but could also be routine drill.  Watching for next 2h."
    ),
    category="observation",
    severity="medium",
    drivers=[
        "Transit -60% vs 7-day baseline",
        "2x Russian ELINT orbits at 34k ft N of city",
        "No advisory posted on official channels",
    ],
    cell_size_deg=0.8,  # city-scale
)

# Drop a contradiction note (amber) when statements conflict with telemetry
await sb.place_analysis_zone(
    lat=36.2, lng=37.1,
    title="Damascus 'normal operations' claim",
    body=(
        "MoD statement at 14:00 claimed 'normal operations' across Syria. "
        "At the same timestamp, Cloudflare radar shows 42% internet "
        "outage across the western corridor and three military bases "
        "went dark on SIGINT.  Worth a closer look."
    ),
    category="contradiction",
    severity="high",
    drivers=[
        "Official statement: 'normal operations'",
        "Cloudflare radar: 42% outage western corridor",
        "3 bases lost SIGINT emissions simultaneously",
    ],
)

# Delete a stale zone
await sb.delete_analysis_zone(zone_id="abc123def456")

# Wipe all zones (use sparingly)
await sb.clear_analysis_zones()

Category → color map:

Severity → fill opacity:

• high — strong fill, use for high-confidence assessments
• medium — default, most zones should use this
• low — faint fill, for tentative notes

Analysis zone rules of engagement:

1. Do NOT spam the map. Only place a zone when you have something genuinely worth noting. A clean map is a useful map.
2. Write the body in your own voice — what you observed, what it might mean, and what you are NOT sure about. 2–6 sentences is ideal. Include uncertainty; the operator wants your reasoning, not a headline.
3. Match category to semantics — do not use warning for speculation, and do not use hypothesis for confirmed threats.
4. Prefer reactive placement over scheduled. Place zones in response to operator questions or events you spot while reviewing telemetry.
5. Clean up after yourself. If a zone you placed is no longer relevant, call delete_analysis_zone on it.
6. Pick a sensible cell_size_deg:
   • 0.3–0.8 — city-scale (neighborhood, metro, single base)
   • 1.0–2.0 — regional (country province, conflict zone)
   • 3.0–5.0 — strategic (full country, maritime theater)
7. Use ttl_hours for time-bound observations so the map self-cleans. Omit it for persistent assessments.

Message Signatures

ALL outbound messages MUST use the branded signature system:

from sb_signatures import sig

# Always start messages with the appropriate signature:
message = f"""{sig('brief')}
Morning Intelligence Digest — Apr 2, 2026 08:00
..."""

Decision Framework

When the user asks a question, follow this decision tree:

1. Is the SSE stream open?

   • If not → open sb.stream_updates() first. It tells you which layers have fresh data, pushes alerts instantly, and eliminates blind polling.

2. Does ShadowBroker have this data already?

   • Natural language → await sb.ask(question) (routes server-side)
   • Batch snapshot → await sb.run_playbook("hot_snapshot")
   • Known domain → find_entity, find_flights, find_ships, search_news, entities_near
   • Unknown domain → find_entity first; only then search_telemetry with confirm_expensive=true
   • Need specific layers → get_layer_slice(["military_flights", "gdelt"]) — only fetches layers that changed since your last call (per-layer incremental).
   • Near a location → entities_near() or get_near_me() (scans all layers within radius)
   • Full dump (rare) → get_telemetry() / get_slow_telemetry() only when targeted commands are insufficient. Always pass compact=true.

3. Does it need geocoding first?

   • User mentions a place name → geocode() first, then query with coordinates

4. Does it need external research?

   • Use your web browser to search, then geocode findings and place pins

5. Should I place pins?

   • YES if the answer has geographic locations
   • Use place_pin() for single locations, place_pins_batch() for multiple
   • Always include source URLs and confidence scores

6. Should I inject into native layers?

   • YES if the user explicitly wants data in a specific layer (CCTV, ships, etc.)
   • Use inject_data() — items tagged automatically for later removal

7. Should I set up persistent monitoring?

   • YES if the user wants ongoing tracking (aircraft, ship, geofence, keyword)
   • Use add_watch — alerts push instantly via SSE, no polling needed

8. Should I send an alert?

   • YES if the user has configured alert channels
   • Use the AlertDispatcher with the correct signature

Telegram rhetoric monitoring (watchdog)

Use watchdog watches for push alerts over SSE — no polling required. Keyword watches now scan Telegram OSINT too (translated and original text).

# Alert when "nuclear" appears in news, GDELT, or Telegram OSINT
await sb.send_command("add_watch", {
    "type": "keyword",
    "params": {"keyword": "nuclear", "include_telegram": True},
})

# Alert on new high-risk Telegram posts (LVL >= 7) — rhetoric/escalation monitor
await sb.send_command("add_watch", {
    "type": "telegram_rhetoric",
    "params": {"min_risk_score": 7, "channels": ["nexta_live", "war_monitor"]},
})

# Combine risk threshold + topic filter
await sb.send_command("add_watch", {
    "type": "telegram_rhetoric",
    "params": {"min_risk_score": 8, "keywords": ["crimea", "escalation", "missile"]},
})

When a watch fires, you receive an SSE alert event. Forward it with sb_alerts.send_intel() if the user has Discord/Telegram notification channels.

Important Rules

1. Open SSE stream first — call sb.stream_updates() at session start and keep it open. It pushes layer_changed events so you know exactly which layers to fetch, and delivers watchdog alerts instantly.
2. Fetch targeted, not everything — use get_layer_slice(), find_flights(), search_telemetry(), entities_near() instead of full get_telemetry() dumps. Per-layer incremental versioning means unchanged layers transfer zero bytes.
3. Always use signatures — every outbound message starts with the appropriate sig() prefix
4. Geocode before pinning — never guess coordinates, always use geocode()
5. Include sources — every pin should have source and source_url when available
6. Set confidence scores — 1.0 for verified/official data, 0.5-0.8 for web research, <0.5 for unverified
7. Set TTL for temporary pins — research results get ttl_hours=48, permanent infrastructure gets 0
8. Use batch for >3 commands — send_batch() runs up to 20 commands concurrently in one HTTP round-trip
9. Check summary first — use get_summary() before fetching full telemetry to save bandwidth
10. Tag injected data — the system auto-tags, but use descriptive source names