From e216331aeb6a9e153ee081d5e98271126dbe22de Mon Sep 17 00:00:00 2001 From: CarterPerez-dev Date: Sun, 3 May 2026 18:20:15 -0400 Subject: [PATCH] feat(monitor/backend): abuseipdb enrichment for bgp hijacks (country + abuse_score in payload) --- .../.env.example | 1 + .../monitor-the-situation-dashboard/README.md | 2 +- .../backend/cmd/api/main.go | 24 ++++ .../internal/collectors/cfradar/collector.go | 41 ++++++- .../collectors/cfradar/collector_test.go | 101 +++++++++++++++++ .../backend/internal/config/config.go | 8 ++ .../internal/enrich/abuseipdb/client.go | 106 ++++++++++++++++++ .../internal/enrich/abuseipdb/client_test.go | 61 ++++++++++ .../abuseipdb/testdata/check_lookup.json | 19 ++++ 9 files changed, 361 insertions(+), 2 deletions(-) create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/enrich/abuseipdb/client.go create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/enrich/abuseipdb/client_test.go create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/enrich/abuseipdb/testdata/check_lookup.json diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/.env.example b/PROJECTS/advanced/monitor-the-situation-dashboard/.env.example index 003d47e0..d597abf8 100644 --- a/PROJECTS/advanced/monitor-the-situation-dashboard/.env.example +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/.env.example @@ -22,6 +22,7 @@ NOTIFICATION_ENCRYPTION_KEY= NVD_API_KEY= CF_RADAR_TOKEN= GREYNOISE_API_KEY= +ABUSEIPDB_API_KEY= CLOUDFLARE_TUNNEL_TOKEN= diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/README.md b/PROJECTS/advanced/monitor-the-situation-dashboard/README.md index e04ffd14..6b62b30c 100644 --- a/PROJECTS/advanced/monitor-the-situation-dashboard/README.md +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/README.md @@ -33,7 +33,7 @@ Operator-grade real-time situational awareness dashboard. Single-binary Go backe | Space weather | NOAA SWPC (5 endpoints) | 1m / 3h | none | | World events | Wikipedia ITN + GDELT v2 API | 5m / 15m | none | | ISS position | wheretheiss.at + CelesTrak | 10s / 24h | none | -| IP enrichment | GreyNoise Community | on-demand | `GREYNOISE_API_KEY` (optional, free tier) | +| IP enrichment (BGP hijacks) | AbuseIPDB | on-demand | `ABUSEIPDB_API_KEY` (optional, free tier 1k/day) | ## Quickstart (development) diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/cmd/api/main.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/cmd/api/main.go index b872c50a..364dd9a3 100644 --- a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/cmd/api/main.go +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/cmd/api/main.go @@ -35,6 +35,7 @@ import ( "github.com/carterperez-dev/monitor-the-situation/backend/internal/collectors/wikipedia" "github.com/carterperez-dev/monitor-the-situation/backend/internal/config" "github.com/carterperez-dev/monitor-the-situation/backend/internal/core" + "github.com/carterperez-dev/monitor-the-situation/backend/internal/enrich/abuseipdb" "github.com/carterperez-dev/monitor-the-situation/backend/internal/health" "github.com/carterperez-dev/monitor-the-situation/backend/internal/middleware" "github.com/carterperez-dev/monitor-the-situation/backend/internal/redisring" @@ -48,6 +49,22 @@ const ( drainDelay = 5 * time.Second ) +type abuseipdbEnricher struct { + client *abuseipdb.Client +} + +func (a abuseipdbEnricher) Lookup(ctx context.Context, ip string) (cfradar.Enrichment, error) { + v, err := a.client.Lookup(ctx, ip) + if err != nil { + return cfradar.Enrichment{}, err + } + return cfradar.Enrichment{ + Country: v.CountryCode, + AbuseConfidence: v.AbuseConfidenceScore, + ISP: v.ISP, + }, nil +} + func main() { configPath := flag.String("config", "config.yaml", "path to config file") flag.Parse() @@ -178,6 +195,12 @@ func run(configPath string) error { } if cfg.Collectors.CFRadar.Enabled { + var enricher cfradar.Enricher + if cfg.Collectors.AbuseIPDB.Enabled && cfg.Collectors.AbuseIPDB.APIKey != "" { + enricher = abuseipdbEnricher{ + client: abuseipdb.NewClient(abuseipdb.ClientConfig{APIKey: cfg.Collectors.AbuseIPDB.APIKey}), + } + } coll := cfradar.NewCollector(cfradar.CollectorConfig{ Interval: cfg.Collectors.CFRadar.Interval, MinConfidence: cfg.Collectors.CFRadar.MinConfidence, @@ -185,6 +208,7 @@ func run(configPath string) error { Repo: cfradar.NewRepo(db.DB), Emitter: eventBus, State: collectorState, + Enricher: enricher, Logger: logger.With("collector", "cfradar"), }) collectorGroup.Go(func() error { return coll.Run(collectorCtx) }) diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/cfradar/collector.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/cfradar/collector.go index d8c9c09a..add1eca8 100644 --- a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/cfradar/collector.go +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/cfradar/collector.go @@ -7,6 +7,7 @@ import ( "context" "encoding/json" "log/slog" + "net/netip" "time" "github.com/lib/pq" @@ -41,6 +42,22 @@ type StateRecorder interface { RecordError(ctx context.Context, name, errMsg string) error } +type Enricher interface { + Lookup(ctx context.Context, ip string) (Enrichment, error) +} + +type Enrichment struct { + Country string `json:"country,omitempty"` + AbuseConfidence int `json:"abuse_confidence,omitempty"` + ISP string `json:"isp,omitempty"` + CheckedIP string `json:"checked_ip,omitempty"` +} + +type EnrichedHijack struct { + HijackEvent + Enrichment *Enrichment `json:"enrichment,omitempty"` +} + type CollectorConfig struct { Interval time.Duration MinConfidence int @@ -48,6 +65,7 @@ type CollectorConfig struct { Repo Repository Emitter Emitter State StateRecorder + Enricher Enricher Logger *slog.Logger } @@ -183,7 +201,20 @@ func (c *Collector) tickHijacks(ctx context.Context) (int64, error) { if known[e.ID] { continue } - rawBytes, _ := json.Marshal(e) + enriched := EnrichedHijack{HijackEvent: e} + if c.cfg.Enricher != nil && len(e.Prefixes) > 0 { + if ip, ok := representativeIP(e.Prefixes[0]); ok { + if v, lerr := c.cfg.Enricher.Lookup(ctx, ip); lerr == nil { + enriched.Enrichment = &Enrichment{ + Country: v.Country, + AbuseConfidence: v.AbuseConfidence, + ISP: v.ISP, + CheckedIP: ip, + } + } + } + } + rawBytes, _ := json.Marshal(enriched) raw := json.RawMessage(rawBytes) row := HijackRow{ ID: e.ID, @@ -210,3 +241,11 @@ func (c *Collector) tickHijacks(ctx context.Context) (int64, error) { } return emitted, nil } + +func representativeIP(cidr string) (string, bool) { + p, err := netip.ParsePrefix(cidr) + if err != nil { + return "", false + } + return p.Addr().String(), true +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/cfradar/collector_test.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/cfradar/collector_test.go index c0f93853..9304271e 100644 --- a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/cfradar/collector_test.go +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/cfradar/collector_test.go @@ -95,6 +95,22 @@ type noopState struct{} func (noopState) RecordSuccess(context.Context, string, int64) error { return nil } func (noopState) RecordError(context.Context, string, string) error { return nil } +type fakeEnricher struct { + calls int + verdict cfradar.Enrichment + lastIP string + returnFn func(ip string) (cfradar.Enrichment, error) +} + +func (e *fakeEnricher) Lookup(_ context.Context, ip string) (cfradar.Enrichment, error) { + e.calls++ + e.lastIP = ip + if e.returnFn != nil { + return e.returnFn(ip) + } + return e.verdict, nil +} + func TestCollector_OnlyEmitsNetNew(t *testing.T) { now := time.Now().UTC() ftch := &fakeFetcher{ @@ -146,6 +162,91 @@ func TestCollector_OnlyEmitsNetNew(t *testing.T) { require.GreaterOrEqual(t, hijackEvents, 1) } +func TestCollector_EnrichesHijackPayloadWhenEnricherProvided(t *testing.T) { + now := time.Now().UTC() + ftch := &fakeFetcher{ + hijacks: cfradar.HijackBody{Events: []cfradar.HijackEvent{ + { + ID: 501, + DetectedAt: now, + StartedAt: now, + Confidence: 9, + HijackerASN: 4242, + Prefixes: []string{"203.0.113.0/24"}, + }, + }}, + } + repo := &fakeRepo{knownHijacks: map[int64]bool{}} + emt := &fakeEmitter{} + enr := &fakeEnricher{verdict: cfradar.Enrichment{ + Country: "RU", + AbuseConfidence: 95, + ISP: "ExampleNet", + }} + + c := cfradar.NewCollector(cfradar.CollectorConfig{ + Interval: 30 * time.Millisecond, + MinConfidence: 7, + Fetcher: ftch, + Repo: repo, + Emitter: emt, + State: noopState{}, + Enricher: enr, + }) + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + _ = c.Run(ctx) + + require.GreaterOrEqual(t, enr.calls, 1) + require.Equal(t, "203.0.113.0", enr.lastIP) + + var found bool + for _, ev := range emt.Events() { + if ev.Topic != events.TopicBGPHijack { + continue + } + body, _ := json.Marshal(ev.Payload) + require.Contains(t, string(body), `"country":"RU"`) + require.Contains(t, string(body), `"abuse_confidence":95`) + require.Contains(t, string(body), `"checked_ip":"203.0.113.0"`) + found = true + } + require.True(t, found, "expected at least one hijack event with enrichment") +} + +func TestCollector_HijackEmitsRawWhenNoEnricher(t *testing.T) { + now := time.Now().UTC() + ftch := &fakeFetcher{ + hijacks: cfradar.HijackBody{Events: []cfradar.HijackEvent{ + {ID: 777, DetectedAt: now, StartedAt: now, Confidence: 9, Prefixes: []string{"198.51.100.0/24"}}, + }}, + } + repo := &fakeRepo{knownHijacks: map[int64]bool{}} + emt := &fakeEmitter{} + + c := cfradar.NewCollector(cfradar.CollectorConfig{ + Interval: 30 * time.Millisecond, + MinConfidence: 7, + Fetcher: ftch, + Repo: repo, + Emitter: emt, + State: noopState{}, + }) + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + _ = c.Run(ctx) + + for _, ev := range emt.Events() { + if ev.Topic != events.TopicBGPHijack { + continue + } + body, _ := json.Marshal(ev.Payload) + require.NotContains(t, string(body), `"enrichment"`) + } +} + func TestCollector_RepeatedTickIsIdempotent(t *testing.T) { now := time.Now().UTC() ftch := &fakeFetcher{ diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/config/config.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/config/config.go index c26931d3..5e69a76d 100644 --- a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/config/config.go +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/config/config.go @@ -34,6 +34,7 @@ type CollectorsConfig struct { KEV SourceConfig `koanf:"kev"` Ransomware SourceConfig `koanf:"ransomware"` GreyNoise GreyNoiseConfig `koanf:"greynoise"` + AbuseIPDB AbuseIPDBConfig `koanf:"abuseipdb"` Coinbase CoinbaseConfig `koanf:"coinbase"` USGS SourceConfig `koanf:"usgs"` SWPC SWPCConfig `koanf:"swpc"` @@ -66,6 +67,11 @@ type GreyNoiseConfig struct { APIKey string `koanf:"api_key"` } +type AbuseIPDBConfig struct { + Enabled bool `koanf:"enabled"` + APIKey string `koanf:"api_key"` +} + type CoinbaseConfig struct { Enabled bool `koanf:"enabled"` URL string `koanf:"url"` @@ -275,6 +281,7 @@ func loadDefaults(k *koanf.Koanf) error { "collectors.ransomware.enabled": true, "collectors.ransomware.interval": "15m", "collectors.greynoise.enabled": true, + "collectors.abuseipdb.enabled": true, "collectors.coinbase.enabled": true, "collectors.coinbase.url": "wss://advanced-trade-ws.coinbase.com", "collectors.coinbase.product_ids": []string{"BTC-USD", "ETH-USD"}, @@ -328,6 +335,7 @@ var envKeyMap = map[string]string{ "NVD_API_KEY": "collectors.cve.nvd_api_key", "CF_RADAR_TOKEN": "collectors.cfradar.bearer_token", "GREYNOISE_API_KEY": "collectors.greynoise.api_key", + "ABUSEIPDB_API_KEY": "collectors.abuseipdb.api_key", } func envKeyReplacer(s string) string { diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/enrich/abuseipdb/client.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/enrich/abuseipdb/client.go new file mode 100644 index 00000000..64c3ecaf --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/enrich/abuseipdb/client.go @@ -0,0 +1,106 @@ +// ©AngelaMos | 2026 +// client.go + +package abuseipdb + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/url" + "strconv" + "time" + + "golang.org/x/time/rate" + + "github.com/carterperez-dev/monitor-the-situation/backend/internal/httpx" +) + +const ( + defaultAIDBBaseURL = "https://api.abuseipdb.com" + pathV2Check = "/api/v2/check" + abuseipdbAPIHeader = "Key" + defaultAIDBRate = 1 * time.Second + defaultAIDBBudget = 3 + defaultAIDBBreakerWin = 5 * time.Minute + defaultAIDBRequestTO = 10 * time.Second + defaultAIDBMaxAgeDays = 90 +) + +var ErrUnknownIP = errors.New("abuseipdb: unknown ip") + +type ClientConfig struct { + BaseURL string + APIKey string +} + +type Client struct { + hx *httpx.Client +} + +func NewClient(cfg ClientConfig) *Client { + if cfg.BaseURL == "" { + cfg.BaseURL = defaultAIDBBaseURL + } + return &Client{ + hx: httpx.New(httpx.Config{ + Name: "abuseipdb", + BaseURL: cfg.BaseURL, + APIKey: cfg.APIKey, + APIKeyHeader: abuseipdbAPIHeader, + Rate: rate.Every(defaultAIDBRate), + Burst: 1, + ConsecutiveFailureBudget: defaultAIDBBudget, + BreakerTimeout: defaultAIDBBreakerWin, + RequestTimeout: defaultAIDBRequestTO, + }), + } +} + +type checkEnvelope struct { + Data Verdict `json:"data"` +} + +type Verdict struct { + IPAddress string `json:"ipAddress"` + IsPublic bool `json:"isPublic"` + IPVersion int `json:"ipVersion"` + IsWhitelisted bool `json:"isWhitelisted"` + AbuseConfidenceScore int `json:"abuseConfidenceScore"` + CountryCode string `json:"countryCode"` + CountryName string `json:"countryName"` + UsageType string `json:"usageType"` + ISP string `json:"isp"` + Domain string `json:"domain"` + IsTor bool `json:"isTor"` + TotalReports int `json:"totalReports"` + NumDistinctUsers int `json:"numDistinctUsers"` + LastReportedAt string `json:"lastReportedAt"` +} + +func (c *Client) Lookup(ctx context.Context, ip string) (Verdict, error) { + q := url.Values{ + "ipAddress": []string{ip}, + "maxAgeInDays": []string{strconv.Itoa(defaultAIDBMaxAgeDays)}, + } + resp, err := c.hx.Get(ctx, pathV2Check, q) + if err != nil { + var se *httpx.StatusError + if errors.As(err, &se) && se.Code == http.StatusNotFound { + return Verdict{}, ErrUnknownIP + } + return Verdict{}, fmt.Errorf("abuseipdb lookup %s: %w", ip, err) + } + defer resp.Body.Close() + + var env checkEnvelope + if err := json.NewDecoder(resp.Body).Decode(&env); err != nil { + return Verdict{}, fmt.Errorf("abuseipdb decode: %w", err) + } + if env.Data.IPAddress == "" { + return Verdict{}, ErrUnknownIP + } + return env.Data, nil +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/enrich/abuseipdb/client_test.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/enrich/abuseipdb/client_test.go new file mode 100644 index 00000000..c3b7332d --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/enrich/abuseipdb/client_test.go @@ -0,0 +1,61 @@ +// ©AngelaMos | 2026 +// client_test.go + +package abuseipdb_test + +import ( + "context" + "net/http" + "net/http/httptest" + "os" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/carterperez-dev/monitor-the-situation/backend/internal/enrich/abuseipdb" +) + +func TestClient_LookupDecodesVerdict(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + require.Equal(t, "/api/v2/check", r.URL.Path) + require.Equal(t, "118.25.6.39", r.URL.Query().Get("ipAddress")) + require.Equal(t, "90", r.URL.Query().Get("maxAgeInDays")) + require.Equal(t, "test-key", r.Header.Get("Key")) + body, err := os.ReadFile("testdata/check_lookup.json") + require.NoError(t, err) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write(body) + })) + defer srv.Close() + + c := abuseipdb.NewClient(abuseipdb.ClientConfig{BaseURL: srv.URL, APIKey: "test-key"}) + v, err := c.Lookup(context.Background(), "118.25.6.39") + require.NoError(t, err) + require.Equal(t, "118.25.6.39", v.IPAddress) + require.Equal(t, "CN", v.CountryCode) + require.Equal(t, 100, v.AbuseConfidenceScore) + require.Equal(t, "Tencent Cloud Computing", v.ISP) +} + +func TestClient_LookupReturnsErrUnknownIPOn404(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusNotFound) + })) + defer srv.Close() + + c := abuseipdb.NewClient(abuseipdb.ClientConfig{BaseURL: srv.URL, APIKey: "k"}) + _, err := c.Lookup(context.Background(), "10.0.0.1") + require.ErrorIs(t, err, abuseipdb.ErrUnknownIP) +} + +func TestClient_LookupReturnsErrUnknownIPOnEmptyData(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"data":{}}`)) + })) + defer srv.Close() + + c := abuseipdb.NewClient(abuseipdb.ClientConfig{BaseURL: srv.URL, APIKey: "k"}) + _, err := c.Lookup(context.Background(), "10.0.0.1") + require.ErrorIs(t, err, abuseipdb.ErrUnknownIP) +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/enrich/abuseipdb/testdata/check_lookup.json b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/enrich/abuseipdb/testdata/check_lookup.json new file mode 100644 index 00000000..5d96ff02 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/enrich/abuseipdb/testdata/check_lookup.json @@ -0,0 +1,19 @@ +{ + "data": { + "ipAddress": "118.25.6.39", + "isPublic": true, + "ipVersion": 4, + "isWhitelisted": false, + "abuseConfidenceScore": 100, + "countryCode": "CN", + "countryName": "China", + "usageType": "Data Center/Web Hosting/Transit", + "isp": "Tencent Cloud Computing", + "domain": "tencent.com", + "hostnames": [], + "isTor": false, + "totalReports": 1, + "numDistinctUsers": 1, + "lastReportedAt": "2026-04-30T12:34:56+00:00" + } +}