ShadowBroker Intelligence Client

# 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}},
])

from sb_query import ShadowBrokerClient
sb = ShadowBrokerClient()  # auto-detects local or remote mode

SHADOWBROKER_URL=https://your-shadowbroker-host:8000
SHADOWBROKER_HMAC_SECRET=your-hmac-secret-here

HMAC-SHA256(secret, METHOD|path|timestamp|nonce|sha256(body))

python openclaw-skills/shadowbroker/verify_hmac.py

# 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']}")

# 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']}")

# 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})

# 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

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

# 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"}]}

# 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

# 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

# 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

# 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")

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")

# 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()