fix: complete intelligence layer refreshes and UAP fallback

This commit is contained in:
BigBodyCobain
2026-08-23 20:14:07 -06:00
parent 2461756da4
commit cd6395f5ee
11 changed files with 389 additions and 26 deletions
+16 -9
View File
@@ -326,6 +326,21 @@ class GT_EarlyWarning:
if item_id:
self._seen_item_ids.add(item_id)
# Keep geolocated feed items plottable even when they do not contain a
# costly-signal keyword. The prior placement below only stored coords
# after a positive classification, so most derived-OSINT regions were
# emitted at [0, 0] and correctly discarded by the map builder.
if isinstance(coords, (list, tuple)) and len(coords) >= 2:
try:
lat = float(coords[0])
lng = float(coords[1])
except (TypeError, ValueError):
pass
else:
with self._lock:
state = self._region_state(region)
state.coords = [lat, lng]
signals = self.classify_signals(text, source)
total_strength = float(sum(signals.values()))
@@ -362,14 +377,6 @@ class GT_EarlyWarning:
)
posteriors[domain] = posterior
if isinstance(coords, (list, tuple)) and len(coords) >= 2:
with self._lock:
state = self._region_state(region)
try:
state.coords = [float(coords[0]), float(coords[1])]
except (TypeError, ValueError):
pass
self._update_graph(region, entities, total_strength, coords if isinstance(coords, list) else None)
composite = self.composite_risk(region)
@@ -590,4 +597,4 @@ class GT_EarlyWarning:
"graph_nodes": self.G.number_of_nodes(),
"graph_edges": self.G.number_of_edges(),
"processed_items": len(self._seen_item_ids),
}
}
+22 -10
View File
@@ -2760,13 +2760,15 @@ async def lifespan(app: FastAPI):
# in _scheduler_loop, so we do NOT call it again in the preload thread.
start_carrier_tracker()
# Start SIGINT grid eagerly — APRS-IS TCP + Meshtastic MQTT connections
# take a few seconds to handshake and start receiving packets. By starting
# now, the bridges are already accumulating signals by the time the first
# fetch_sigint() reads them during the preload cycle.
from services.sigint_bridge import sigint_grid
# Route startup through the bounded production APRS bridge. The old
# SIGINTGrid.start() also opened an unbounded public APRS-IS feed.
from services.fetchers._store import is_any_active
from services.fetchers.sigint import _reconcile_sigint_bridges
sigint_grid.start()
_reconcile_sigint_bridges(
aprs_requested=is_any_active("sigint_aprs"),
mesh_requested=is_any_active("sigint_meshtastic"),
)
# Start Reticulum bridge (optional)
try:
@@ -2898,7 +2900,11 @@ async def lifespan(app: FastAPI):
stop_scheduler()
stop_carrier_tracker()
try:
from services.aprs_is_bridge import aprs_is_bridge
from services.sigint_bridge import sigint_grid
sigint_grid.stop()
aprs_is_bridge.stop()
except Exception:
pass
if not _MESH_ONLY:
@@ -4028,6 +4034,7 @@ async def update_layers(update: LayerUpdate, request: Request):
# Start/stop SIGINT bridges on transition
from services.sigint_bridge import sigint_grid
from services.aprs_is_bridge import aprs_is_bridge
if old_mesh and not new_mesh:
try:
@@ -4058,16 +4065,21 @@ async def update_layers(update: LayerUpdate, request: Request):
)
if old_aprs and not new_aprs:
sigint_grid.aprs.stop()
logger.info("APRS bridge stopped (layer disabled)")
aprs_is_bridge.reconcile(False)
logger.info("Bounded APRS-IS bridge stopped (layer disabled)")
elif not old_aprs and new_aprs:
sigint_grid.aprs.start()
logger.info("APRS bridge started (layer enabled)")
aprs_is_bridge.reconcile(True)
logger.info("Bounded APRS-IS bridge reconciled (layer enabled)")
if not old_viirs and new_viirs:
_queue_viirs_change_refresh()
logger.info("VIIRS change refresh queued (layer enabled)")
if old_mesh != new_mesh or old_aprs != new_aprs:
from services.fetchers.sigint import fetch_sigint
threading.Thread(target=fetch_sigint, daemon=True, name="sigint-layer-refresh").start()
refresh_newly_enabled_layers(layers_before)
return {"status": "ok"}
+15 -4
View File
@@ -611,6 +611,7 @@ async def update_layers(update: LayerUpdate, request: Request):
start_ais_stream()
logger.info("AIS stream started (ship layer enabled)")
from services.sigint_bridge import sigint_grid
from services.aprs_is_bridge import aprs_is_bridge
if old_mesh and not new_mesh:
try:
from services.meshtastic_mqtt_settings import mqtt_bridge_enabled
@@ -637,14 +638,24 @@ async def update_layers(update: LayerUpdate, request: Request):
"(set MESH_MQTT_ENABLED=true to participate in the public broker)"
)
if old_aprs and not new_aprs:
sigint_grid.aprs.stop()
logger.info("APRS bridge stopped (layer disabled)")
# #533/#534: never start or stop the legacy SIGINTGrid APRS client.
# It used an unbounded public APRS-IS filter. The replacement bridge
# validates bounded operator configuration and fails closed.
aprs_is_bridge.reconcile(False)
logger.info("Bounded APRS-IS bridge stopped (layer disabled)")
elif not old_aprs and new_aprs:
sigint_grid.aprs.start()
logger.info("APRS bridge started (layer enabled)")
aprs_is_bridge.reconcile(True)
logger.info("Bounded APRS-IS bridge reconciled (layer enabled)")
if not old_viirs and new_viirs:
_queue_viirs_change_refresh()
logger.info("VIIRS change refresh queued (layer enabled)")
if old_mesh != new_mesh or old_aprs != new_aprs:
# Publish a single merged snapshot immediately so enabling both radio
# layers behaves like one "scan all" action while the bridges continue
# receiving independently in their own bounded lifecycles.
from services.fetchers.sigint import fetch_sigint
threading.Thread(target=fetch_sigint, daemon=True, name="sigint-layer-refresh").start()
refresh_newly_enabled_layers(layers_before)
return {"status": "ok"}
+43 -1
View File
@@ -878,6 +878,40 @@ def _load_nuforc_sightings_cache(*, force_refresh: bool = False) -> list[dict] |
return None
def _load_nuforc_last_known_sightings() -> list[dict] | None:
"""Return the last non-empty snapshot for display when refresh is unavailable.
This deliberately skips freshness and rolling-window checks. It is only a
stale-data safety net after a refresh has produced no usable rows; the
normal cache loader remains strict so stale data is never presented as
fresh.
"""
with _data_lock:
current = latest_data.get("uap_sightings")
if isinstance(current, list):
in_memory = [row.copy() for row in current if isinstance(row, dict)]
if in_memory:
return in_memory
if not _NUFORC_SIGHTINGS_CACHE_FILE.exists():
return None
try:
raw = json.loads(_NUFORC_SIGHTINGS_CACHE_FILE.read_text(encoding="utf-8"))
sightings = raw.get("sightings")
if not isinstance(sightings, list):
return None
last_known = [row.copy() for row in sightings if isinstance(row, dict)]
if last_known:
logger.warning(
"UAP sightings: using %d last-known cached reports because the refresh returned no usable rows",
len(last_known),
)
return last_known or None
except Exception as e:
logger.warning("UAP sightings: last-known cache load error: %s", e)
return None
def _save_nuforc_sightings_cache(sightings: list[dict]) -> None:
if not sightings:
logger.warning("UAP sightings: refusing to save empty daily cache")
@@ -1693,9 +1727,17 @@ def fetch_uap_sightings(*, force_refresh: bool = False):
if sightings:
sightings = _filter_uap_sightings_recent(sightings)
fresh_snapshot = bool(sightings)
if not sightings:
sightings = _load_nuforc_last_known_sightings()
if sightings:
logger.warning(
"UAP sightings: retaining last-known snapshot; current refresh did not produce usable recent reports"
)
with _data_lock:
latest_data["uap_sightings"] = sightings or []
if sightings:
if fresh_snapshot:
_mark_fresh("uap_sightings")
return
+48 -1
View File
@@ -17,7 +17,18 @@ _INSTANT_LAYER_KEYS: frozenset[str] = frozenset(
# Background — network-bound OR large local scans (full CCTV SELECT can stall
# the single uvicorn worker if run inline on enable).
_SLOW_LAYER_KEYS: frozenset[str] = frozenset(
{"cctv", "firms", "psk_reporter", "fishing_activity"}
{
"cctv",
"firms",
"psk_reporter",
"fishing_activity",
"uap_sightings",
"malware_c2",
"cyber_threats",
"scm_suppliers",
"telegram_osint",
"gt_risk",
}
)
@@ -85,6 +96,42 @@ def _slow_fetch(key: str) -> None:
fetch_fishing_activity()
logger.info("Fishing activity loaded (layer enabled)")
return
if key == "uap_sightings":
from services.fetchers.earth_observation import fetch_uap_sightings
fetch_uap_sightings()
logger.info("UAP sightings loaded (layer enabled)")
return
if key == "malware_c2":
from services.fetchers.malware import fetch_malware_threats
fetch_malware_threats()
logger.info("Malware C2 loaded (layer enabled)")
return
if key == "cyber_threats":
from services.fetchers.cyber_status import fetch_cyber_threats
fetch_cyber_threats()
logger.info("Cyber threats loaded (layer enabled)")
return
if key == "scm_suppliers":
from services.scm.suppliers import fetch_scm_suppliers
fetch_scm_suppliers()
logger.info("SCM suppliers loaded (layer enabled)")
return
if key == "telegram_osint":
from services.fetchers.telegram_osint import fetch_telegram_osint
fetch_telegram_osint()
logger.info("Telegram OSINT loaded (layer enabled)")
return
if key == "gt_risk":
from analytics.integration import maybe_refresh_gt_analytics
maybe_refresh_gt_analytics()
logger.info("Strategic Risk Analytics refreshed (layer enabled)")
return
raise KeyError(key)
@@ -2,6 +2,7 @@
from __future__ import annotations
import inspect
from unittest.mock import MagicMock
import pytest
@@ -199,3 +200,13 @@ def test_fetch_path_never_calls_legacy_grid_start(monkeypatch: pytest.MonkeyPatc
sigint_fetcher.fetch_sigint()
legacy_start.assert_not_called()
def test_layer_toggle_does_not_start_legacy_global_aprs_client() -> None:
"""The layer endpoint must route APRS through the bounded bridge only."""
from routers import data as data_router
source = inspect.getsource(data_router.update_layers)
assert "sigint_grid.aprs.start" not in source
assert "sigint_grid.aprs.stop" not in source
assert "aprs_is_bridge.reconcile" in source
+18 -1
View File
@@ -88,6 +88,23 @@ def test_heatmap_returns_geojson_features(engine: GT_EarlyWarning) -> None:
assert feature["geometry"]["type"] == "Point"
def test_heatmap_keeps_geolocated_items_without_costly_signal(engine: GT_EarlyWarning) -> None:
"""Coordinates must survive even when the item only contributes baseline risk."""
engine.process_feed_item(
{
"id": "geo-baseline-1",
"text": "Routine regional update.",
"source": "news",
"region": "lisbon",
"coords": [38.72, -9.14],
}
)
features = engine.get_risk_heatmap()["features"]
assert features
assert features[0]["geometry"]["coordinates"] == [-9.14, 38.72]
def test_dossier_includes_recent_signals(engine: GT_EarlyWarning) -> None:
engine.process_feed_item(
{
@@ -147,4 +164,4 @@ def test_refresh_from_latest_data_processes_telegram(monkeypatch: pytest.MonkeyP
}
summary = refresh_from_latest_data(latest, persist=False)
assert summary["enabled"] is True
assert summary["processed"] >= 1
assert summary["processed"] >= 1
@@ -88,3 +88,58 @@ def test_cctv_enable_reuses_nonempty_catalog():
count.assert_called_once()
seed.assert_not_called()
fetch_cctv.assert_called_once()
def test_refreshes_intelligence_layers_on_enable():
"""Cold optional feeds must not wait for the next slow-tier scheduler tick."""
keys = {
"uap_sightings",
"malware_c2",
"cyber_threats",
"scm_suppliers",
"telegram_osint",
"gt_risk",
}
before = snapshot_active_layers()
for key in keys:
active_layers[key] = True
try:
with patch("services.data_fetcher._SLOW_EXECUTOR") as slow_exec:
refresh_newly_enabled_layers({**before, **{key: False for key in keys}})
slow_exec.submit.assert_called_once()
assert set(slow_exec.submit.call_args.args[1]) == keys
finally:
for key in keys:
active_layers[key] = before.get(key, False)
def test_slow_fetch_supports_intelligence_layers_without_network():
"""Each newly supported layer dispatches to its existing fetcher."""
from services.layer_enable_refresh import _slow_fetch
with (
patch("services.fetchers.earth_observation.fetch_uap_sightings") as uap,
patch("services.fetchers.malware.fetch_malware_threats") as malware,
patch("services.fetchers.cyber_status.fetch_cyber_threats") as cyber,
patch("services.scm.suppliers.fetch_scm_suppliers") as scm,
patch("services.fetchers.telegram_osint.fetch_telegram_osint") as telegram,
patch("analytics.integration.maybe_refresh_gt_analytics") as gt,
):
for key in (
"uap_sightings",
"malware_c2",
"cyber_threats",
"scm_suppliers",
"telegram_osint",
"gt_risk",
):
_slow_fetch(key)
uap.assert_called_once_with()
malware.assert_called_once_with()
cyber.assert_called_once_with()
scm.assert_called_once_with()
telegram.assert_called_once_with()
gt.assert_called_once_with()
@@ -226,6 +226,32 @@ def test_fetch_uap_sightings_succeeds_when_fallback_returns_data(monkeypatch):
assert canary_calls == [], "canary should not trip when fallback supplies data"
def test_fetch_uap_sightings_retains_last_known_snapshot_when_refresh_empty(monkeypatch):
"""An empty weekly refresh must not erase the last visible UAP snapshot."""
from services.fetchers import earth_observation as eo
from services.fetchers import _store
monkeypatch.setattr(_store, "is_any_active", lambda layer: True)
monkeypatch.setattr(eo, "_load_nuforc_sightings_cache", lambda force_refresh=False: None)
monkeypatch.setattr(
eo,
"_build_recent_uap_sightings",
lambda: (_ for _ in ()).throw(RuntimeError("NUFORC unavailable")),
)
monkeypatch.setattr(eo, "_build_uap_sightings_from_hf_mirror", lambda: [])
last_known = [{"id": "older-uap", "date_time": "2025-01-01", "lat": 39.7, "lng": -104.9}]
monkeypatch.setattr(eo, "_load_nuforc_last_known_sightings", lambda: last_known)
monkeypatch.setattr(eo, "_mark_fresh", lambda *keys: None)
with _store._data_lock:
_store.latest_data["uap_sightings"] = []
eo.fetch_uap_sightings()
with _store._data_lock:
assert _store.latest_data["uap_sightings"] == last_known
def test_uap_scheduler_runs_weekly():
"""UAP layer refreshes weekly so each install pulls live NUFORC on a steady cadence."""
from services import data_fetcher
+2
View File
@@ -19,6 +19,7 @@ import { endInfonetTerminalSession } from '@/lib/infonetTerminalSession';
import ShodanPanel from '@/components/ShodanPanel';
import ReconPanel from '@/components/ReconPanel';
import ScmPanel from '@/components/ScmPanel';
import CyberThreatPanel from '@/components/CyberThreatPanel';
import EntityGraphPanel from '@/components/EntityGraphPanel';
import { isEntityGraphEligible } from '@/lib/entityGraph';
import AIIntelPanel from '@/components/AIIntelPanel';
@@ -693,6 +694,7 @@ export default function Dashboard() {
<div className="contents" style={{ direction: 'ltr' }}>
<ReconPanel />
<ScmPanel layerEnabled={activeLayers.scm_suppliers} />
<CyberThreatPanel layerEnabled={activeLayers.cyber_threats} />
</div>
)}
@@ -0,0 +1,133 @@
'use client';
import React, { useCallback, useEffect, useState } from 'react';
import { AlertTriangle, Minus, Plus, RefreshCw } from 'lucide-react';
import { API_BASE } from '@/lib/api';
import { useTranslation } from '@/i18n';
interface CyberThreat {
id: string;
name: string;
vendor?: string;
product?: string;
severity?: string;
date?: string;
due?: string;
source?: string;
}
interface CyberPayload {
threats: CyberThreat[];
stats?: {
active_cves?: number;
threat_level?: string;
cisa_total?: number;
};
}
interface Props {
layerEnabled?: boolean;
}
export default function CyberThreatPanel({ layerEnabled = false }: Props) {
const { t } = useTranslation();
const [isMinimized, setIsMinimized] = useState(true);
const [data, setData] = useState<CyberPayload | null>(null);
const [loading, setLoading] = useState(false);
const refresh = useCallback(async () => {
if (!layerEnabled) {
setData(null);
return;
}
setLoading(true);
try {
const res = await fetch(`${API_BASE}/api/cyber-threats`, { cache: 'no-store' });
if (res.ok) setData(await res.json());
} catch {
/* non-fatal */
} finally {
setLoading(false);
}
}, [layerEnabled]);
useEffect(() => {
refresh();
if (!layerEnabled) return undefined;
const id = setInterval(refresh, 5 * 60_000);
return () => clearInterval(id);
}, [refresh, layerEnabled]);
const threats = data?.threats || [];
const threatLevel = data?.stats?.threat_level || '—';
return (
<div className="pointer-events-auto flex-shrink-0 border border-amber-700/40 bg-black/75 backdrop-blur-sm shadow-[0_0_18px_rgba(245,158,11,0.10)]">
<div
className="flex items-center justify-between border-b border-amber-700/30 bg-amber-950/20 px-3 py-2.5 cursor-pointer hover:bg-amber-950/40 transition-colors"
onClick={() => setIsMinimized((prev) => !prev)}
>
<div className="flex items-center gap-2">
<AlertTriangle size={16} className="text-amber-400" />
<span className="text-[12px] font-mono font-bold tracking-widest text-amber-400">
{t('layers.cyberThreats').toUpperCase()}
</span>
{layerEnabled && threats.length > 0 && (
<span className="text-[11px] font-mono px-1.5 py-0.5 bg-red-900/30 border border-red-700/40 text-red-300 tracking-wider">
{threats.length} ACTIVE
</span>
)}
</div>
<div className="flex items-center gap-2">
<button
type="button"
onClick={(e) => {
e.stopPropagation();
refresh();
}}
title="Refresh CISA KEV overlay"
className="text-amber-600 transition-colors hover:text-amber-400 p-0.5"
>
<RefreshCw size={11} className={loading ? 'animate-spin' : ''} />
</button>
{isMinimized ? <Plus size={16} className="text-amber-400" /> : <Minus size={16} className="text-amber-400" />}
</div>
</div>
{!isMinimized && (
<div className="px-3 py-2 max-h-52 overflow-y-auto styled-scrollbar space-y-1.5">
{!layerEnabled ? (
<div className="text-[11px] font-mono tracking-wider text-amber-600/70 py-1">
Enable the Cyber Threats layer to load CISA KEV.
</div>
) : threats.length === 0 ? (
<div className="text-[11px] font-mono tracking-wider text-amber-500/80 py-1">
No CISA KEV additions in the last 30 days.
</div>
) : (
<>
<div className="text-[10px] font-mono tracking-widest text-amber-500/80 pb-1">
THREAT LEVEL: {threatLevel}
</div>
{threats.map((threat) => (
<div key={threat.id} className="border border-amber-700/30 bg-amber-950/15 px-2 py-1.5">
<div className="text-[11px] font-mono font-bold tracking-wide text-amber-200 leading-tight">
{threat.id}
</div>
<div className="text-[10px] font-mono text-amber-500/90 mt-0.5 leading-tight">
{threat.name}
</div>
{(threat.vendor || threat.product) && (
<div className="text-[10px] font-mono text-amber-600/80 mt-0.5">
{[threat.vendor, threat.product].filter(Boolean).join(' · ')}
</div>
)}
</div>
))}
</>
)}
</div>
)}
</div>
);
}