From cd6395f5ee7d5232fc7b553bf74f5e514deaafdb Mon Sep 17 00:00:00 2001 From: BigBodyCobain <43977454+BigBodyCobain@users.noreply.github.com> Date: Sun, 23 Aug 2026 20:14:07 -0600 Subject: [PATCH] fix: complete intelligence layer refreshes and UAP fallback --- backend/analytics/gt_early_warning.py | 25 ++-- backend/main.py | 32 +++-- backend/routers/data.py | 19 ++- .../services/fetchers/earth_observation.py | 44 +++++- backend/services/layer_enable_refresh.py | 49 ++++++- backend/tests/test_aprs_is_resource_safety.py | 11 ++ backend/tests/test_gt_early_warning.py | 19 ++- backend/tests/test_layer_enable_refresh.py | 55 ++++++++ backend/tests/test_uap_hf_fallback_cutoff.py | 26 ++++ frontend/src/app/page.tsx | 2 + frontend/src/components/CyberThreatPanel.tsx | 133 ++++++++++++++++++ 11 files changed, 389 insertions(+), 26 deletions(-) create mode 100644 frontend/src/components/CyberThreatPanel.tsx diff --git a/backend/analytics/gt_early_warning.py b/backend/analytics/gt_early_warning.py index ae5c315..6b0e4dd 100644 --- a/backend/analytics/gt_early_warning.py +++ b/backend/analytics/gt_early_warning.py @@ -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), - } \ No newline at end of file + } diff --git a/backend/main.py b/backend/main.py index de567ff..e3b46a1 100644 --- a/backend/main.py +++ b/backend/main.py @@ -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"} diff --git a/backend/routers/data.py b/backend/routers/data.py index 9a5e131..5dff773 100644 --- a/backend/routers/data.py +++ b/backend/routers/data.py @@ -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"} diff --git a/backend/services/fetchers/earth_observation.py b/backend/services/fetchers/earth_observation.py index 257ed2a..f99686a 100644 --- a/backend/services/fetchers/earth_observation.py +++ b/backend/services/fetchers/earth_observation.py @@ -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 diff --git a/backend/services/layer_enable_refresh.py b/backend/services/layer_enable_refresh.py index 336be52..22aafaa 100644 --- a/backend/services/layer_enable_refresh.py +++ b/backend/services/layer_enable_refresh.py @@ -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) diff --git a/backend/tests/test_aprs_is_resource_safety.py b/backend/tests/test_aprs_is_resource_safety.py index 297f501..73f292e 100644 --- a/backend/tests/test_aprs_is_resource_safety.py +++ b/backend/tests/test_aprs_is_resource_safety.py @@ -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 diff --git a/backend/tests/test_gt_early_warning.py b/backend/tests/test_gt_early_warning.py index a094c37..536bfda 100644 --- a/backend/tests/test_gt_early_warning.py +++ b/backend/tests/test_gt_early_warning.py @@ -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 \ No newline at end of file + assert summary["processed"] >= 1 diff --git a/backend/tests/test_layer_enable_refresh.py b/backend/tests/test_layer_enable_refresh.py index 061093d..f5eeb16 100644 --- a/backend/tests/test_layer_enable_refresh.py +++ b/backend/tests/test_layer_enable_refresh.py @@ -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() diff --git a/backend/tests/test_uap_hf_fallback_cutoff.py b/backend/tests/test_uap_hf_fallback_cutoff.py index 54fa7cf..c4a9467 100644 --- a/backend/tests/test_uap_hf_fallback_cutoff.py +++ b/backend/tests/test_uap_hf_fallback_cutoff.py @@ -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 diff --git a/frontend/src/app/page.tsx b/frontend/src/app/page.tsx index dca49ae..a0a7cf3 100644 --- a/frontend/src/app/page.tsx +++ b/frontend/src/app/page.tsx @@ -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() {
+
)} diff --git a/frontend/src/components/CyberThreatPanel.tsx b/frontend/src/components/CyberThreatPanel.tsx new file mode 100644 index 0000000..c2c5020 --- /dev/null +++ b/frontend/src/components/CyberThreatPanel.tsx @@ -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(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 ( +
+
setIsMinimized((prev) => !prev)} + > +
+ + + {t('layers.cyberThreats').toUpperCase()} + + {layerEnabled && threats.length > 0 && ( + + {threats.length} ACTIVE + + )} +
+
+ + {isMinimized ? : } +
+
+ + {!isMinimized && ( +
+ {!layerEnabled ? ( +
+ Enable the Cyber Threats layer to load CISA KEV. +
+ ) : threats.length === 0 ? ( +
+ No CISA KEV additions in the last 30 days. +
+ ) : ( + <> +
+ THREAT LEVEL: {threatLevel} +
+ {threats.map((threat) => ( +
+
+ {threat.id} +
+
+ {threat.name} +
+ {(threat.vendor || threat.product) && ( +
+ {[threat.vendor, threat.product].filter(Boolean).join(' · ')} +
+ )} +
+ ))} + + )} +
+ )} +
+ ); +}