From 5ad1ff8e1432aaef1d12118b7c7b88941a9ae2dd Mon Sep 17 00:00:00 2001 From: CarterPerez-dev Date: Fri, 1 May 2026 22:41:24 -0400 Subject: [PATCH] feat(monitor/collectors/cve): 2h CVE collector (NVD pull + EPSS enrichment per tick) --- .../internal/collectors/cve/collector.go | 146 +++++++++++++++ .../internal/collectors/cve/collector_test.go | 172 ++++++++++++++++++ 2 files changed, 318 insertions(+) create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/cve/collector.go create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/cve/collector_test.go diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/cve/collector.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/cve/collector.go new file mode 100644 index 00000000..b90b2eb0 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/cve/collector.go @@ -0,0 +1,146 @@ +// ©AngelaMos | 2026 +// collector.go + +package cve + +import ( + "context" + "encoding/json" + "log/slog" + "time" + + "github.com/carterperez-dev/monitor-the-situation/backend/internal/events" +) + +const ( + Name = "cve" + defaultCVEInterval = 2 * time.Hour + defaultCVEWindow = 2 * time.Hour +) + +type NVDFetcher interface { + Fetch(ctx context.Context, start, end time.Time) (NVDResponse, error) +} + +type EPSSFetcher interface { + LookupBatch(ctx context.Context, cveIDs []string) (map[string]EPSSScore, error) +} + +type Repository interface { + Upsert(ctx context.Context, row Row) error + UpdateEPSS(ctx context.Context, cveID string, score, percentile float64) error +} + +type Emitter interface { + Emit(ev events.Event) +} + +type StateRecorder interface { + RecordSuccess(ctx context.Context, name string, eventCount int64) error + RecordError(ctx context.Context, name, errMsg string) error +} + +type CollectorConfig struct { + Interval time.Duration + Window time.Duration + NVD NVDFetcher + EPSS EPSSFetcher + Repo Repository + Emitter Emitter + State StateRecorder + Logger *slog.Logger +} + +type Collector struct { + cfg CollectorConfig + logger *slog.Logger +} + +func NewCollector(cfg CollectorConfig) *Collector { + if cfg.Interval <= 0 { + cfg.Interval = defaultCVEInterval + } + if cfg.Window <= 0 { + cfg.Window = defaultCVEWindow + } + if cfg.Logger == nil { + cfg.Logger = slog.Default() + } + return &Collector{cfg: cfg, logger: cfg.Logger} +} + +func (c *Collector) Name() string { return Name } + +func (c *Collector) Run(ctx context.Context) error { + ticker := time.NewTicker(c.cfg.Interval) + defer ticker.Stop() + + c.tick(ctx) + + for { + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + c.tick(ctx) + } + } +} + +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 + } + + 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) + row := Row{ + CveID: v.CVE.ID, + Published: v.CVE.Published.Time, + LastModified: v.CVE.LastModified.Time, + Severity: severity, + CVSS: score, + Payload: raw, + } + rows = append(rows, row) + ids = append(ids, v.CVE.ID) + } + + scores, err := c.cfg.EPSS.LookupBatch(ctx, ids) + if err != nil { + c.logger.Warn("epss lookup", "err", err) + } + + emitted := int64(0) + for _, row := range rows { + if err := c.cfg.Repo.Upsert(ctx, row); err != nil { + c.logger.Warn("upsert cve", "id", row.CveID, "err", err) + continue + } + if s, ok := scores[row.CveID]; ok { + if err := c.cfg.Repo.UpdateEPSS(ctx, row.CveID, s.Score, s.Percentile); err != nil { + c.logger.Warn("update epss", "id", row.CveID, "err", err) + } + 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), + }) + emitted++ + } + _ = c.cfg.State.RecordSuccess(ctx, Name, emitted) +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/cve/collector_test.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/cve/collector_test.go new file mode 100644 index 00000000..d7eea8e1 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/cve/collector_test.go @@ -0,0 +1,172 @@ +// ©AngelaMos | 2026 +// collector_test.go + +package cve_test + +import ( + "context" + "encoding/json" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/carterperez-dev/monitor-the-situation/backend/internal/collectors/cve" + "github.com/carterperez-dev/monitor-the-situation/backend/internal/events" +) + +type stubNVD struct { + resp cve.NVDResponse + err error +} + +func (s *stubNVD) Fetch(context.Context, time.Time, time.Time) (cve.NVDResponse, error) { + return s.resp, s.err +} + +type stubEPSS struct { + scores map[string]cve.EPSSScore + err error +} + +func (s *stubEPSS) LookupBatch(context.Context, []string) (map[string]cve.EPSSScore, error) { + return s.scores, s.err +} + +type stubCVERepo struct { + mu sync.Mutex + upserts map[string]cve.Row + patches map[string][2]float64 +} + +func newStubCVERepo() *stubCVERepo { + return &stubCVERepo{ + upserts: map[string]cve.Row{}, + patches: map[string][2]float64{}, + } +} + +func (r *stubCVERepo) Upsert(_ context.Context, row cve.Row) error { + r.mu.Lock() + defer r.mu.Unlock() + r.upserts[row.CveID] = row + return nil +} + +func (r *stubCVERepo) UpdateEPSS(_ context.Context, id string, score, pct float64) error { + r.mu.Lock() + defer r.mu.Unlock() + r.patches[id] = [2]float64{score, pct} + return nil +} + +func (r *stubCVERepo) Upserts() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.upserts) +} + +func (r *stubCVERepo) Patches() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.patches) +} + +type stubCVEEmitter struct { + mu sync.Mutex + events []events.Event +} + +func (e *stubCVEEmitter) Emit(ev events.Event) { + e.mu.Lock() + defer e.mu.Unlock() + e.events = append(e.events, ev) +} + +func (e *stubCVEEmitter) Events() []events.Event { + e.mu.Lock() + defer e.mu.Unlock() + out := make([]events.Event, len(e.events)) + copy(out, e.events) + return out +} + +type stubCVEState struct{} + +func (stubCVEState) RecordSuccess(context.Context, string, int64) error { return nil } +func (stubCVEState) RecordError(context.Context, string, string) error { return nil } + +func TestCVECollector_FetchesEnrichesAndEmits(t *testing.T) { + now := time.Now().UTC() + nvd := &stubNVD{resp: cve.NVDResponse{ + Vulnerabilities: []cve.NVDVulnRoot{ + {CVE: cve.NVDCVE{ + ID: "CVE-2026-0001", Published: cve.NVDTime{Time: now.Add(-time.Hour)}, + LastModified: cve.NVDTime{Time: now}, + Metrics: cve.NVDMetrics{CVSSv31: []cve.NVDMetricEntry{ + {CVSSData: cve.NVDCVSSData{BaseScore: 9.8, BaseSeverity: "CRITICAL"}}, + }}, + }}, + {CVE: cve.NVDCVE{ + ID: "CVE-2026-0002", Published: cve.NVDTime{Time: now.Add(-time.Hour)}, + LastModified: cve.NVDTime{Time: now}, + }}, + }, + }} + epss := &stubEPSS{scores: map[string]cve.EPSSScore{ + "CVE-2026-0001": {CveID: "CVE-2026-0001", Score: 0.9, Percentile: 0.99}, + "CVE-2026-0002": {CveID: "CVE-2026-0002", Score: 0.1, Percentile: 0.5}, + }} + repo := newStubCVERepo() + emt := &stubCVEEmitter{} + + c := cve.NewCollector(cve.CollectorConfig{ + Interval: 30 * time.Millisecond, + NVD: nvd, + EPSS: epss, + Repo: repo, + Emitter: emt, + State: stubCVEState{}, + }) + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + _ = c.Run(ctx) + + require.GreaterOrEqual(t, repo.Upserts(), 2) + require.GreaterOrEqual(t, repo.Patches(), 2) + + evs := emt.Events() + require.GreaterOrEqual(t, len(evs), 2) + for _, ev := range evs { + require.Equal(t, events.TopicCVENew, ev.Topic) + require.Equal(t, cve.Name, ev.Source) + raw, ok := ev.Payload.(json.RawMessage) + require.True(t, ok) + require.Contains(t, string(raw), "CVE-2026-") + } +} + +func TestCVECollector_NVDFetchErrorRecordsState(t *testing.T) { + nvd := &stubNVD{err: context.DeadlineExceeded} + repo := newStubCVERepo() + emt := &stubCVEEmitter{} + + c := cve.NewCollector(cve.CollectorConfig{ + Interval: 20 * time.Millisecond, + NVD: nvd, + EPSS: &stubEPSS{}, + Repo: repo, + Emitter: emt, + State: stubCVEState{}, + }) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond) + defer cancel() + _ = c.Run(ctx) + + require.Zero(t, repo.Upserts()) + require.Zero(t, repo.Patches()) + require.Zero(t, len(emt.Events())) +}