feat: vulnerability scanning pipeline, matcher, scheduler and API
Completes tasks 10-15 and fixes what was outstanding: - vulndb.Pull implemented with oras-go, streaming the ~50MB layer and staging both files before replacing either, so a failed pull leaves the previous database intact rather than a half-written one. - db.go: Vulnerability.Severity is a string, not trivy Severity, so the int conversion did not compile. Severity now resolves vendor (highest when vendors disagree) then NVD then unknown, and CVSS is read too. - findings.go: added sweepFixedFindings plus the fleet query, severity counts, rescan flag and accept/unaccept the API needs. - vulnrules.go: added rule CRUD and the digest builder. ResolveTargets returns []models.Server, not []string, so filterByServers was wrong. - api/vulnerabilities.go was an empty file while handlers.go registered twelve routes against it; written, grouped by CVE. - shared/mail: added the missing sender. The templates were orphaned and the HTML one was a copy of the text one, defining "subject" (which html/template would escape) and emitting no markup. render.go parses every template in init(), so a bad one panics server, admin and sitesvc at boot — go build never runs init(), which is why nothing complained. - notify: digests dispatch through their own path so SMTP gets the digest template rather than arriving dressed as a monitor alert.
This commit is contained in:
@@ -0,0 +1,238 @@
|
||||
// Package vulnsched owns the vulnerability scan loop.
|
||||
//
|
||||
// It runs inside bus.RunAsLeader("housekeeping", …) alongside monitorsched,
|
||||
// workflowsched and the sweepers: one role, one lock. N replicas each running
|
||||
// this loop would mean N copies of the ~50MB database resident, N rescans of
|
||||
// the same fleet on every database refresh, and N digests reaching the
|
||||
// customer for one set of findings.
|
||||
package vulnsched
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/db"
|
||||
"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/models"
|
||||
"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/services"
|
||||
"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/vulndb"
|
||||
"go.mongodb.org/mongo-driver/v2/bson"
|
||||
"go.mongodb.org/mongo-driver/v2/mongo/options"
|
||||
)
|
||||
|
||||
const (
|
||||
tickInterval = 60 * time.Second
|
||||
// trivy-db is rebuilt every six hours; pulling more often buys nothing.
|
||||
dbMaxAge = 6 * time.Hour
|
||||
)
|
||||
|
||||
// Deps are injected from main.go rather than imported, following
|
||||
// workflowsched. It keeps this package's reach explicit and reviewable.
|
||||
type Deps struct {
|
||||
LogEvent func(instanceID, eventType, actor, serverID, keyID, details string)
|
||||
SendDigest func(instanceID string, newly []models.VulnFinding)
|
||||
}
|
||||
|
||||
type scheduler struct {
|
||||
deps Deps
|
||||
dir string
|
||||
store *vulndb.Store
|
||||
version int
|
||||
pulled time.Time
|
||||
}
|
||||
|
||||
func Start(ctx context.Context, deps Deps) {
|
||||
if vulndb.Disabled() {
|
||||
log.Println("vulnsched: disabled by VANTAGE_VULNDB_DISABLED")
|
||||
return
|
||||
}
|
||||
|
||||
dir, err := os.MkdirTemp("", "vantage-vulndb-")
|
||||
if err != nil {
|
||||
log.Printf("vulnsched: temp dir: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
s := &scheduler{deps: deps, dir: dir}
|
||||
|
||||
go func() {
|
||||
defer os.RemoveAll(dir)
|
||||
defer s.closeStore()
|
||||
|
||||
ticker := time.NewTicker(tickInterval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
s.tick(ctx)
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
func (s *scheduler) tick(ctx context.Context) {
|
||||
if err := s.ensureDB(ctx); err != nil {
|
||||
// Keep the last good database and carry on scanning against it. A
|
||||
// network blip must never clear findings or read as "all fixed".
|
||||
log.Printf("vulnsched: database unavailable: %v", err)
|
||||
s.recordDBError(ctx, err)
|
||||
if s.store == nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
s.scanPending(ctx)
|
||||
}
|
||||
|
||||
// ensureDB pulls a fresh database when the local copy is stale, and marks the
|
||||
// whole fleet for rescanning when the version changes — which is what makes a
|
||||
// newly published CVE flag existing servers within a minute rather than at the
|
||||
// next agent report.
|
||||
func (s *scheduler) ensureDB(ctx context.Context) error {
|
||||
if s.store != nil && time.Since(s.pulled) < dbMaxAge {
|
||||
return nil
|
||||
}
|
||||
|
||||
version, err := vulndb.Pull(ctx, s.dir)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
s.closeStore()
|
||||
store, err := vulndb.Open(s.dir)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
s.store = store
|
||||
s.pulled = time.Now()
|
||||
|
||||
changed := version != s.version
|
||||
s.version = version
|
||||
|
||||
_, _ = db.Col("vulndb_meta").UpdateOne(ctx, bson.M{},
|
||||
bson.M{"$set": bson.M{"db_version": version, "pulled_at": s.pulled}, "$unset": bson.M{"last_error": ""}},
|
||||
options.UpdateOne().SetUpsert(true),
|
||||
)
|
||||
|
||||
if changed {
|
||||
res, err := db.Col("server_packages").UpdateMany(ctx,
|
||||
bson.M{"status": bson.M{"$ne": models.ScanStatusUnsupported}},
|
||||
bson.M{"$set": bson.M{"scan_pending": true}},
|
||||
)
|
||||
if err != nil {
|
||||
log.Printf("vulnsched: mark fleet pending: %v", err)
|
||||
} else {
|
||||
log.Printf("vulnsched: database version %d, %d servers marked for rescan", version, res.ModifiedCount)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *scheduler) recordDBError(ctx context.Context, err error) {
|
||||
_, _ = db.Col("vulndb_meta").UpdateOne(ctx, bson.M{},
|
||||
bson.M{"$set": bson.M{"last_error": err.Error()}},
|
||||
options.UpdateOne().SetUpsert(true),
|
||||
)
|
||||
}
|
||||
|
||||
func (s *scheduler) scanPending(ctx context.Context) {
|
||||
cur, err := db.Col("server_packages").Find(ctx, bson.M{"scan_pending": true})
|
||||
if err != nil {
|
||||
log.Printf("vulnsched: find pending: %v", err)
|
||||
return
|
||||
}
|
||||
defer cur.Close(ctx)
|
||||
|
||||
var pending []models.ServerPackages
|
||||
if err := cur.All(ctx, &pending); err != nil {
|
||||
log.Printf("vulnsched: decode pending: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
// Newly opened findings are collected across the whole tick and sent as one
|
||||
// digest per instance. A database refresh can open several hundred findings
|
||||
// at once; one message per finding would rate-limit the webhook or get the
|
||||
// channel muted, and either way the alerts stop being read.
|
||||
newly := map[string][]models.VulnFinding{}
|
||||
|
||||
for _, sp := range pending {
|
||||
if ctx.Err() != nil {
|
||||
// Leadership lost. scan_pending is still set, so the next leader
|
||||
// picks these up — which is why it lives on the document.
|
||||
return
|
||||
}
|
||||
opened := s.scanOne(ctx, sp)
|
||||
newly[sp.InstanceID] = append(newly[sp.InstanceID], opened...)
|
||||
}
|
||||
|
||||
for instanceID, findings := range newly {
|
||||
if len(findings) > 0 && s.deps.SendDigest != nil {
|
||||
s.deps.SendDigest(instanceID, findings)
|
||||
}
|
||||
}
|
||||
|
||||
_, _ = db.Col("vulndb_meta").UpdateOne(ctx, bson.M{},
|
||||
bson.M{"$set": bson.M{"last_full_scan_at": time.Now()}},
|
||||
options.UpdateOne().SetUpsert(true),
|
||||
)
|
||||
}
|
||||
|
||||
func (s *scheduler) scanOne(ctx context.Context, sp models.ServerPackages) []models.VulnFinding {
|
||||
now := time.Now()
|
||||
|
||||
results, err := vulndb.Match(s.store, sp.OS, sp.Packages)
|
||||
if err != nil {
|
||||
// We hold no feed for this distribution, so we cannot answer whether it
|
||||
// is vulnerable. Say "unsupported" — reporting zero findings here would
|
||||
// be indistinguishable from reporting a clean host, and one of those is
|
||||
// a lie.
|
||||
status := models.ScanStatusUnsupported
|
||||
if !errors.Is(err, vulndb.ErrUnsupportedFamily) {
|
||||
log.Printf("vulnsched: scan %s: %v", sp.ServerID, err)
|
||||
status = sp.Status
|
||||
}
|
||||
s.clearPending(ctx, sp.ID, status, now)
|
||||
return nil
|
||||
}
|
||||
|
||||
existing, err := services.ListFindings(ctx, sp.InstanceID, sp.ServerID)
|
||||
if err != nil {
|
||||
log.Printf("vulnsched: list findings %s: %v", sp.ServerID, err)
|
||||
return nil
|
||||
}
|
||||
|
||||
diff := services.DiffFindings(existing, results, now)
|
||||
if err := services.ApplyFindingDiff(ctx, sp.InstanceID, sp.ServerID, diff, now); err != nil {
|
||||
log.Printf("vulnsched: apply diff %s: %v", sp.ServerID, err)
|
||||
return nil
|
||||
}
|
||||
|
||||
s.clearPending(ctx, sp.ID, models.ScanStatusOK, now)
|
||||
|
||||
for i := range diff.NewlyOpened {
|
||||
diff.NewlyOpened[i].ServerID = sp.ServerID
|
||||
}
|
||||
return diff.NewlyOpened
|
||||
}
|
||||
|
||||
func (s *scheduler) clearPending(ctx context.Context, id bson.ObjectID, status string, now time.Time) {
|
||||
_, _ = db.Col("server_packages").UpdateOne(ctx,
|
||||
bson.M{"_id": id},
|
||||
bson.M{"$set": bson.M{
|
||||
"scan_pending": false,
|
||||
"status": status,
|
||||
"scanned_at": now,
|
||||
"db_version": s.version,
|
||||
}},
|
||||
)
|
||||
}
|
||||
|
||||
func (s *scheduler) closeStore() {
|
||||
if s.store != nil {
|
||||
_ = s.store.Close()
|
||||
s.store = nil
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user