feat(monitor/collectors/cve): 2h CVE collector (NVD pull + EPSS enrichment per tick)
This commit is contained in:
CarterPerez-dev
parent
35da1f45a6
commit
5ad1ff8e14
|
|
@ -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)
|
||||
}
|
||||
|
|
@ -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()))
|
||||
}
|
||||
Loading…
Reference in New Issue