add learn folder to IM MONITORING THE SITUATION DASHBOARD project
This commit is contained in:
CarterPerez-dev
parent
5e7f2c36de
commit
1b9fcbac13
|
|
@ -8,3 +8,13 @@
|
|||
DEMO-TRACKER.md
|
||||
|
||||
PROJECTS/beginner/hash-cracker/docs/
|
||||
PROJECTS/advanced/monitor-the-situation-dashboard/docs/
|
||||
|
||||
# Python virtualenvs and caches
|
||||
.venv/
|
||||
venv/
|
||||
__pycache__/
|
||||
*.pyc
|
||||
.mypy_cache/
|
||||
.pytest_cache/
|
||||
.ruff_cache/
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
// ©AngelaMos | 2026
|
||||
// BottomTicker.tsx
|
||||
|
||||
import { useEffect, useLayoutEffect, useRef } from 'react'
|
||||
import { useLayoutEffect, useRef } from 'react'
|
||||
import { useTicker } from '@/stores/ticker'
|
||||
import { useUIStore } from '@/stores/ui'
|
||||
import styles from './BottomTicker.module.scss'
|
||||
|
|
@ -27,17 +27,21 @@ export function BottomTicker(): React.ReactElement | null {
|
|||
useLayoutEffect(() => {
|
||||
const el = trackRef.current
|
||||
if (!el) return
|
||||
el.style.animationDuration = `${measureDuration(el)}s`
|
||||
}, [])
|
||||
|
||||
useEffect(() => {
|
||||
const onResize = (): void => {
|
||||
const el = trackRef.current
|
||||
if (!el) return
|
||||
const update = (): void => {
|
||||
el.style.animationDuration = `${measureDuration(el)}s`
|
||||
}
|
||||
window.addEventListener('resize', onResize)
|
||||
return () => window.removeEventListener('resize', onResize)
|
||||
|
||||
update()
|
||||
|
||||
const ro = new ResizeObserver(update)
|
||||
ro.observe(el)
|
||||
window.addEventListener('resize', update)
|
||||
|
||||
return () => {
|
||||
ro.disconnect()
|
||||
window.removeEventListener('resize', update)
|
||||
}
|
||||
}, [])
|
||||
|
||||
if (isPresentation) return null
|
||||
|
|
|
|||
|
|
@ -9,6 +9,8 @@ import { Panel } from './Panel'
|
|||
|
||||
const OUTAGE_ROW_LIMIT = 6
|
||||
const REGIME_CC = 'IR'
|
||||
const PERSIA_LABEL = 'Persia'
|
||||
const REGIME_LABEL = 'regime 👎'
|
||||
const FLASH_DURATION_MS = 600
|
||||
const STALE_AFTER_MS = 1_800_000
|
||||
const MS_PER_HOUR = 3_600_000
|
||||
|
|
@ -79,7 +81,7 @@ export function OutagePanel(): React.ReactElement {
|
|||
title={o.reason ?? o.outageType ?? ''}
|
||||
>
|
||||
{isRegime(o.locations)
|
||||
? 'Regime 👎'
|
||||
? REGIME_LABEL
|
||||
: fmtCause(o.reason, o.outageType)}
|
||||
</td>
|
||||
<td className={styles.state}>{fmtState(o.endDate)}</td>
|
||||
|
|
@ -101,7 +103,8 @@ function isRegime(locations: string[] | undefined): boolean {
|
|||
|
||||
function fmtCC(locations: string[] | undefined): string {
|
||||
if (!locations || locations.length === 0) return '—'
|
||||
const first = locations[0] ?? '—'
|
||||
const raw = locations[0] ?? '—'
|
||||
const first = raw === REGIME_CC ? PERSIA_LABEL : raw
|
||||
if (locations.length === 1) return first
|
||||
return `${first} +${locations.length - 1}`
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,118 @@
|
|||
<!--
|
||||
©AngelaMos | 2026
|
||||
00-OVERVIEW.md
|
||||
-->
|
||||
|
||||
# Monitor the Situation Dashboard
|
||||
|
||||
## What This Is
|
||||
|
||||
A real-time situational awareness dashboard that fuses eleven live intelligence feeds (mass-scan firehose, BGP hijacks, internet outages, fresh CVEs with EPSS scoring, CISA KEV diff, ransomware victim leak posts, M2.5+ earthquakes, NOAA space weather, ISS orbital track, Wikipedia ITN, GDELT spikes, plus BTC/ETH ticks) into a single SOC view. A single Go binary runs every collector as an `errgroup` goroutine, feeds events through an in-process bus, and fans them out to browsers over WebSocket.
|
||||
|
||||
The phrase "monitoring the situation" started as a Twitter/X meme in mid-2025. This is the build that actually monitors the situation.
|
||||
|
||||
## Why This Matters
|
||||
|
||||
Most "dashboards" you see are static Grafana charts pointed at one Prometheus job. Real situational awareness, the kind a SOC actually uses, requires fusing wildly different feeds at wildly different cadences and presenting them coherently to a human under stress. That is hard.
|
||||
|
||||
A few real incidents this kind of dashboard would have caught minutes earlier than reading Twitter:
|
||||
|
||||
- **June 2024 Cloudflare BGP hijack** by Eletel (AS267613) for Orange España traffic. CF Radar published the hijack confidence score within minutes. A dashboard polling that endpoint surfaces it before the news cycle catches up.
|
||||
- **Log4Shell, December 2021 (CVE-2021-44228)**. EPSS hit 0.97 within hours of disclosure, KEV listing followed shortly after. A CVE velocity panel weighted by EPSS and KEV diff is exactly how you spot "this one matters" in a flood of CVEs.
|
||||
- **2024 CrowdStrike outage**. Cloudflare Radar's outage feed flagged abnormal connectivity dips per ASN before anyone identified the root cause. An operator with the outage panel open sees something is wrong before they know what.
|
||||
|
||||
The point of this project is to teach how live data infrastructure for security operations actually gets built: the collector pattern, the bus pattern, WebSocket fan-out, and how to keep all of that observable and correct at three orders of magnitude of cadence variation (sub-second to daily).
|
||||
|
||||
## What You'll Learn
|
||||
|
||||
**Security and operations theory**
|
||||
|
||||
- **Threat intel feed economics**: which feeds actually cost money (NVD API key throttling, AbuseIPDB lookups), which are free but fragile (DShield scraping), which are commercial but worth it (Cloudflare Radar). What latency budget each one has.
|
||||
- **EPSS vs CVSS**: why "Critical CVSS 9.8" is not the same as "this will be exploited tomorrow" and why EPSS percentile is the correct prioritization signal for an operator's first 30 minutes of attention.
|
||||
- **CISA KEV**: what Known Exploited Vulnerabilities actually means, why a KEV diff alert is the loudest single signal a practitioner can subscribe to, and how it correlates (or doesn't) with EPSS.
|
||||
- **BGP hijacks vs route leaks**: how MOAS (multiple origin AS) detection works, why confidence scoring exists, and what an ASN really represents on the global routing table.
|
||||
- **Why coordinate transforms matter**: ISS position needs SGP4 propagation from a TLE, not a polling endpoint, because the polling endpoint is `wheretheiss.at` and the TLE source is CelesTrak, and they update at different cadences. Mixing them naively gives you teleporting satellites.
|
||||
|
||||
**Distributed systems patterns**
|
||||
|
||||
- **In-process event bus** as a pragmatic alternative to Kafka or NATS when your scale is "one box, eleven producers, one fan-out". The single-binary architecture is a feature, not a constraint.
|
||||
- **Backpressure and lossy fan-out**: when to drop events, when to block the producer, and why slow WebSocket subscribers must never block collector goroutines.
|
||||
- **Snapshot-then-stream protocol**: how a connecting browser gets the current world state via REST and then transitions cleanly to WebSocket without missing or double-counting events.
|
||||
- **BRIN indexes for append-mostly time-series in Postgres**, and why you reach for them before partitioning.
|
||||
- **Single-binary `errgroup` orchestration**: how to start eleven heterogeneous workers, propagate cancellation, and clean up on `SIGTERM` without writing a process supervisor.
|
||||
|
||||
**Auth and crypto**
|
||||
|
||||
- **JWT with auto-rotated Ed25519 (ES256)**: why asymmetric signing belongs in front of any public dashboard, how `JWKS` exposure works, and why it lets you change keys without reissuing every active token.
|
||||
- **AES-256-GCM at rest** for webhook secrets so a database leak does not become a Discord/Slack/Telegram pwn of every alerted user.
|
||||
- **Argon2id** for password hashing instead of bcrypt, with the parameter tuning rationale.
|
||||
|
||||
**Frontend telemetry**
|
||||
|
||||
- A 3D MapLibre globe with per-country outage shading, ASN dot density for mass-scan sources, ransomware victim markers, earthquake epicenters, and a live ISS track propagated on the client.
|
||||
- TanStack Query v5 + Zustand for the snapshot/stream split: REST for cold start, WebSocket pushes into the same store, components subscribe to selectors.
|
||||
|
||||
## Prerequisites
|
||||
|
||||
**You should be comfortable with:**
|
||||
|
||||
- **Go** — goroutines, channels, contexts, `errgroup`, the `context.Context` cancellation model. If you have never written `select { case <-ctx.Done(): ... }` you will be lost.
|
||||
- **PostgreSQL basics** — `CREATE INDEX`, `EXPLAIN`, JSONB columns. Not advanced query planning, just enough to read a migration.
|
||||
- **HTTP and WebSocket fundamentals** — how a WS upgrade handshake works, what ping/pong frames are for, what backpressure means.
|
||||
- **TypeScript and React** — hooks, state management. You do not need MapLibre or D3 expertise; the project uses them but the patterns are explained.
|
||||
|
||||
**Nice to have but not required:**
|
||||
|
||||
- BGP and ASN intuition (CIDR, prefix length, "what is AS15169"). Read the [Cloudflare Radar BGP docs](https://developers.cloudflare.com/radar/) if you want a primer.
|
||||
- SGP4/TLE astrodynamics for the ISS panel (the project uses `joshuaferrara/go-satellite` so you do not have to implement it).
|
||||
- CEL (Common Expression Language) for the alert rule predicates. The Google `cel-go` README is the right starting point.
|
||||
|
||||
**Tools:**
|
||||
|
||||
- Go 1.25+
|
||||
- Node 20+ and `pnpm`
|
||||
- Docker and Docker Compose
|
||||
- `just` ([install](https://github.com/casey/just)) — every recipe is in the `Justfile`
|
||||
- Optional: an NVD API key (raises CVE poll budget from 5 req/30s to 50 req/30s) and a Cloudflare API token (`CF_RADAR_TOKEN`) for the Radar feeds.
|
||||
|
||||
## Quick Start
|
||||
|
||||
```bash
|
||||
cd PROJECTS/advanced/monitor-the-situation-dashboard
|
||||
|
||||
# Optional: fill in API keys (nothing is required to boot)
|
||||
cp .env.example .env
|
||||
|
||||
# Bring up postgres, redis, backend, frontend in dev mode
|
||||
just dev-up
|
||||
|
||||
# In another shell, run migrations
|
||||
just migrate
|
||||
```
|
||||
|
||||
Visit `http://localhost:8432`. You should see the globe initialize, then panels populate one by one as their first poll cycle returns. The heartbeat panel shows green within five seconds.
|
||||
|
||||
If you see a `NO SIGNAL` ribbon, the WebSocket failed. Check `just logs backend` — the most common cause is `notifications: encryption key not set`, which disables the alert engine but not the dashboard itself.
|
||||
|
||||
## Expected Output
|
||||
|
||||
Within about a minute of `just dev-up`:
|
||||
|
||||
- **Globe panel** rotating, with the ISS track drawn in a polar orbit. Earthquake dots appear as USGS publishes them (typically 5-30 per hour worldwide above M2.5).
|
||||
- **CVE velocity panel** populated with the last two hours of NVD entries, sorted by EPSS. Several rows tagged `KEV` if any made the catalog recently.
|
||||
- **DShield panel** showing top scanning ASNs. If you waited the full hour for the first poll, the panel shows the top 10 sources for mass scanning by report volume.
|
||||
- **BTC/ETH panels** ticking live (Coinbase has the lowest cadence in the system, sub-second).
|
||||
- **Outage panel** quiet most days. Cloudflare Radar surfaces something interesting roughly weekly.
|
||||
|
||||
## Where To Go Next
|
||||
|
||||
This file is the orientation. The actual learning material is split across:
|
||||
|
||||
| File | Focus | Read time |
|
||||
|---|---|---|
|
||||
| [01-CONCEPTS.md](01-CONCEPTS.md) | Threat intel theory: EPSS, KEV, BGP hijacks, mass-scan firehoses, situational awareness as a discipline | 20 min |
|
||||
| [02-ARCHITECTURE.md](02-ARCHITECTURE.md) | Single-binary collector pipeline, event bus, snapshot store, WebSocket fan-out, BRIN time-series storage | 30 min |
|
||||
| [03-IMPLEMENTATION.md](03-IMPLEMENTATION.md) | Code walkthrough across collectors, bus, ws, alerts engine, snapshot REST, frontend store wiring | 45 min |
|
||||
| [04-CHALLENGES.md](04-CHALLENGES.md) | Extension ideas, extra feeds to wire in, alert engine improvements, deployment hardening | 15 min |
|
||||
|
||||
Read in order if this is your first encounter with the project. Skip to `03-IMPLEMENTATION` if you have already deployed similar systems and want to compare patterns.
|
||||
|
|
@ -0,0 +1,187 @@
|
|||
<!--
|
||||
©AngelaMos | 2026
|
||||
01-CONCEPTS.md
|
||||
-->
|
||||
|
||||
# Concepts
|
||||
|
||||
Before walking through code, you need to understand the security and operations theory the dashboard is built around. Skip this if you already know what EPSS percentile means, what a BGP hijack actually is, and why a SOC analyst with one screen needs a different feed mix than one with eight.
|
||||
|
||||
## Situational Awareness as a Discipline
|
||||
|
||||
The military origin of "situational awareness" (Endsley, 1995) breaks the term into three levels:
|
||||
|
||||
1. **Perception** — what is happening right now in the environment.
|
||||
2. **Comprehension** — what those raw observations mean together.
|
||||
3. **Projection** — what is likely to happen next.
|
||||
|
||||
Most security dashboards stop at level 1. They show alerts, traffic graphs, and CVE counts. A useful dashboard has to make level 2 cheap (visual fusion across feeds) and at least nudge toward level 3 (KEV diffs, EPSS climbing curves, ransomware leak posts that historically precede public breach disclosure by 3-7 days).
|
||||
|
||||
The eleven feeds in this project are chosen so that no single page-of-screen view requires the operator to mentally cross-reference. The globe spatializes ASN-bound data (outages, hijacks, scan sources) on the same surface as physical events (earthquakes, ISS, ransomware victims by country). The CVE panel shows EPSS bars next to KEV badges so "should I care?" is a visual decision, not a database query.
|
||||
|
||||
## The Eleven Feeds
|
||||
|
||||
| Feed | What it measures | Update cadence | Latency budget |
|
||||
|---|---|---|---|
|
||||
| DShield (SANS Internet Storm Center) | Top scanning source ASNs and target ports | 1h | 5 min |
|
||||
| Cloudflare Radar — outages | Country-level connectivity drops with ASN attribution | 5 min | 30 sec |
|
||||
| Cloudflare Radar — BGP hijacks | MOAS prefix anomalies with confidence scoring | 5 min | 30 sec |
|
||||
| NVD CVE 2.0 | Newly published or modified CVEs | 2h | 30 min |
|
||||
| FIRST EPSS | Exploit prediction percentile per CVE | 2h | hours |
|
||||
| CISA KEV | Officially "exploited in the wild" catalog | 1h | hours |
|
||||
| ransomware.live | Ransomware leak-site victim posts | 15m | minutes |
|
||||
| Coinbase WS | BTC/ETH spot ticks | persistent | sub-second |
|
||||
| USGS GeoJSON | Global earthquakes M2.5+ | 1m | 30 sec |
|
||||
| NOAA SWPC | Kp index, Bz GSM, X-ray flux, solar wind speed/density | 1m / 3h | minutes |
|
||||
| Wikipedia ITN + GDELT | World news headlines and event-tone spikes | 5m / 15m | minutes |
|
||||
| ISS position + passes | Subsatellite point and TLE | 10s / 24h | seconds |
|
||||
|
||||
The cadence column is what the operator sees. The latency budget is how long the system can lag the source before the data is wrong enough to misinform a decision. Coinbase has a sub-second budget because BTC moves; KEV has hours because adding to KEV is itself a slow process. **Building a dashboard means matching every panel's freshness indicator to the right budget**, not just polling everything as fast as possible.
|
||||
|
||||
## EPSS vs CVSS, and Why It Matters
|
||||
|
||||
CVSS (Common Vulnerability Scoring System) gives a 0-10 severity number. CVSS is a *risk model* in the worst sense: a static rubric scored once at disclosure that asks "if exploited, how bad?" It does not answer "will it be exploited?"
|
||||
|
||||
EPSS (Exploit Prediction Scoring System, run by FIRST.org) answers the second question. It produces, for every CVE in NVD, a probability between 0 and 1 that the CVE will be exploited in the next 30 days, plus a percentile against the rest of the population. The model is updated daily based on observed exploitation telemetry.
|
||||
|
||||
The practical numbers:
|
||||
|
||||
- About 6% of all published CVEs ever get exploited in the wild.
|
||||
- An EPSS percentile above ~95 is a useful triage threshold for "look at this today".
|
||||
- KEV catalog membership is binary, hand-curated, and lags EPSS. KEV is the most expensive evidence (a CVE made it onto KEV because real attacks were observed) and EPSS is the cheapest forward-looking signal.
|
||||
|
||||
The CVE panel in this dashboard uses EPSS percentile as the primary sort key, with a KEV badge as a secondary visual marker. CVSS is shown but de-emphasized.
|
||||
|
||||
**Real example**: CVE-2017-5638 (Apache Struts, Equifax breach). CVSS 10.0. EPSS at disclosure was 0.97. KEV listed it eventually. By the time a CVSS-only dashboard would have moved it to "critical" attention, EPSS already said "this will be exploited within 30 days at 97% probability."
|
||||
|
||||
## CISA KEV and Why a Diff Matters
|
||||
|
||||
The CISA Known Exploited Vulnerabilities catalog is a hand-maintained list of CVEs CISA has observed being exploited against US federal systems or in widespread campaigns. Each entry includes a `dateAdded` and a `dueDate` (when federal civilian agencies must patch).
|
||||
|
||||
The catalog is small (~1000 entries) and grows by 1-10 per week. The interesting signal is not the snapshot, it is the **diff**. When a CVE is *added* to KEV today, it means CISA just confirmed in-the-wild exploitation that did not have public visibility yesterday. That diff is among the loudest single signals an enterprise defender can subscribe to.
|
||||
|
||||
The KEV collector polls the catalog, computes the new-since-last-poll set, and emits one `kev_added` event per new CVE. The frontend renders a banner and the alert engine optionally fires Telegram/Discord/Slack notifications.
|
||||
|
||||
## BGP Hijacks and Route Leaks
|
||||
|
||||
The internet routes traffic by trusting that ASNs (Autonomous Systems, the entities that run networks) honestly announce which IP prefixes they own. They do not all do this honestly.
|
||||
|
||||
A **BGP hijack** is when an ASN announces a prefix it does not own. Two flavors:
|
||||
|
||||
- **Origin hijack (MOAS, Multiple Origin AS)**: two ASNs both announce the same prefix. Cloudflare Radar's confidence scoring fuses this with prefix-validation history.
|
||||
- **Path manipulation**: AS announces a more specific prefix or a shorter path so traffic prefers it. This is how the 2018 Amazon Route 53 hijack stole MyEtherWallet credentials.
|
||||
|
||||
A **route leak** is similar but accidental: a misconfigured router announces customer routes to peers they should not have been propagated to.
|
||||
|
||||
The Radar BGP endpoint returns hijack candidates with: `started_at`, `duration_sec`, `confidence` (0-100), `hijacker_asn`, `victim_asns`, `prefixes`. The collector filters by configured `MinConfidence` (default 60) before persisting and emitting. The globe panel renders hijack regions on the great-circle path between hijacker and victim ASN geolocations.
|
||||
|
||||
**Real incidents to read up on**:
|
||||
|
||||
- April 2018, AS10297 (eNet) hijacked Route 53 prefixes for two hours, redirected MyEtherWallet, stole ~$150k of ETH.
|
||||
- June 2024, AS267613 (Eletel) hijacked Orange España prefixes, took the operator partly offline.
|
||||
- April 2010, China Telecom (AS23724) accidentally announced 50,000 prefixes worldwide for ~18 minutes ("the great Chinese internet diversion").
|
||||
|
||||
## Mass-Scan Firehose
|
||||
|
||||
DShield, run by SANS Internet Storm Center, aggregates firewall logs from thousands of voluntary submitters. They publish:
|
||||
|
||||
- Top reporting sources (which IPs scanned the most submitters in the last 24h).
|
||||
- Top targeted ports (which TCP/UDP ports got hit hardest).
|
||||
- Country-level scan distribution.
|
||||
|
||||
This is the only feed in the system that gives a global, attacker-side view of "what is being scanned right now". It is *not* an attack feed in the IDS sense; it is the noise floor of the internet. The signal is in the *delta*: when port 7547 (TR-069 / CWMP) jumps to the top of the table, you are likely watching the next Mirai-class IoT botnet ramp up. When port 8728 (MikroTik Winbox) jumps, you are watching MikroTik routers get exploited again.
|
||||
|
||||
The DShield collector merges three endpoints (top sources, top ports, country distribution) into a single snapshot blob and emits it as `scan_firehose`.
|
||||
|
||||
## Snapshot vs Stream — The Two-Phase Protocol
|
||||
|
||||
The dashboard speaks two protocols that have to coexist:
|
||||
|
||||
1. **REST snapshot** — a fresh browser asks `GET /v1/snapshot` and gets the latest state of every topic in one JSON blob. Sourced from Redis, populated continuously by the bus.
|
||||
2. **WebSocket stream** — same browser opens `/v1/ws?topics=cve_new,kev_added,...` and receives events as they happen.
|
||||
|
||||
The two have to be sequenced correctly or the browser sees impossible state. If the snapshot lags the stream by even a second, you can get a `cve_updated` event for a CVE the snapshot does not know about. The frontend handles this with a buffer:
|
||||
|
||||
- WS opens. Frontend buffers all messages.
|
||||
- Snapshot REST returns. Frontend hydrates state.
|
||||
- Frontend calls `setReady()` on the WS handle. Buffered messages flush into the same state.
|
||||
- Subsequent messages flow live.
|
||||
|
||||
This is a classic snapshot+stream pattern (used by Kafka Connect, the Linux kernel, Postgres logical replication). Getting it wrong is the most common cause of "data is missing" bugs in real-time UIs.
|
||||
|
||||
## Why an In-Process Bus Instead of Kafka or NATS
|
||||
|
||||
A common reflex is "real-time fan-out → broker". For this scale (eleven producers, dozens of WS subscribers, single box) a broker is overkill. The cost calculus:
|
||||
|
||||
| Option | Pros | Cons at this scale |
|
||||
|---|---|---|
|
||||
| Kafka | Durable, horizontally scalable, replayable | Multi-node operational cost, persistent volumes, ACL system, Zookeeper/KRaft, full week of ops setup |
|
||||
| NATS | Lighter than Kafka, simpler ops | Still a separate process, still a separate auth surface, still adds a network hop |
|
||||
| Redis Pub/Sub | Already running Redis | Lossy, no backpressure model, no persistence, blocks on slow subscriber |
|
||||
| In-process channel | Zero ops, lossy by design (drop-on-full), one-binary deploy | Cannot scale beyond one process |
|
||||
|
||||
The dashboard does not need to scale beyond one process. Eleven collectors at the cadences listed above produce well under 100 events/sec peak. A buffered Go channel with size 512 is more than enough. **The "constraint" of single-binary is actually the feature** — operations becomes `docker compose up` and a Justfile.
|
||||
|
||||
If you ever do need to scale (e.g., rendering this for a multi-tenant SOC product), you replace the bus with NATS JetStream and keep the rest. Collectors and the frontend do not change.
|
||||
|
||||
## Slow-Consumer Handling
|
||||
|
||||
A WebSocket subscriber that stops reading is a denial-of-service risk to every other subscriber. If the hub blocks on the slow client's `Write`, latency for everyone goes up. If the hub buffers unbounded, memory grows until OOM.
|
||||
|
||||
The pattern in this project: **bounded per-subscriber buffer, drop+disconnect on overflow**. The hub gives each subscriber a 256-message channel. On send, the hub does a non-blocking select. If the channel is full, the hub asynchronously closes the connection with `StatusPolicyViolation: slow consumer`. The client reconnects, gets a fresh snapshot from REST, and continues.
|
||||
|
||||
256 is not arbitrary. Heartbeat fires every 5s, Coinbase peaks ~8 ticks/sec, ISS every 10s. A buffer that only holds 16 messages fills during a single browser GC pause and triggers a reconnect loop. 256 absorbs typical jitter without papering over genuinely-stuck clients.
|
||||
|
||||
## Time-Series Storage with BRIN
|
||||
|
||||
The dashboard's tables are append-mostly: events arrive monotonically by `occurred_at`, and queries almost always filter by a recent time range. This is the workload BRIN (Block Range Index) was built for.
|
||||
|
||||
A BRIN index stores per-block-range summaries (min/max of the indexed column) instead of per-row entries. For a column that is naturally correlated with physical row order (which `occurred_at` is, because rows arrive in time order), BRIN is **dozens to hundreds of times smaller than B-tree** with comparable scan-range performance.
|
||||
|
||||
Trade-off: BRIN is great for `WHERE ts > now() - interval '24h'` and useless for `WHERE id = ?`. The migrations in this project use B-tree on primary keys and BRIN on time columns:
|
||||
|
||||
```sql
|
||||
CREATE INDEX idx_quakes_time_brin ON earthquakes USING BRIN (occurred_at);
|
||||
CREATE INDEX idx_quakes_mag ON earthquakes (mag DESC, occurred_at DESC);
|
||||
```
|
||||
|
||||
The first index serves "show me earthquakes in the last hour". The second serves "show me the strongest earthquakes from the last week".
|
||||
|
||||
You should reach for BRIN before considering native partitioning. Partitioning solves problems BRIN does not (per-partition retention via `DROP PARTITION`, parallel pruning), but adds operational complexity. For this dataset size — measured in tens of millions of rows over months, not billions — BRIN alone holds up.
|
||||
|
||||
## Auth: ES256 with Auto-Rotated Keys
|
||||
|
||||
JWT signing has two viable shapes for a public-facing dashboard:
|
||||
|
||||
- **HS256** (HMAC SHA-256, symmetric). Simple. Same secret signs and verifies. Anything with the secret can mint tokens.
|
||||
- **ES256** (ECDSA P-256, asymmetric). Private key signs. Public key verifies. You can publish the public key freely (JWKS endpoint) so other services can verify without holding the signing secret.
|
||||
|
||||
This project uses ES256 (Ed25519 in some legacy notes — the actual implementation is P-256 ECDSA per RFC 7518). The reasons:
|
||||
|
||||
- The dashboard exposes `/.well-known/jwks.json`, so a future microservice could verify dashboard tokens without trusting the dashboard's filesystem.
|
||||
- Private key rotation only requires updating the signer; verifiers fetch the new JWKS automatically.
|
||||
- Argon2id password hashes plus ES256 tokens means a database leak alone does not let an attacker mint admin sessions.
|
||||
|
||||
The bootstrap code generates the keypair if missing (see `ensureJWTKeys` in `cmd/api/main.go`). For real deployments you mount the keys from a secret store, but the auto-generation makes dev frictionless.
|
||||
|
||||
## Webhook Secret Encryption
|
||||
|
||||
Configurable alerts (Slack, Discord, Telegram) require storing webhook URLs and bot tokens. These are credentials. A compromised database without encryption gives the attacker every alerted user's webhook, which they can immediately use to spam, phish, or push malicious links into the user's communication channels.
|
||||
|
||||
The `notifications` package wraps user-supplied webhook configs with AES-256-GCM. The encryption key is a 32-byte value loaded from `NOTIFICATION_ENCRYPTION_KEY`. Encryption produces `(ciphertext, nonce)`; both are stored, neither is logged. Decryption happens on demand at alert-fire time and the plaintext never persists.
|
||||
|
||||
If the env var is not set, the entire alert engine is disabled with a logged warning. **Failing closed** is the correct security posture here: it is better to have no alerts than to have plaintext webhook credentials in your database.
|
||||
|
||||
## Testing Your Understanding
|
||||
|
||||
Before moving to architecture, you should be able to answer:
|
||||
|
||||
1. Why is EPSS percentile a better triage signal than CVSS for an operator's first 30 minutes of attention?
|
||||
2. What is the difference between an origin BGP hijack (MOAS) and a path manipulation hijack? Which one would a confidence-scored Radar feed catch more reliably?
|
||||
3. Why does the system buffer WebSocket messages on the client until the snapshot REST returns, instead of just connecting and rendering as events arrive?
|
||||
4. What is the actual scale ceiling of an in-process Go channel bus, and at what point would you switch to NATS or Kafka?
|
||||
5. Why is BRIN the right choice for `occurred_at` columns and the wrong choice for primary key columns?
|
||||
6. What goes wrong if `NOTIFICATION_ENCRYPTION_KEY` is set to a 16-byte value instead of 32?
|
||||
7. Why does the dashboard close slow WebSocket consumers instead of buffering more aggressively?
|
||||
|
||||
If any of these are unclear, re-read the relevant section. If they all make sense, continue to [02-ARCHITECTURE.md](02-ARCHITECTURE.md).
|
||||
|
|
@ -0,0 +1,343 @@
|
|||
<!--
|
||||
©AngelaMos | 2026
|
||||
02-ARCHITECTURE.md
|
||||
-->
|
||||
|
||||
# Architecture
|
||||
|
||||
This document covers the runtime shape of the system: how a single Go process orchestrates eleven heterogeneous data feeds, how events get from a collector goroutine to a browser, and how state survives a restart.
|
||||
|
||||
## High-Level Picture
|
||||
|
||||
```
|
||||
┌──────────────────────────────────────────────────────────────────────┐
|
||||
│ External feeds (HTTP / WS, varying cadences) │
|
||||
│ DShield · CF Radar · NVD · EPSS · KEV · ransomware.live · Coinbase │
|
||||
│ · USGS · NOAA SWPC · Wikipedia · GDELT · CelesTrak · wheretheiss │
|
||||
└───┬──────────────────────────────────────────────────────────────────┘
|
||||
│
|
||||
│ ▼ each collector is one errgroup goroutine
|
||||
▼
|
||||
┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐
|
||||
│ collector │ │ collector │ │ collector │ │ collector │
|
||||
│ (cve) │ │ (kev) │ │ (cfradar) │ │ … │
|
||||
└──────┬───────┘ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘
|
||||
│ │ │ │
|
||||
│ Emit(events.Event{Topic, Source, Timestamp, Payload})
|
||||
▼ ▼ ▼ ▼
|
||||
╔═══════════════════════════════════════════════════════╗
|
||||
║ bus.Bus (buffered channel, 512) ║
|
||||
║ ║
|
||||
║ 1. Persister.Save(ev) → snapshot store (Redis) ║
|
||||
║ 2. Broadcaster.Broadcast(topic, payload) → ws.Hub ║
|
||||
║ 3. fanout(ev) → channel subscribers (alerts engine) ║
|
||||
╚═══════════════════════════════════════════════════════╝
|
||||
│ │ │
|
||||
▼ ▼ ▼
|
||||
┌─────────┐ ┌──────────┐ ┌─────────────┐
|
||||
│ Redis │ │ ws.Hub │ │ alerts. │
|
||||
│ snapshot│ │ per-conn │ │ Dispatcher │
|
||||
│ store │ │ buf 256 │ │ (CEL eval) │
|
||||
└─────┬───┘ └────┬─────┘ └──────┬──────┘
|
||||
│ │ │
|
||||
▼ ▼ ▼
|
||||
GET /v1/snapshot /v1/ws Telegram /
|
||||
(REST hydrate) (WS stream) Discord / Slack
|
||||
|
||||
Append-mostly Postgres tables (BRIN indexed) live alongside Redis
|
||||
for historical queries via /v1/intel/* REST endpoints (CVE search,
|
||||
KEV list, hijack timeline, ransomware victims, earthquakes).
|
||||
```
|
||||
|
||||
The bus is the spine. Everything else is either a producer (collector), a consumer (snapshot store, ws hub, alerts dispatcher), or a query path that bypasses the bus entirely (the historical Postgres queries served by the `intel` package).
|
||||
|
||||
## The Single-Binary Decision
|
||||
|
||||
The whole backend is `cmd/api/main.go` running one binary. There is no scheduler, no separate worker, no message broker. This is intentional and load-bearing.
|
||||
|
||||
The orchestration is a single `errgroup.Group`:
|
||||
|
||||
```go
|
||||
collectorGroup, collectorCtx := errgroup.WithContext(ctx)
|
||||
collectorGroup.Go(func() error { return eventBus.Run(collectorCtx) })
|
||||
collectorGroup.Go(func() error { return beat.Run(collectorCtx) })
|
||||
|
||||
if cfg.Collectors.DShield.Enabled {
|
||||
coll := dshield.NewCollector(dshield.CollectorConfig{...})
|
||||
collectorGroup.Go(func() error { return coll.Run(collectorCtx) })
|
||||
}
|
||||
// ... ten more collectors ...
|
||||
```
|
||||
|
||||
`errgroup.WithContext` gives you two guarantees:
|
||||
|
||||
1. **Shared cancellation**. If any goroutine returns an error, every other goroutine's context is cancelled. One collector exploding does not silently leave the others running on a half-broken process.
|
||||
2. **Wait-for-all teardown**. `collectorGroup.Wait()` blocks until every goroutine has actually returned. Combined with a `signal.NotifyContext(SIGINT, SIGTERM)` parent context and a 5-second drain delay before the HTTP server stops, every event in flight has a chance to be persisted and broadcast before the process exits.
|
||||
|
||||
The cost of this design: you cannot horizontally scale a single user's view across multiple processes (there is no shared bus). The benefit: zero distributed-systems failure modes for ~99% of deployments. **If you ever want multi-process, you replace `bus.Bus` with NATS JetStream and keep everything else.**
|
||||
|
||||
## The Event Bus
|
||||
|
||||
`bus.Bus` is a buffered Go channel with three downstream effects per event. The shape:
|
||||
|
||||
```go
|
||||
type Bus struct {
|
||||
ch chan events.Event // size 512
|
||||
persister Persister // redis snapshot store
|
||||
broadcaster Broadcaster // ws hub
|
||||
subscribers []chan events.Event // alerts dispatcher etc
|
||||
}
|
||||
|
||||
func (b *Bus) Emit(ev events.Event) {
|
||||
select {
|
||||
case b.ch <- ev:
|
||||
default:
|
||||
b.dropped.Add(1)
|
||||
b.logger.Warn("event bus full, dropped", ...)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
Three things are non-obvious here:
|
||||
|
||||
- **`Emit` is non-blocking**. A backed-up bus drops events instead of stalling collectors. This is the right choice when collectors are time-sensitive (Coinbase, USGS) and would rather miss a tick than fall behind real time.
|
||||
- **One goroutine drains the channel**. The `Run` loop pulls events off the buffer, persists, broadcasts, and fans out, in that order. If persistence is slow, broadcast lags. This is acceptable because the snapshot is the source of truth for late-joining clients.
|
||||
- **Subscribers are separate from broadcast**. The alerts engine subscribes through `Bus.Subscribe()` which returns a 256-buffered channel. Slow alert evaluation does not back up the bus; events are dropped at the subscriber boundary instead.
|
||||
|
||||
The bus is the only place the system tolerates lossiness. Below the bus (Postgres writes inside collector repositories) is durable. Above the bus (snapshot store) is eventually consistent with last-write-wins semantics — a dropped event simply means the next event for that topic overwrites the lost one a few seconds later.
|
||||
|
||||
## The Collector Pattern
|
||||
|
||||
Every collector implements roughly:
|
||||
|
||||
```go
|
||||
type CollectorConfig struct {
|
||||
Interval time.Duration
|
||||
Fetcher Fetcher // calls the upstream API
|
||||
Repo Repository // persists historical rows
|
||||
Emitter Emitter // bus.Emit
|
||||
State StateRecorder // health/freshness tracking
|
||||
Logger *slog.Logger
|
||||
}
|
||||
|
||||
func (c *Collector) Run(ctx context.Context) error {
|
||||
ticker := time.NewTicker(c.cfg.Interval)
|
||||
defer ticker.Stop()
|
||||
c.tick(ctx) // immediate first poll
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-ticker.C:
|
||||
c.tick(ctx)
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
The `tick` body is upstream-specific but the structure is the same: fetch, persist to its own table, emit a single event. The fetcher, repository, emitter, and state recorder are interfaces — every collector ships with a fake of each so the unit tests do not need network or database.
|
||||
|
||||
Some collectors are weirder than others:
|
||||
|
||||
- **DShield** merges three endpoints into one snapshot. Optional `Enricher` enriches the top-IPs blob with country code from AbuseIPDB and classification from GreyNoise.
|
||||
- **Coinbase** is a persistent WebSocket, not a poller. The `coinbase` package decomposes into `client` (raw WS), `readloop` (frame-by-frame read with sequencer), `aggregator` (tick → minute bar), `reconnect` (exponential backoff), and `collector` (orchestration).
|
||||
- **ISS** has two cadences: `PositionInterval` (10s) for live position via `wheretheiss.at`, and `TLEInterval` (24h) to refresh the orbital element set from CelesTrak. The cached TLE is propagated client-side between polls so the globe is not blocked on the network.
|
||||
- **SWPC** has both a fast (1m) and slow (3h) cadence. Kp index updates every 3h; magnetometer Bz updates every minute. One collector handles both.
|
||||
|
||||
Every collector reports via `state.RecordSuccess` or `state.RecordError` on each tick. The `collector_state` table is the source of truth for the `/v1/admin/health` endpoint, which lights up red panels when a feed has not refreshed within its expected window.
|
||||
|
||||
## The Snapshot Store
|
||||
|
||||
A new browser must show useful state immediately, before any WebSocket events arrive. The snapshot store solves that.
|
||||
|
||||
`snapshot.Store` is a thin Redis wrapper. The persister side, attached to the bus, calls `Save` for every event:
|
||||
|
||||
```go
|
||||
func (s StorePersister) Save(ctx context.Context, ev events.Event) error {
|
||||
raw, _ := json.Marshal(ev.Payload)
|
||||
if ev.Topic == events.TopicCoinbasePrice {
|
||||
return s.Store.MergeSymbolMap(ctx, ev.Topic, raw)
|
||||
}
|
||||
return s.Store.PutLatest(ctx, ev.Topic, raw)
|
||||
}
|
||||
```
|
||||
|
||||
For most topics, "latest" is one Redis key (`state:cve_new`, `state:kev_added`, etc.) with the raw JSON payload. For Coinbase, where there are multiple symbols, the topic is a Redis hash with one field per symbol so a BTC tick does not stomp on the latest ETH tick.
|
||||
|
||||
The serve side, behind `GET /v1/snapshot`, scans `state:*` and returns one merged JSON document. The frontend hydrates everything in one round trip, then catches up via WebSocket.
|
||||
|
||||
The store does **not** persist historical data. Postgres does. The store is a cold-start cache, not the system of record. If Redis is wiped, the dashboard reboots cold for ~2 hours (until the slowest collector cycles) but loses no durable data.
|
||||
|
||||
## The WebSocket Hub
|
||||
|
||||
`ws.Hub` is a fan-out router with bounded per-subscriber buffers and aggressive slow-consumer eviction.
|
||||
|
||||
```go
|
||||
const (
|
||||
defaultSubscriberBuf = 256
|
||||
defaultPingInterval = 30 * time.Second
|
||||
defaultMaxSubs = 5000
|
||||
)
|
||||
|
||||
func (h *Hub) Broadcast(topic events.Topic, payload []byte) {
|
||||
env, _ := EncodeEnvelope(string(topic), payload)
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
for sub := range h.subs {
|
||||
if _, ok := sub.topics[topic]; !ok {
|
||||
continue
|
||||
}
|
||||
select {
|
||||
case sub.msgs <- env:
|
||||
default:
|
||||
go sub.closeSlow()
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
Things to notice:
|
||||
|
||||
- **Per-subscriber topic filter**. The connecting client sends `?topics=cve_new,kev_added,...` and the hub only sends matching topics. A panel that does not need Coinbase ticks does not pay for them.
|
||||
- **Non-blocking send**. If the subscriber's 256-message channel is full, the hub does not wait. It schedules a `closeSlow()` goroutine that closes the connection with `StatusPolicyViolation`.
|
||||
- **Heartbeat pings every 30s**. The browser's reconnect logic uses missing pongs as the disconnect signal; this is faster and more reliable than waiting for TCP RST.
|
||||
- **Capacity limit**. `MaxSubscribers: 5000` (configurable) protects against connection-flood DoS. Over capacity, new connections get `StatusTryAgainLater`.
|
||||
|
||||
The envelope format is intentionally minimal:
|
||||
|
||||
```json
|
||||
{"ch":"cve_new","data":{...},"ts":"2026-05-08T18:30:12.123456789Z"}
|
||||
```
|
||||
|
||||
`ch` is the topic, `data` is the raw payload, `ts` is the server-side timestamp at fan-out. No protocol versioning, no auth in-band — auth is on the upgrade handshake.
|
||||
|
||||
## The Snapshot+Stream Handshake
|
||||
|
||||
The frontend's `createDashboardWS` factory implements the buffer-then-flush pattern:
|
||||
|
||||
```ts
|
||||
return {
|
||||
connect() { attach() },
|
||||
setReady() {
|
||||
ready = true
|
||||
sendInitIfReady()
|
||||
const flush = buffer
|
||||
buffer = []
|
||||
for (const ev of flush) onEvent(ev)
|
||||
},
|
||||
...
|
||||
}
|
||||
```
|
||||
|
||||
The lifecycle:
|
||||
|
||||
1. App boots. `connect()` opens the WebSocket. `ready = false`. Incoming messages go into `buffer`.
|
||||
2. App fetches `GET /v1/snapshot` via TanStack Query. Response hydrates Zustand stores.
|
||||
3. App calls `setReady()`. The buffered messages are replayed through `onEvent`. `ready = true`. Subsequent messages flow live.
|
||||
|
||||
The protocol's only client→server message is `{"op":"init"}`, sent by `setReady()`. The server-side hub reads and discards client messages — the protocol is server-push only — but the previous code closed the connection with `StatusPolicyViolation` on receiving any data frame, which turned every client `init` into a forced reconnect. The current `Serve` loop reads-and-discards in a separate goroutine to keep the connection alive.
|
||||
|
||||
This handshake is the single most error-prone part of the system. The unit tests in `frontend/src/api/ws.test.ts` and the backend's `ws/hub_test.go` exist specifically because we got it wrong twice during development.
|
||||
|
||||
## Storage: Redis + Postgres
|
||||
|
||||
The split is deliberate.
|
||||
|
||||
**Redis** holds:
|
||||
|
||||
- The snapshot store (`state:*`) — one key per topic, last-write-wins.
|
||||
- TLE for ISS — small JSON, refreshed daily, read continuously.
|
||||
- Rate limit counters — sliding window per IP/user.
|
||||
- Alert engine cooldowns — per (rule, channel) `SET NX EX` keys.
|
||||
- Auth session tokens (refresh token rotation).
|
||||
|
||||
**Postgres** holds everything that needs to be queryable historically: CVEs, KEV entries, earthquakes, ransomware victims, BTC/ETH ticks and minute bars, outage events, BGP hijacks, DShield snapshots, world events, alert rule definitions, alert fire history, user accounts, notification channel configs.
|
||||
|
||||
Redis is the hot path. Postgres is the cold path. The two never depend on each other for correctness; if Redis is down, the dashboard has no live state but the `intel` REST endpoints (`/v1/intel/cve/search`, `/v1/intel/kev/list`, etc.) still work against Postgres directly.
|
||||
|
||||
The Postgres schema uses BRIN on every time column and B-tree on primary keys plus a few selective indexes (severity+lastmod for CVE, mag+occurred for earthquakes, confidence+detected for hijacks). All payload columns are `jsonb` because schemas evolve faster than migrations.
|
||||
|
||||
## The Alert Engine
|
||||
|
||||
The alert engine is an in-process consumer subscribed to the bus. Per event, it looks up rules indexed by topic, evaluates each rule's CEL predicate against the payload, and if matching, dispatches to the user's configured channels.
|
||||
|
||||
Indexed by topic so the hot path is `O(rules-for-this-topic)` not `O(all-rules)`:
|
||||
|
||||
```go
|
||||
e.rulesByTopic.Store(&indexed) // atomic.Pointer[map[string][]compiledRule]
|
||||
```
|
||||
|
||||
The pointer swap means rule reloads (every 30s) are lock-free for the eval path. Refresh just compiles every rule's CEL predicate into a `cel.Program` and atomically swaps the index.
|
||||
|
||||
Cooldowns are per `(rule, channel)`:
|
||||
|
||||
```go
|
||||
key := fmt.Sprintf("alert_cooldown:%s:%s:%s", rule.ID, ch.Type, ch.ID)
|
||||
ok, err := e.cooldowns.TryAcquire(ctx, key, cooldown)
|
||||
```
|
||||
|
||||
`TryAcquire` is a Redis `SET NX EX`. A user with rules on Slack and Telegram for the same predicate has independent cooldown windows on each — important because Slack and Telegram have different rate limits and a single global cooldown would cause one to mute the other.
|
||||
|
||||
CEL was chosen over a custom DSL or Lua. Reasons: it is sandboxed by design (no syscalls, no I/O), has a tiny Go binary footprint, and predicate authoring is familiar to anyone who has used IAM conditions, Cloud Armor rules, or Envoy filters.
|
||||
|
||||
## Auth Pipeline
|
||||
|
||||
The auth flow is conventional ES256 JWT with auto-rotated keys:
|
||||
|
||||
```
|
||||
1. POST /v1/auth/register
|
||||
→ Argon2id(password) → users.password_hash
|
||||
→ seed default alert rules
|
||||
|
||||
2. POST /v1/auth/login
|
||||
→ verify Argon2id
|
||||
→ mint access_token (15m, ES256, kid in header)
|
||||
→ mint refresh_token (30d, opaque, stored in Redis)
|
||||
→ set HttpOnly Secure cookies in production
|
||||
|
||||
3. Every authenticated request
|
||||
→ middleware.Authenticator(verifier)
|
||||
→ verifier checks ES256 signature + revocation list
|
||||
→ ctx is populated with user identity
|
||||
|
||||
4. Public endpoints get a guest path
|
||||
→ /v1/snapshot, /v1/ws, /v1/intel/* are read-only and unauthenticated
|
||||
→ /v1/alerts/* and /v1/notifications/* require auth
|
||||
|
||||
5. Admin endpoints
|
||||
→ middleware.RequireAdmin checks email vs cfg.App.AdminEmail
|
||||
→ /v1/admin/db, /v1/admin/redis, /v1/admin/users
|
||||
```
|
||||
|
||||
The signing key lives at `cfg.JWT.PrivateKeyPath`. If missing on boot, `ensureJWTKeys` generates a fresh ES256 keypair into `keys/`. The public side is exposed at `/.well-known/jwks.json` so any external service can verify dashboard tokens without trusting the dashboard's filesystem.
|
||||
|
||||
Refresh tokens are rotated on every use (Redis swap-and-set), so a leaked refresh token has at most one use before invalidation.
|
||||
|
||||
## Frontend State Architecture
|
||||
|
||||
Roughly the inverse of the backend:
|
||||
|
||||
- **TanStack Query v5** — REST. Owns the snapshot fetch, the `/v1/intel/*` historical queries, and any user-action mutations (create rule, delete channel).
|
||||
- **Zustand stores** — live state. One store per topic family: `prices`, `ticker`, `cve`, `kev`, `bgpHijack`, `outage`, `earthquake`, `ransomware`, `freshness`, `globeEvents`, `heartbeat`, `audio`, `ui`.
|
||||
- **Component selectors** — components subscribe to specific store slices via Zustand's selector pattern, so a CVE update does not re-render the BTC panel.
|
||||
|
||||
The WebSocket message handler is a single dispatch table that routes incoming `{ch, data}` envelopes into the right store. Adding a new feed on the backend requires adding one entry to this table and one new Zustand store.
|
||||
|
||||
The globe (`pages/globe/Globe.tsx`) is the heaviest component. It uses MapLibre for the projection and renders five distinct layer types: country choropleth (outages), great-circle paths (BGP hijacks), point clouds (DShield ASN dots, earthquake epicenters, ransomware markers), and an animated trail (ISS ground track). Layer updates are batched per animation frame because individually mutating MapLibre layers per WebSocket message is the fastest way to drop frame rate.
|
||||
|
||||
## Failure Modes the Architecture Survives
|
||||
|
||||
| Failure | Effect | Recovery |
|
||||
|---|---|---|
|
||||
| Single feed goes down (e.g., USGS 502) | That collector logs error, `RecordError`, dashboard shows stale-yellow on that panel. Other panels unaffected. | Next tick succeeds, freshness recovers. |
|
||||
| Redis goes down | Snapshot returns empty, WS still works, intel endpoints still work. | Bus persister logs errors, drops nothing else. |
|
||||
| Postgres goes down | Collectors that need to upsert log errors. Bus emit + WS still work. Intel REST returns 503. | Postgres reconnects, collectors resume on next tick. |
|
||||
| Bus channel full | `Emit` drops the event, increments counter. Snapshot misses one update. | Next event for that topic overwrites in snapshot. |
|
||||
| WS subscriber stalls | Hub closes them with `StatusPolicyViolation`. | Browser reconnects, gets fresh snapshot, continues. |
|
||||
| Process killed mid-write | `errgroup.Wait()` blocks shutdown until in-flight work finishes (within `ShutdownTimeout`). | Restart picks up where it left off, no data loss for durable tables. |
|
||||
|
||||
The system is **not** designed to survive losing the host. There is no replication, no leader election, no quorum. It is a single-binary single-host dashboard. If you need HA, you put a second instance behind a load balancer with its own Postgres replica and accept that the two instances will see slightly different live states (because Coinbase WS ticks arrive at different moments). For the original use case — one operator, one screen, one box — that complexity is the wrong trade.
|
||||
|
||||
## Where To Go Next
|
||||
|
||||
Continue to [03-IMPLEMENTATION.md](03-IMPLEMENTATION.md) for the actual code walkthrough across collectors, bus, ws, alerts, snapshot, and frontend wiring.
|
||||
|
|
@ -0,0 +1,714 @@
|
|||
<!--
|
||||
©AngelaMos | 2026
|
||||
03-IMPLEMENTATION.md
|
||||
-->
|
||||
|
||||
# Implementation
|
||||
|
||||
This is the code-level walkthrough. It is organized by concern, not by file. We follow the path of one event from external API to a pixel on the globe, then loop back to cover the cross-cutting code.
|
||||
|
||||
All snippets are excerpts from the actual code. Snippets are intentionally not annotated with line numbers — formatting and small refactors invalidate line numbers in days, and the package + function name is enough to find any block again with `grep`.
|
||||
|
||||
## The Event Type
|
||||
|
||||
Everything in the system funnels through `events.Event`. It is deliberately tiny.
|
||||
|
||||
```go
|
||||
// internal/events/event.go
|
||||
package events
|
||||
|
||||
import "time"
|
||||
|
||||
type Event struct {
|
||||
Topic Topic `json:"topic"`
|
||||
Timestamp time.Time `json:"timestamp"`
|
||||
Source string `json:"source"`
|
||||
Payload any `json:"payload"`
|
||||
}
|
||||
```
|
||||
|
||||
`Payload any` is the interesting choice. We could have made it a sealed union of typed payloads, one per topic. We chose not to. The reasons:
|
||||
|
||||
- **Heterogeneous payload shapes.** A KEV add is a single CVE ID with a date. A DShield snapshot is a fused blob from three endpoints. A Coinbase tick is a price plus volume. Forcing them into one Go type tree adds zero runtime safety because the consumers (snapshot store, ws hub, alerts engine) all immediately re-marshal to JSON.
|
||||
- **The bus is the only typed boundary on the producer side.** Every collector packs its own payload as `json.RawMessage` before emitting, so the JSON shape is committed at the collector. Subscribers pay one `json.Unmarshal` if they need fields.
|
||||
|
||||
The trade is real: a typo in a payload field name in a collector will not be caught at compile time. We mitigate with golden-file tests under each collector's `testdata/` directory.
|
||||
|
||||
## The Topic Enum
|
||||
|
||||
```go
|
||||
// internal/events/topic.go
|
||||
type Topic string
|
||||
|
||||
const (
|
||||
TopicHeartbeat Topic = "heartbeat"
|
||||
TopicScanFirehose Topic = "scan_firehose"
|
||||
TopicInternetOutage Topic = "internet_outage"
|
||||
TopicBGPHijack Topic = "bgp_hijack"
|
||||
TopicCVENew Topic = "cve_new"
|
||||
TopicCVEUpdated Topic = "cve_updated"
|
||||
TopicEPSS Topic = "epss"
|
||||
TopicKEVAdded Topic = "kev_added"
|
||||
TopicRansomwareVictim Topic = "ransomware_victim"
|
||||
TopicCoinbasePrice Topic = "coinbase_price"
|
||||
TopicEarthquake Topic = "earthquake"
|
||||
TopicSpaceWeather Topic = "space_weather"
|
||||
TopicWikipediaITN Topic = "wiki_itn"
|
||||
TopicGDELTSpike Topic = "gdelt_spike"
|
||||
TopicISSPosition Topic = "iss_position"
|
||||
TopicCollectorState Topic = "collector_state"
|
||||
)
|
||||
|
||||
func (t Topic) IsValid() bool { ... }
|
||||
func AllTopics() []Topic { ... }
|
||||
```
|
||||
|
||||
`IsValid` is used by the WebSocket handler to whitelist `?topics=` query parameters. `AllTopics` is the default fallback when a client connects without specifying topics. Adding a feed means adding one constant here, one case in `IsValid`, and one entry in `AllTopics`. Then a collector emits with the new topic and the WS layer picks it up automatically.
|
||||
|
||||
## The Bus
|
||||
|
||||
`internal/bus/bus.go` is the spine.
|
||||
|
||||
```go
|
||||
type Bus struct {
|
||||
ch chan events.Event
|
||||
persister Persister
|
||||
broadcaster Broadcaster
|
||||
logger *slog.Logger
|
||||
dropped atomic.Uint64
|
||||
|
||||
subsMu sync.RWMutex
|
||||
subscribers []chan events.Event
|
||||
subDropped atomic.Uint64
|
||||
}
|
||||
|
||||
func (b *Bus) Emit(ev events.Event) {
|
||||
select {
|
||||
case b.ch <- ev:
|
||||
default:
|
||||
b.dropped.Add(1)
|
||||
b.logger.Warn("event bus full, dropped", "topic", ev.Topic, "source", ev.Source)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
`Emit` is the producer-side API. The `select` with a `default` makes it non-blocking — if the buffer is full, we drop and increment the counter rather than block the calling collector goroutine.
|
||||
|
||||
The drain side, in `Run`, is the consumer:
|
||||
|
||||
```go
|
||||
func (b *Bus) Run(ctx context.Context) error {
|
||||
defer b.closeSubscribers()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case ev := <-b.ch:
|
||||
if b.persister != nil {
|
||||
if err := b.persister.Save(ctx, ev); err != nil {
|
||||
b.logger.Error("persist event failed", "err", err, "topic", ev.Topic)
|
||||
}
|
||||
}
|
||||
if b.broadcaster != nil {
|
||||
payload, err := json.Marshal(ev.Payload)
|
||||
if err != nil {
|
||||
b.logger.Error("marshal payload failed", "err", err, "topic", ev.Topic)
|
||||
continue
|
||||
}
|
||||
b.broadcaster.Broadcast(string(ev.Topic), payload)
|
||||
}
|
||||
b.fanout(ev)
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
The order matters. Persist first so the snapshot is correct for any browser that connects between this event and the next. Broadcast second so live consumers see the event right after persistence. Fan out to channel subscribers (alerts engine) last because they may evaluate slow CEL predicates.
|
||||
|
||||
`fanout` is the second drop site:
|
||||
|
||||
```go
|
||||
func (b *Bus) fanout(ev events.Event) {
|
||||
b.subsMu.RLock()
|
||||
subs := b.subscribers
|
||||
b.subsMu.RUnlock()
|
||||
for _, ch := range subs {
|
||||
select {
|
||||
case ch <- ev:
|
||||
default:
|
||||
b.subDropped.Add(1)
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
A slow alerts engine cannot back up the bus. It just loses alert evaluations, which is logged and visible via `SubscriberDroppedCount()` for ops monitoring.
|
||||
|
||||
## A Walking Tour of One Collector
|
||||
|
||||
Every collector has the same skeleton: poll, fetch, persist row(s), emit event(s), record state. We will use CVE because it is the most representative — pollable HTTP, dependent on a second API (EPSS), with both upsert and per-row event emission.
|
||||
|
||||
### CVE Collector — The Tick
|
||||
|
||||
```go
|
||||
// internal/collectors/cve/collector.go
|
||||
func (c *Collector) tick(ctx context.Context) {
|
||||
end := time.Now().UTC()
|
||||
start := end.Add(-c.cfg.Window)
|
||||
|
||||
resp, err := c.cfg.NVD.Fetch(ctx, start, end)
|
||||
if err != nil {
|
||||
c.logger.Warn("nvd fetch", "err", err)
|
||||
c.cfg.State.RecordError(ctx, Name, err.Error())
|
||||
return
|
||||
}
|
||||
```
|
||||
|
||||
Note the structure: a window of `[end - Window, end]`, not "everything". CVE feeds are append-mostly but can re-publish modifications, so a 2-hour window with a 2-hour poll interval gives a 1-window overlap that catches late modifications without re-processing the whole history.
|
||||
|
||||
```go
|
||||
ids := make([]string, 0, len(resp.Vulnerabilities))
|
||||
rows := make([]Row, 0, len(resp.Vulnerabilities))
|
||||
for _, v := range resp.Vulnerabilities {
|
||||
score, severity := v.PrimarySeverity()
|
||||
raw, _ := json.Marshal(v)
|
||||
rows = append(rows, Row{
|
||||
CveID: v.CVE.ID,
|
||||
Published: v.CVE.Published.Time,
|
||||
LastModified: v.CVE.LastModified.Time,
|
||||
Severity: severity,
|
||||
CVSS: score,
|
||||
Payload: raw,
|
||||
})
|
||||
ids = append(ids, v.CVE.ID)
|
||||
}
|
||||
|
||||
scores, err := c.cfg.EPSS.LookupBatch(ctx, ids)
|
||||
```
|
||||
|
||||
Two API calls per tick: NVD for the CVE bodies, EPSS for the scores. We do them sequentially because EPSS is keyed on the CVE IDs the NVD call just returned. The EPSS error is logged but does not abort the tick — a CVE without an EPSS score is still useful, just not as well prioritized.
|
||||
|
||||
```go
|
||||
for _, row := range rows {
|
||||
if err := c.cfg.Repo.Upsert(ctx, row); err != nil { ... }
|
||||
if s, ok := scores[row.CveID]; ok {
|
||||
_ = c.cfg.Repo.UpdateEPSS(ctx, row.CveID, s.Score, s.Percentile)
|
||||
row.EPSSScore = &s.Score
|
||||
row.EPSSPercentile = &s.Percentile
|
||||
}
|
||||
body, _ := json.Marshal(row)
|
||||
c.cfg.Emitter.Emit(events.Event{
|
||||
Topic: events.TopicCVENew,
|
||||
Timestamp: end,
|
||||
Source: Name,
|
||||
Payload: json.RawMessage(body),
|
||||
})
|
||||
}
|
||||
c.cfg.State.RecordSuccess(ctx, Name, emitted)
|
||||
}
|
||||
```
|
||||
|
||||
Persist-then-emit, per row. We emit `TopicCVENew` for every row in the window, even ones that were already in the table. The reason: a CVE's EPSS score can change without the CVE itself changing, and the frontend should reflect that. We could split into `cve_new` vs `cve_updated` based on insert vs update, but the consumers do not need that distinction yet.
|
||||
|
||||
`RecordSuccess` writes one row to `collector_state` so `/v1/admin/health` and the freshness ribbon on the frontend can both see "CVE collector ran 12 minutes ago, processed 47 rows".
|
||||
|
||||
### DShield — Multi-Endpoint Fusion
|
||||
|
||||
DShield is the cleanest example of the "fuse multiple upstream endpoints into one event" pattern.
|
||||
|
||||
```go
|
||||
// internal/collectors/dshield/collector.go
|
||||
snaps, err := c.cfg.Fetcher.FetchAll(ctx) // hits 3 endpoints concurrently
|
||||
|
||||
merged := map[string]json.RawMessage{"ts": tsRaw}
|
||||
for _, s := range snaps {
|
||||
payload := s.Payload
|
||||
if s.Kind == KindTopIPs {
|
||||
payload = c.enrichSources(ctx, payload)
|
||||
}
|
||||
_ = c.cfg.Persister.PutSnapshot(ctx, now, s.Kind, payload)
|
||||
merged[s.Kind] = payload
|
||||
}
|
||||
|
||||
body, _ := json.Marshal(merged)
|
||||
c.cfg.Emitter.Emit(events.Event{
|
||||
Topic: events.TopicScanFirehose,
|
||||
Timestamp: now,
|
||||
Source: Name,
|
||||
Payload: json.RawMessage(body),
|
||||
})
|
||||
```
|
||||
|
||||
The merged JSON has shape `{ts, top_ips, top_ports, country_dist}`. The frontend's `DShieldPanel` consumes the merged blob in one render pass. The Postgres persistence is per-kind so we can query "show me top IPs from 2026-04-01" without parsing a fused blob.
|
||||
|
||||
The `enrichSources` step is optional and runs only if an `Enricher` is configured (AbuseIPDB + GreyNoise). It iterates the top-IPs list, looks up each IP, and returns enriched rows with country code, classification, and threat actor name. If the lookup fails or the enricher is nil, it falls through to the original payload.
|
||||
|
||||
### Coinbase — Persistent WebSocket
|
||||
|
||||
Coinbase is the only collector that does not poll. The decomposition into multiple files reflects how messy real WebSocket clients get:
|
||||
|
||||
- `client.go` — raw connection, `ReadFrame`, `WriteFrame`.
|
||||
- `readloop.go` — frame loop with sequencer reset on snapshot frames.
|
||||
- `aggregator.go` — tick-to-minute-bar OHLC.
|
||||
- `reconnect.go` — exponential backoff on disconnect.
|
||||
- `sequencer.go` — gap detection on `sequence_num`.
|
||||
- `collector.go` — orchestration.
|
||||
|
||||
The aggregator is the most interesting:
|
||||
|
||||
```go
|
||||
// internal/collectors/coinbase/aggregator.go
|
||||
func (a *Aggregator) Push(t Tick) (*MinuteBar, MinuteBar) {
|
||||
minute := t.TS.UTC().Truncate(time.Minute)
|
||||
cur, exists := a.open[t.Symbol]
|
||||
|
||||
if !exists {
|
||||
cur = MinuteBar{Symbol: t.Symbol, Minute: minute, Open: t.Price, High: t.Price, Low: t.Price, Close: t.Price, Volume24hAtClose: t.Volume24h}
|
||||
a.open[t.Symbol] = cur
|
||||
return nil, cur
|
||||
}
|
||||
|
||||
if minute.After(cur.Minute) {
|
||||
closed := cur
|
||||
cur = MinuteBar{...} // start new minute
|
||||
a.open[t.Symbol] = cur
|
||||
return &closed, cur
|
||||
}
|
||||
|
||||
// same minute, update OHLC
|
||||
if t.Price.GreaterThan(cur.High) { cur.High = t.Price }
|
||||
if t.Price.LessThan(cur.Low) { cur.Low = t.Price }
|
||||
cur.Close = t.Price
|
||||
cur.Volume24hAtClose = t.Volume24h
|
||||
a.open[t.Symbol] = cur
|
||||
return nil, cur
|
||||
}
|
||||
```
|
||||
|
||||
Returns `(closed, current)`. When a tick crosses a minute boundary, the previous minute is returned as `closed` and the collector persists it to `btc_eth_minute`. The current minute is always returned for live display.
|
||||
|
||||
This is also why the minute volume column was renamed in migration `0005`: the original column was called `volume` and we documented "per-minute volume", but the Coinbase ticker channel only gives 24-hour rolling volume, not per-trade size. The rename to `volume_24h_at_close` makes the actual semantic visible in the schema.
|
||||
|
||||
### ISS — Two Cadences, Client-Side Propagation
|
||||
|
||||
```go
|
||||
// cmd/api/main.go
|
||||
coll := iss.NewCollector(iss.CollectorConfig{
|
||||
PositionInterval: cfg.Collectors.ISS.PositionInterval, // 10s
|
||||
TLEInterval: cfg.Collectors.ISS.TLEInterval, // 24h
|
||||
Fetcher: iss.NewClient(iss.ClientConfig{}),
|
||||
TLEStore: iss.NewTLEStore(redis.Client),
|
||||
...
|
||||
})
|
||||
```
|
||||
|
||||
The ISS collector polls `wheretheiss.at` every 10s for the live position (this is the source the dashboard's "current ISS pin" uses). It also polls CelesTrak every 24 hours for the TLE (Two-Line Element set), caches it in Redis, and emits it.
|
||||
|
||||
The frontend has `frontend/src/lib/sgp4.ts` (or equivalent) that propagates the cached TLE on each animation frame to draw the orbital track. So the visible orbit is computed client-side from a 24-hour-old TLE; only the "current pin" is server-pushed every 10s.
|
||||
|
||||
This is a good example of **moving compute to the client** when the algorithm is well-defined and the data input is small. SGP4 is ~200 lines of math; pushing the propagated position over WebSocket every animation frame would saturate the network for no benefit.
|
||||
|
||||
## The WebSocket Hub
|
||||
|
||||
`internal/ws/hub.go`:
|
||||
|
||||
```go
|
||||
const (
|
||||
defaultSubscriberBuf = 256
|
||||
defaultPingInterval = 30 * time.Second
|
||||
defaultPingTimeout = 10 * time.Second
|
||||
defaultWriteTimeout = 5 * time.Second
|
||||
defaultMaxSubs = 5000
|
||||
)
|
||||
|
||||
func (h *Hub) Broadcast(topic events.Topic, payload []byte) {
|
||||
env, err := EncodeEnvelope(string(topic), payload)
|
||||
if err != nil {
|
||||
h.logger.Error("encode envelope", "err", err, "topic", topic)
|
||||
return
|
||||
}
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
for sub := range h.subs {
|
||||
if _, ok := sub.topics[topic]; !ok {
|
||||
continue
|
||||
}
|
||||
select {
|
||||
case sub.msgs <- env:
|
||||
default:
|
||||
go sub.closeSlow()
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
The lock spans the whole iteration. This is fine because the iteration is fast (just a non-blocking channel send per sub) and the lock serializes against `add`/`remove`, which are the only other writers to `h.subs`. The connection close happens in a separate goroutine to avoid holding the lock through any I/O.
|
||||
|
||||
The slow-consumer comment in the constant block is worth quoting in full because it is the kind of thing you only learn by getting it wrong:
|
||||
|
||||
> Per-connection outbound buffer. Must absorb peak event rate × browser stutter window: heartbeat (5s) + coinbase (~8/s peak) + ISS (10s) + occasional bursts of CVE/KEV/ransomware. 16 is too small — a single browser GC pause fills it and triggers slow-consumer close, which the user sees as "NO SIGNAL".
|
||||
|
||||
### Serve Loop
|
||||
|
||||
```go
|
||||
// internal/ws/hub.go
|
||||
func (h *Hub) Serve(ctx context.Context, c *cdrws.Conn, topics []events.Topic) error {
|
||||
sub := newSubscriber(topics, h.bufSize, func() {
|
||||
_ = c.Close(cdrws.StatusPolicyViolation, "slow consumer")
|
||||
})
|
||||
if !h.add(sub) {
|
||||
_ = c.Close(cdrws.StatusTryAgainLater, "server at capacity")
|
||||
return ErrAtCapacity
|
||||
}
|
||||
defer h.remove(sub)
|
||||
|
||||
connCtx, cancelRead := context.WithCancel(ctx)
|
||||
defer cancelRead()
|
||||
go func() {
|
||||
for {
|
||||
if _, _, err := c.Read(connCtx); err != nil {
|
||||
cancelRead()
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
pingT := time.NewTicker(h.pingInterval)
|
||||
defer pingT.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case msg := <-sub.msgs:
|
||||
wctx, cancel := context.WithTimeout(connCtx, h.writeTimeout)
|
||||
err := c.Write(wctx, cdrws.MessageText, msg)
|
||||
cancel()
|
||||
if err != nil { return err }
|
||||
case <-pingT.C:
|
||||
pctx, cancel := context.WithTimeout(connCtx, h.pingTimeout)
|
||||
err := c.Ping(pctx)
|
||||
cancel()
|
||||
if err != nil { return err }
|
||||
case <-connCtx.Done():
|
||||
_ = c.Close(cdrws.StatusNormalClosure, "")
|
||||
return connCtx.Err()
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
The dedicated read goroutine is the post-mortem of a real bug. The `coder/websocket` library has a helper `c.CloseRead(ctx)` that auto-closes the connection on any received frame. We used it. The frontend's `setReady()` sends `{"op":"init"}` after snapshot hydration. That message tripped `CloseRead`'s "received unexpected data" path, killing the connection with `StatusPolicyViolation`. The user saw a reconnect loop.
|
||||
|
||||
The fix: drain client→server messages explicitly with `c.Read(connCtx)` and discard them. Connection stays alive; ping/pong frames still get handled by the library.
|
||||
|
||||
### Envelope
|
||||
|
||||
```go
|
||||
// internal/ws/envelope.go
|
||||
func EncodeEnvelope(channel string, payload []byte) ([]byte, error) {
|
||||
var buf bytes.Buffer
|
||||
buf.WriteString(`{"ch":`)
|
||||
chRaw, _ := json.Marshal(channel)
|
||||
buf.Write(chRaw)
|
||||
buf.WriteString(`,"data":`)
|
||||
if len(payload) == 0 {
|
||||
buf.WriteString("null")
|
||||
} else {
|
||||
buf.Write(payload)
|
||||
}
|
||||
buf.WriteString(`,"ts":`)
|
||||
tsRaw, _ := json.Marshal(time.Now().UTC().Format(time.RFC3339Nano))
|
||||
buf.Write(tsRaw)
|
||||
buf.WriteByte('}')
|
||||
return buf.Bytes(), nil
|
||||
}
|
||||
```
|
||||
|
||||
Built by hand, not via `json.Marshal` of a struct. Reason: the `data` field is already-encoded JSON. Marshalling a struct with `Data json.RawMessage` works but goes through a generic encoder path. Concatenation is faster and avoids re-allocating for the largest field. At ~50k events/min peak, this matters.
|
||||
|
||||
## The Snapshot Store
|
||||
|
||||
```go
|
||||
// internal/snapshot/store.go
|
||||
const (
|
||||
keyPrefix = "state:"
|
||||
keyHashPrice = "state:coinbase_price"
|
||||
scanCount = 100
|
||||
wrongTypeMarker = "WRONGTYPE"
|
||||
)
|
||||
|
||||
func (s *Store) PutLatest(ctx context.Context, topic events.Topic, payload json.RawMessage) error {
|
||||
return s.rdb.Set(ctx, keyPrefix+string(topic), []byte(payload), 0).Err()
|
||||
}
|
||||
|
||||
func (s *Store) MergeSymbolMap(ctx context.Context, topic events.Topic, payload json.RawMessage) error {
|
||||
var tick struct{ Symbol string `json:"symbol"` }
|
||||
if err := json.Unmarshal(payload, &tick); err != nil || tick.Symbol == "" {
|
||||
return fmt.Errorf("coinbase merge: no symbol in payload")
|
||||
}
|
||||
key := keyPrefix + string(topic)
|
||||
err := s.rdb.HSet(ctx, key, tick.Symbol, []byte(payload)).Err()
|
||||
if err == nil { return nil }
|
||||
if !isWrongType(err) { return fmt.Errorf("redis hset %s/%s: %w", topic, tick.Symbol, err) }
|
||||
if delErr := s.rdb.Del(ctx, key).Err(); delErr != nil { ... }
|
||||
// retry as hash
|
||||
return s.rdb.HSet(ctx, key, tick.Symbol, []byte(payload)).Err()
|
||||
}
|
||||
```
|
||||
|
||||
`MergeSymbolMap` has a self-healing migration path: if the key exists as a string (legacy format) and `HSet` fails with `WRONGTYPE`, we delete and retry as a hash. This means a developer who upgraded mid-tick does not have to manually `redis-cli del`.
|
||||
|
||||
`GetAll` uses `SCAN` rather than `KEYS`. `KEYS *` is O(n) blocking on the Redis main thread; `SCAN` is incremental. For a snapshot store with ~16 keys this is overkill, but the habit is the right one — `KEYS` in any production code path is a foot-gun.
|
||||
|
||||
```go
|
||||
func (s *Store) readKey(ctx context.Context, key string) (json.RawMessage, error) {
|
||||
t, _ := s.rdb.Type(ctx, key).Result()
|
||||
switch t {
|
||||
case "string":
|
||||
v, err := s.rdb.Get(ctx, key).Bytes()
|
||||
return json.RawMessage(v), err
|
||||
case "hash":
|
||||
fields, _ := s.rdb.HGetAll(ctx, key).Result()
|
||||
m := make(map[string]json.RawMessage, len(fields))
|
||||
for k, v := range fields {
|
||||
m[k] = json.RawMessage(v)
|
||||
}
|
||||
return json.Marshal(m)
|
||||
}
|
||||
return nil, fmt.Errorf("unsupported redis type %q for key %s", t, key)
|
||||
}
|
||||
```
|
||||
|
||||
Type-aware deserialization. The snapshot HTTP handler is dumb — `core.OK(w, all)` and a no-cache header — because all the work happens in the store.
|
||||
|
||||
## The Alert Engine
|
||||
|
||||
`internal/alerts/engine.go` is the most complex part of the system because it has to compile predicates, refresh them, and dispatch with cooldowns and parallel delivery.
|
||||
|
||||
### Compile and Refresh
|
||||
|
||||
```go
|
||||
func (e *Engine) reload(ctx context.Context) error {
|
||||
all, _ := e.repo.ListAll(ctx)
|
||||
indexed := make(map[string][]compiledRule, len(all))
|
||||
for _, r := range all {
|
||||
prog, err := e.compile(r.Predicate)
|
||||
if err != nil {
|
||||
e.logger.Warn("alerts: skipping rule with bad predicate",
|
||||
"rule_id", r.ID, "topic", r.Topic, "err", err)
|
||||
continue
|
||||
}
|
||||
indexed[r.Topic] = append(indexed[r.Topic], compiledRule{rule: r, program: prog})
|
||||
}
|
||||
e.rulesByTopic.Store(&indexed)
|
||||
return nil
|
||||
}
|
||||
```
|
||||
|
||||
`atomic.Pointer[map[string][]compiledRule]` gives a lock-free read path. The map itself is never mutated after `Store`; reload always builds a fresh map and swaps the pointer. Readers grab the pointer once and operate on a stable map.
|
||||
|
||||
Compile errors do not fail the whole reload — a single bad rule predicate is logged and skipped. This matters: the rule UI lets users author predicates, and one user typing `event.severity = "high"` (assignment instead of comparison) should not silently break alerts for everyone.
|
||||
|
||||
### Evaluate
|
||||
|
||||
```go
|
||||
func (e *Engine) Evaluate(ctx context.Context, ev events.Event) {
|
||||
idx := e.rulesByTopic.Load()
|
||||
if idx == nil { return }
|
||||
rules, ok := (*idx)[string(ev.Topic)]
|
||||
if !ok || len(rules) == 0 { return }
|
||||
|
||||
payload, err := normalizePayload(ev.Payload)
|
||||
if err != nil {
|
||||
e.logger.Warn("alerts: payload normalize failed", ...)
|
||||
return
|
||||
}
|
||||
|
||||
for _, cr := range rules {
|
||||
if !cr.rule.Enabled { continue }
|
||||
match, err := evalPredicate(cr.program, payload)
|
||||
if err != nil { ... continue }
|
||||
if !match { continue }
|
||||
e.fire(ctx, cr.rule, ev, payload)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
`normalizePayload` handles the `any` type discipline. The bus carries `Payload any`; CEL needs a `map[string]any` for field access. The function tries the cheap path (already a map), then `json.RawMessage` (unmarshal once), then full round-trip via `json.Marshal`+`Unmarshal`. The bulk of events come in as `json.RawMessage` because that is what the collectors produce.
|
||||
|
||||
### Fire
|
||||
|
||||
```go
|
||||
func (e *Engine) fire(ctx context.Context, rule Rule, ev events.Event, payload map[string]any) {
|
||||
channels, _ := e.loader.LoadChannels(ctx, rule.UserID)
|
||||
cooldown := time.Duration(rule.CooldownSec) * time.Second
|
||||
message := formatMessage(rule, ev, payload)
|
||||
|
||||
var wg sync.WaitGroup
|
||||
var mu sync.Mutex
|
||||
delivered := []string{}
|
||||
deliveryErrs := map[string]string{}
|
||||
|
||||
for _, ch := range channels {
|
||||
ch := ch
|
||||
key := fmt.Sprintf("alert_cooldown:%s:%s:%s", rule.ID, ch.Type, ch.ID)
|
||||
ok, _ := e.cooldowns.TryAcquire(ctx, key, cooldown)
|
||||
if !ok { continue }
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
err := e.notifier.SendAlert(ctx, ch, message)
|
||||
mu.Lock(); defer mu.Unlock()
|
||||
if err != nil {
|
||||
deliveryErrs[ch.ID] = err.Error()
|
||||
return
|
||||
}
|
||||
delivered = append(delivered, ch.ID)
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
body, _ := json.Marshal(payload)
|
||||
errBody, _ := json.Marshal(deliveryErrs)
|
||||
_ = e.repo.RecordFire(ctx, HistoryRow{
|
||||
RuleID: rule.ID, UserID: rule.UserID,
|
||||
FiredAt: time.Now().UTC(),
|
||||
Payload: body,
|
||||
DeliveredTo: delivered,
|
||||
DeliveryErrors: errBody,
|
||||
})
|
||||
}
|
||||
```
|
||||
|
||||
Three things worth pointing at:
|
||||
|
||||
- **Per-channel cooldown via Redis `SET NX EX`**. The `RedisCooldown.TryAcquire` is a single round-trip and the TTL ensures the key auto-cleans.
|
||||
- **Parallel delivery**. Telegram is slow (2-5s for a botless rate-limited send), Slack is fast (~200ms). Sequential delivery would pace everything to the slowest. We dispatch all channels concurrently, then `wg.Wait` for all to return.
|
||||
- **Fire history is recorded once per rule firing, not per channel**. `delivered` lists which channels succeeded; `deliveryErrs` maps channel ID to error text. That makes the UI's "alert history" page useful for debugging integration failures.
|
||||
|
||||
## Webhook Encryption
|
||||
|
||||
```go
|
||||
// internal/notifications/crypto.go
|
||||
func (e *Encryptor) Encrypt(plaintext []byte) (ciphertext, nonce []byte, err error) {
|
||||
block, _ := aes.NewCipher(e.key)
|
||||
gcm, _ := cipher.NewGCM(block)
|
||||
nonce = make([]byte, gcm.NonceSize())
|
||||
if _, err := io.ReadFull(rand.Reader, nonce); err != nil { ... }
|
||||
ciphertext = gcm.Seal(nil, nonce, plaintext, nil)
|
||||
return ciphertext, nonce, nil
|
||||
}
|
||||
```
|
||||
|
||||
Standard library AES-256-GCM. The key is base64-decoded from `NOTIFICATION_ENCRYPTION_KEY` and validated to be exactly 32 bytes at startup:
|
||||
|
||||
```go
|
||||
func NewEncryptor(b64Key string) (*Encryptor, error) {
|
||||
key, _ := base64.StdEncoding.DecodeString(b64Key)
|
||||
if len(key) != 32 {
|
||||
return nil, fmt.Errorf("encryption key must be 32 bytes, got %d", len(key))
|
||||
}
|
||||
return &Encryptor{key: key}, nil
|
||||
}
|
||||
```
|
||||
|
||||
If `NewEncryptor` returns an error, the entire alert engine is disabled for the lifetime of the process — `bridge` and `notifBridge` stay nil, so `notifBridge != nil` checks in `cmd/api/main.go` skip the engine setup. This is **fail-closed** behaviour: misconfiguration disables the feature rather than running it with broken/no encryption.
|
||||
|
||||
## Frontend Wiring
|
||||
|
||||
### The WebSocket Driver
|
||||
|
||||
`frontend/src/api/ws.ts`:
|
||||
|
||||
```ts
|
||||
export function createDashboardWS(opts: CreateDashboardWSOpts): DashboardWS {
|
||||
const backoff = opts.backoff ?? DEFAULT_BACKOFF
|
||||
const onEvent = opts.onEvent ?? (() => undefined)
|
||||
|
||||
let driver: WSDriver | null = null
|
||||
let ready = false
|
||||
let opened = false
|
||||
let closed = false
|
||||
let buffer: WSEvent[] = []
|
||||
let nextDelay = backoff.initialMs
|
||||
let reconnectTimer: ReturnType<typeof setTimeout> | null = null
|
||||
|
||||
function attach() {
|
||||
if (closed) return
|
||||
const d = opts.driver()
|
||||
driver = d
|
||||
opened = false
|
||||
d.onOpen = () => {
|
||||
nextDelay = backoff.initialMs
|
||||
opened = true
|
||||
sendInitIfReady()
|
||||
}
|
||||
d.onMessage = (data) => {
|
||||
let parsed: WSEvent
|
||||
try { parsed = JSON.parse(data) as WSEvent } catch { return }
|
||||
if (ready) onEvent(parsed)
|
||||
else buffer.push(parsed)
|
||||
}
|
||||
d.onClose = () => {
|
||||
driver = null
|
||||
opened = false
|
||||
if (closed) return
|
||||
const delay = nextDelay
|
||||
nextDelay = Math.min(nextDelay * 2, backoff.maxMs)
|
||||
reconnectTimer = setTimeout(attach, delay)
|
||||
}
|
||||
}
|
||||
return {
|
||||
connect() { attach() },
|
||||
setReady() {
|
||||
ready = true
|
||||
sendInitIfReady()
|
||||
const flush = buffer
|
||||
buffer = []
|
||||
for (const ev of flush) onEvent(ev)
|
||||
},
|
||||
disconnect() { ... },
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
The factory takes a `driver` thunk so tests can inject a fake. `browserDriver` is the production implementation that wraps the platform `WebSocket`. The buffer-flush in `setReady` is the snapshot+stream handshake on the client side.
|
||||
|
||||
Reconnect uses exponential backoff with `nextDelay * 2` capped at `maxMs` (default 30s). On a clean open, `nextDelay` resets so a reconnect after long uptime starts at 1s again.
|
||||
|
||||
### Store Routing
|
||||
|
||||
The `App` boot sequence (paraphrased):
|
||||
|
||||
1. Mount providers (TanStack Query client, Zustand stores already global).
|
||||
2. Open the WebSocket via `createDashboardWS`. Hand it an `onEvent` that dispatches by `ev.ch` into the right store.
|
||||
3. Fetch `/v1/snapshot`. On success, hydrate every store from the response.
|
||||
4. Call `ws.setReady()`. Buffered events flush through `onEvent` into the now-hydrated stores.
|
||||
|
||||
Stores are flat by topic. For example, `stores/cve.ts` holds the recent CVE list and exposes a setter (`upsertCVE`) that the WS dispatcher calls. Components subscribe via `useCVEStore(state => state.list)` so a Coinbase tick does not re-render the CVE panel.
|
||||
|
||||
### The Globe
|
||||
|
||||
`frontend/src/pages/globe/Globe.tsx` is the big consumer. It subscribes to the `globeEvents` store, which is a fan-in of the topic-specific stores: outages, BGP hijacks, earthquakes, ransomware victims, mass-scan ASN dots, ISS position. The globe component listens, computes the layer mutations, and applies them in a `requestAnimationFrame` batch.
|
||||
|
||||
The TLE-driven ISS orbit track is propagated client-side. Every animation frame, the SGP4 propagator advances the satellite's mean anomaly and emits a (lat, lon) the globe layer renders as a polyline. The TLE itself comes from the snapshot or a `iss_position` event with a `tle` field; the position pin (separate from the orbit track) comes from the `wheretheiss.at` 10-second poll.
|
||||
|
||||
## Common Pitfalls
|
||||
|
||||
A short list of mistakes we made and you might too.
|
||||
|
||||
**Forgetting to register a topic in `IsValid`/`AllTopics`.** Adding a topic constant is not enough. The WebSocket handler validates `?topics=` against `IsValid`, and the default fallback uses `AllTopics`. Miss either and your new topic silently never reaches the browser.
|
||||
|
||||
**Putting expensive work in `Bus.Run`.** The drain loop is single-goroutine. If you add a synchronous DB write before the broadcast, you cap the entire system's broadcast rate at the DB write rate. Move expensive work into a subscriber channel or the persister implementation.
|
||||
|
||||
**Using `c.CloseRead` on `coder/websocket`.** It looks helpful. It closes the connection on any received frame, including the client `init` op the dashboard uses. Either drain reads explicitly or design a strict server-push protocol with close-on-receive semantics — pick one.
|
||||
|
||||
**Not validating the encryption key length at startup.** AES-256 silently accepts a 16-byte key as AES-128 if you do not check. The `NewEncryptor` constructor must enforce `len(key) == 32`.
|
||||
|
||||
**Returning `*MinuteBar` from the aggregator and forgetting to nil-check.** The `Push` return signature `(*MinuteBar, MinuteBar)` puts the closed bar behind a pointer. Persisting `*closed` without a nil check is the most common bug we caught in code review on the Coinbase package.
|
||||
|
||||
**B-tree on time columns at scale.** Tempting because it's the default. For tens of millions of append-mostly rows, BRIN is dozens of times smaller and just as fast for range scans. If you use B-tree on `occurred_at`, your index is bigger than the table within months.
|
||||
|
||||
**Keying alert cooldowns globally instead of per-channel.** A user with Telegram and Slack rules on the same predicate wants an independent cooldown on each. Keying on `rule_id` alone causes the channel that fires first to mute the other.
|
||||
|
||||
## Where To Go Next
|
||||
|
||||
Continue to [04-CHALLENGES.md](04-CHALLENGES.md) for extension ideas, harder feature work, and deployment hardening.
|
||||
|
|
@ -0,0 +1,181 @@
|
|||
<!--
|
||||
©AngelaMos | 2026
|
||||
04-CHALLENGES.md
|
||||
-->
|
||||
|
||||
# Challenges
|
||||
|
||||
You have a working dashboard. The interesting question is what to add to it. This file is a graded list of extensions, from "evening project" to "you should publish a blog post about it". Each entry includes a hint and a real-world reason it would matter.
|
||||
|
||||
## Easy
|
||||
|
||||
These take a few hours each. The goal is to build muscle memory for the collector + topic + frontend store wiring.
|
||||
|
||||
### Add an additional vendor advisory feed
|
||||
|
||||
Pick one of: Microsoft Security Response Center (MSRC), Cisco PSIRT, Red Hat Security Advisories, Project Zero. Each publishes RSS or JSON.
|
||||
|
||||
**Hint**: copy the `kev` collector layout. You need a `Fetcher` that does HTTP+parse, a `Repository` for upsert, a new topic constant, and a frontend store. The MSRC CVRF feed is a good first pick because it ships JSON, not custom XML.
|
||||
|
||||
**Why it matters**: NVD is consistently 1-7 days behind vendor disclosure. A direct vendor advisory feed is the single highest-leverage upgrade you can make for a defensive dashboard.
|
||||
|
||||
### Add a Tor exit node feed
|
||||
|
||||
The `tornull/exit-list` (or `dan.me.uk/torlist`) publishes the current Tor exit node list. Wire it up as a feed; on the frontend, paint Tor exits with a different color when they appear in the DShield top-IPs panel.
|
||||
|
||||
**Hint**: the simplest implementation is a 30-minute poller into a Redis set. The DShield enricher then does an `SISMEMBER` per IP at enrichment time and stamps `is_tor: true` on the payload.
|
||||
|
||||
**Why it matters**: Tor exit IPs scanning your edge are categorically different from compromised home routers scanning your edge. Visual differentiation is faster than a tooltip.
|
||||
|
||||
### Persist alert fire history with a UI page
|
||||
|
||||
The schema and `RecordFire` already exist (see `internal/alerts/repository.go`). What is missing is a `/v1/alerts/history` endpoint and a settings-page table.
|
||||
|
||||
**Hint**: model the endpoint after `/v1/intel/cve/search` for pagination, sort, and filter shape. Frontend route is already in the settings page tree (`pages/settings/Alerts.tsx` or similar).
|
||||
|
||||
**Why it matters**: when an alert fires and the user did not see it, the first question is "did it actually go to Slack?" The history page answers that without a database round trip.
|
||||
|
||||
### Add a "test alert" button
|
||||
|
||||
For each configured channel, render a button that posts to `/v1/notifications/channels/:id/test`. The endpoint sends a fixed `"This is a test alert from monitor-the-situation"` message via the channel's notifier.
|
||||
|
||||
**Hint**: this is a one-route endpoint plus a one-button frontend change. The work is the wiring, not the logic. Reuse `notifSender.Send(...)` directly.
|
||||
|
||||
**Why it matters**: the most common "alerts don't work" support question is actually "I configured the webhook wrong". A test button shifts that left.
|
||||
|
||||
## Medium
|
||||
|
||||
These are weekend projects. They require touching the bus, the storage layer, or both.
|
||||
|
||||
### Add MITRE ATT&CK technique tagging to ransomware victims
|
||||
|
||||
ransomware.live posts include the threat actor name. Map actor names to known ATT&CK technique sets (CISA publishes mappings; so does MITRE's ATT&CK Navigator). Store the technique IDs on the ransomware row and surface them as small badges on the panel.
|
||||
|
||||
**Hint**: the mapping itself is data, not code. Build a `actor_tactics.yaml` and load it at startup. The collector enriches the payload before persistence and emit. Bonus: when a previously-unmapped actor appears, log a warning so you remember to extend the mapping.
|
||||
|
||||
**Why it matters**: knowing that REvil uses T1486 (Data Encrypted for Impact) and T1078 (Valid Accounts) tells a defender what to check when REvil shows up in the wild. Names are not actionable; techniques are.
|
||||
|
||||
### Add a "follow-the-money" Bitcoin ransomware tracker
|
||||
|
||||
When a ransomware victim post includes a wallet address (some leak posts do), look it up against Chainalysis-style public sources or use blockchair's API for transaction history. Render the wallet's recent inbound transfers as a sparkline.
|
||||
|
||||
**Hint**: this is the only feed in the project that has an unbounded fan-out (one wallet → many tx history calls), so put it behind a per-wallet cache with a long TTL. Do not re-fetch on every page render.
|
||||
|
||||
**Why it matters**: ransomware payment flows are public information on the blockchain. Surfacing them next to the victim post is unique to the dashboard's "fuse signals" thesis.
|
||||
|
||||
### Replace the in-process bus with NATS JetStream
|
||||
|
||||
Keep the same `Bus` interface (`Emit`, `Subscribe`, `Run`). Swap the implementation from a Go channel to NATS JetStream. Now the dashboard can run multiple instances behind a load balancer.
|
||||
|
||||
**Hint**: JetStream's per-stream subjects map cleanly to topics. Use a durable consumer per backend instance for the alerts dispatcher, and pull-based subscription for the WebSocket hub broadcast. Keep persistence of the snapshot in Redis, not JetStream — Redis stays as the read-side cache.
|
||||
|
||||
**Why it matters**: the natural next step when one operator becomes a small team. The architecture chapter promised this would be a one-component replacement; the challenge is to actually verify that.
|
||||
|
||||
### Add OpenTelemetry traces across the bus
|
||||
|
||||
`core.NewTelemetry` already initializes the OTel tracer. Wire trace context through `Bus.Emit` so a single CVE event can be traced from NVD fetch → upsert → bus → snapshot → ws broadcast → frontend ack.
|
||||
|
||||
**Hint**: add a `context.Context` parameter to `Emit` (today it does not take one). Inject the span context into the event payload as a `traceparent` header. The persister and broadcaster pick it up. Frontend logs a `console.debug` with the trace ID so you can correlate in Tempo or Honeycomb.
|
||||
|
||||
**Why it matters**: when one panel is "stale" and another is fresh, distributed tracing tells you which collector is slow. Without traces, you read logs and guess.
|
||||
|
||||
### Migrate from polling to NVD's CVE 2.0 event stream when it lands
|
||||
|
||||
NVD has been promising an event-driven feed for years. When it ships (or if it has by the time you read this), replacing the 2-hour CVE poll with a streaming subscription cuts dashboard latency on new CVEs from "up to two hours" to "seconds". The collector layout is already prepared for this — `coinbase` is the streaming reference implementation.
|
||||
|
||||
**Hint**: model on `internal/collectors/coinbase/`. The hardest part is gap recovery on reconnect, which `sequencer.go` shows.
|
||||
|
||||
**Why it matters**: if you build dashboards for a SOC, latency is a feature. The team using your tool will notice.
|
||||
|
||||
## Hard
|
||||
|
||||
These are blog-post-grade projects. Each is a multi-week investment.
|
||||
|
||||
### Add a YARA-based malware feed via Malshare or VirusTotal Hunting
|
||||
|
||||
Malshare publishes a daily feed of new malware hashes. VirusTotal Hunting has live YARA matches. Wire either in, persist matches with the YARA rule that matched, and surface "this rule fired N times in the last 24h" on a new panel.
|
||||
|
||||
**Hint**: VT Hunting requires a paid API key. Start with Malshare. The frontend panel can be a sparkline per rule plus a click-through to the latest matched hash.
|
||||
|
||||
**Why it matters**: connecting a YARA hit to a CVE diff to a KEV entry to a ransomware leak is the dashboard's whole pitch. This adds the file-side signal to the network-side and patch-side ones already there.
|
||||
|
||||
### Build a "what just changed?" diff replay
|
||||
|
||||
Right now every panel shows current state. Add a time-machine slider: drag back, the dashboard reconstructs every panel as of that moment.
|
||||
|
||||
**Hint**: this is what BRIN was built for. Every panel store needs an `at(t time.Time)` query against the historical Postgres tables. The frontend needs to disconnect the WebSocket, render from the historical query, and reconnect when the user releases the slider. Watch out for the Coinbase panel — minute bars are sparse compared to ticks; you have to choose which to show.
|
||||
|
||||
**Why it matters**: post-incident review on "what was the dashboard showing 30 minutes before the incident page" is the scariest question to leave unanswered.
|
||||
|
||||
### Add MOAS hijack detection from a public BGP collector (not Cloudflare)
|
||||
|
||||
CF Radar's hijack detection is closed-source. RIPE's RIS Live (`ris-live.ripe.net`) is a free public BGP feed at firehose volume. Implement your own MOAS detector from the raw stream: maintain a sliding window of `(prefix, origin_as)` observations, alert when a prefix has two distinct origin ASNs simultaneously.
|
||||
|
||||
**Hint**: this is at least a week of work. Start by writing a `client.go` that subscribes to RIS Live's WebSocket. Keep observations in a bounded LRU. Tune confidence by AS-relationships (sibling ASNs are not hijacks, peers might be).
|
||||
|
||||
**Why it matters**: building a hijack detector from raw BGP data is the canonical "you understand internet routing now" project. CF Radar is convenient; RIS Live is educational.
|
||||
|
||||
### Multi-tenant SaaS-ify the dashboard
|
||||
|
||||
Right now there is one user (the operator). Turn it into a hosted SaaS where each customer gets isolated alert rules, channel configs, and rate-limit budgets. Same data feeds (the world is the world), per-tenant alert state.
|
||||
|
||||
**Hint**: the `users` table and `auth` package are already most of the way there. The hard part is data partitioning: alert rules and channels are per-tenant; CVE/KEV/ransomware data is shared. The frontend has to gate admin pages by org membership. Stripe billing is its own thing. Plan for at least a month.
|
||||
|
||||
**Why it matters**: this is the path from "personal project" to "actual product". Doing it well teaches multi-tenancy patterns that apply to every B2B SaaS.
|
||||
|
||||
### Add a WebGL flow visualization on top of the globe
|
||||
|
||||
Today the globe shows discrete events: dots, hijack arcs, ISS pin. Add a WebGL particle system that flows along the BGP arcs continuously, with intensity proportional to detected hijack confidence. Then, when an outage is active, render the country in pulsing red with rate proportional to the affected ASN count.
|
||||
|
||||
**Hint**: MapLibre's custom layers API supports raw WebGL passes. Look at deck.gl's TripsLayer and ArcLayer for reference. Particle counts get expensive fast — cap at ~5000 active particles per frame.
|
||||
|
||||
**Why it matters**: this is what makes a screenshot of the dashboard go viral. Pretty matters when a tool's audience is operators who stare at it for 8 hours a day.
|
||||
|
||||
## Deployment Hardening
|
||||
|
||||
Real production deployments need these. The current Justfile gets you 80% there.
|
||||
|
||||
### Run the backend behind a non-Cloudflare proxy as well
|
||||
|
||||
`cloudflared.compose.yml` covers Cloudflare Tunnel. Add an alternate `compose.traefik.yml` and `compose.nginx-direct.yml` for self-hosted edge. The configs differ in TLS termination, header forwarding, and WebSocket upgrade rules.
|
||||
|
||||
**Hint**: the `TrustedProxyHops` config is the key knob. Cloudflare adds known CF-Connecting-IP headers; Traefik adds `X-Forwarded-For`. Get the trusted hop count right per setup or your rate limits and WS-per-IP caps run on the proxy IP, not the client IP.
|
||||
|
||||
### Add a Postgres read replica for `/v1/intel/*`
|
||||
|
||||
The historical query endpoints are read-heavy and tolerant of seconds-of-lag. Point a `pgx` read pool at a streaming replica; route `intel.Handler` reads to it.
|
||||
|
||||
**Hint**: this is one new pool in `core` and one config knob. The collectors keep writing to the primary. Only `intel/handler.go` reads need re-pointing.
|
||||
|
||||
### Add Prometheus metrics for the bus and hub
|
||||
|
||||
`Bus.DroppedCount`, `Bus.SubscriberDroppedCount`, `Hub.SubscriberCount` are already exposed in code but only logged. Wire them to a `prometheus.GaugeFunc`. Add latency histograms on collector ticks via `core.Telemetry`.
|
||||
|
||||
**Hint**: Otel already exports to Prometheus if you set the exporter to `prometheus`. The work is naming the metrics consistently and writing the alerting rules (e.g., page on `bus_dropped_total` rate > 0 for 5m).
|
||||
|
||||
### Dependabot, Renovate, and supply-chain pinning
|
||||
|
||||
The Go module file already pins by checksum. Add `.github/dependabot.yml` for Go and pnpm. Lock the docker base images by digest, not tag. Fail CI when `go mod download -x` reports any unverified module.
|
||||
|
||||
**Hint**: the `bug-bounty-platform` project in this repo has a CI hardening setup worth copying.
|
||||
|
||||
### Build SBOMs and attest provenance
|
||||
|
||||
`syft` produces SBOMs. `slsa-github-generator` produces SLSA provenance. Add a release workflow that emits both, signs the binary with `cosign`, and pushes signed images to GHCR with attestation.
|
||||
|
||||
**Hint**: the SBOM is one `syft packages dir:.` invocation. The provenance is a GitHub-actions reusable workflow. The hard part is shipping `cosign verify` into the deploy pipeline so unsigned images cannot deploy.
|
||||
|
||||
## Connections to Other Projects in This Repo
|
||||
|
||||
A few extension ideas that pull in code from sibling projects:
|
||||
|
||||
- **Honeypot-network** publishes attacker IPs. Pipe them in as a 13th feed and color the DShield panel rows that overlap.
|
||||
- **API rate limiter** — its sliding-window store would be a drop-in upgrade for the per-IP WebSocket connection cap.
|
||||
- **Bug bounty platform** — when a CVE on the dashboard maps to an in-scope program, surface the program link on the CVE row.
|
||||
- **AI threat detection** — wire its anomaly score on attacker IPs into the DShield enricher so high-anomaly IPs render with a different glyph.
|
||||
|
||||
## Final Note
|
||||
|
||||
The dashboard is not "done". It is an opinionated foundation. Every extension on this list has been sketched, and a few prototyped, in the lifetime of the project. None of them require ripping the architecture apart, because the architecture was specifically chosen to keep the producer-bus-consumer separation clean.
|
||||
|
||||
If you build something interesting on top of it, send a PR or open an issue. Especially the hard category — the goal is for this codebase to be the cleanest reference implementation of "real-time security operations dashboard" anyone has published.
|
||||
|
|
@ -126,7 +126,7 @@ Tools, courses, certifications, communities, and frameworks for cybersecurity pr
|
|||
| **[Network Covert Channel](./SYNOPSES/advanced/Network.Covert.Channel.md)**<br>Data exfiltration techniques |    | Covert channels • Data exfiltration • Steganography<br>[Learn More](./SYNOPSES/advanced/Network.Covert.Channel.md) |
|
||||
| **[Automated Penetration Testing](./SYNOPSES/advanced/Automated.Penetration.Testing.md)**<br>Full pentest automation |    | Pentest automation • Recon to exploitation • Report generation<br>[Learn More](./SYNOPSES/advanced/Automated.Penetration.Testing.md) |
|
||||
| **[Haskell Reverse Proxy](./PROJECTS/advanced/haskell-reverse-proxy)**<br>Functional reverse proxy with security middleware |    | Functional programming • Reverse proxy design • Security middleware • Haskell<br>[Source Code](./PROJECTS/advanced/haskell-reverse-proxy) |
|
||||
| **["Monitor the Situation" Dashboard](./SYNOPSES/advanced/Monitor.The.Situation.Dashboard.md)**<br>Real-time cyber threat situational awareness |     | Threat intelligence • CVE tracking • MITRE ATT&CK • OSINT • Real-time dashboards<br>[Source Code](./PROJECTS/advanced/monitor-the-situation-dashboard) |
|
||||
| **["Monitor the Situation" Dashboard](./PROJECTS/advanced/monitor-the-situation-dashboard)**<br>Real-time cyber threat situational awareness |      | Threat intel feeds • EPSS/KEV/CVE velocity • BGP hijacks • WebSocket fan-out • 3D globe SOC view<br>[Source Code](./PROJECTS/advanced/monitor-the-situation-dashboard) \| [Docs](./PROJECTS/advanced/monitor-the-situation-dashboard/learn) |
|
||||
| **[Honeypot Network](./PROJECTS/advanced/honeypot-network)**<br>Multi-service honeypot deployment & analysis |      | Honeypot deployment • Attacker behavior analysis • IOC extraction • MITRE mapping<br>[Source Code](./PROJECTS/advanced/honeypot-network) \| [Docs](./PROJECTS/advanced/honeypot-network/learn) |
|
||||
| **[Supply Chain Security Analyzer](./SYNOPSES/advanced/Supply.Chain.Security.Analyzer.md)**<br>Dependency vulnerability analysis |    | Supply chain security • Dependency analysis • Malicious packages<br>[Learn More](./SYNOPSES/advanced/Supply.Chain.Security.Analyzer.md) |
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue