feat: store agent package reports and serve the collect flag

VulnScanningEnabled reads GetLicenseState(...).Feature("vuln_scanning") and requires an active licence, never switching on tier. ReportPackages re-checks it server-side: the agent flag is the optimisation, this is the boundary.
2026-08-06 13:19:39 +01:00
parent a92c3190c2
commit 583f60771c
2 changed files with 182 additions and 1 deletions
@@ -63,7 +63,13 @@ func (s *vantageServer) SyncKeys(ctx context.Context, req *pb.SyncRequest) (*pb.
 		return nil, status.Errorf(codes.Internal, "failed to build authorized keys: %v", err)
 	}

-	return &pb.SyncResponse{PublicKeys: keys}, nil
+	// Carried on the 30s key poll rather than its own RPC: the agent needs it
+	// before its hourly package report, and this is the only message it already
+	// receives that often.
+	return &pb.SyncResponse{
+		PublicKeys:      keys,
+		CollectPackages: services.VulnScanningEnabled(srv.InstanceID),
+	}, nil
 }

 func (s *vantageServer) UploadGeneratedKey(ctx context.Context, req *pb.UploadKeyRequest) (*pb.UploadKeyResponse, error) {
@@ -104,6 +110,59 @@ func (s *vantageServer) ReportUpdates(ctx context.Context, req *pb.ReportUpdates
 	return &pb.ReportUpdatesResponse{}, nil
 }

+// ReportPackages stores a server's installed package set.
+//
+// It does NOT match against the vulnerability database. Matching happens in
+// vulnsched, on the leader: every replica would otherwise need the ~50MB
+// database resident, and a database refresh would have N replicas racing to
+// rescan the same fleet and sending N digests to the customer.
+func (s *vantageServer) ReportPackages(ctx context.Context, req *pb.ReportPackagesRequest) (*pb.ReportPackagesResponse, error) {
+	srv, err := services.ValidateAgentToken(req.ServerId, req.AgentToken)
+	if err != nil {
+		return nil, status.Errorf(codes.Unauthenticated, "invalid agent token")
+	}
+
+	// The agent's collect_packages flag is an optimisation; this is the
+	// boundary. An agent that ignores the flag still stores nothing.
+	if !services.VulnScanningEnabled(srv.InstanceID) {
+		return &pb.ReportPackagesResponse{NeedFull: false}, nil
+	}
+
+	// The offer call: a hash and no packages. Answering NeedFull=false here is
+	// what saves the ~150KB body on the overwhelming majority of reports.
+	if len(req.Packages) == 0 {
+		known, err := services.HasPackageHash(srv.InstanceID, srv.ServerID, req.Hash)
+		if err != nil {
+			log.Printf("package hash lookup for %s: %v", srv.ServerID, err)
+			return nil, status.Errorf(codes.Internal, "package hash lookup failed")
+		}
+		return &pb.ReportPackagesResponse{NeedFull: !known}, nil
+	}
+
+	pkgs := make([]models.InstalledPackage, len(req.Packages))
+	for i, p := range req.Packages {
+		pkgs[i] = models.InstalledPackage{
+			Name:       p.Name,
+			Version:    p.Version,
+			Epoch:      int(p.Epoch),
+			Arch:       p.Arch,
+			SourceName: p.SourceName,
+		}
+	}
+
+	os := models.OSRelease{
+		Family:    req.Os.Family,
+		VersionID: req.Os.VersionId,
+		Arch:      req.Os.Arch,
+	}
+
+	if err := services.StorePackages(srv.InstanceID, srv.ServerID, os, req.Hash, pkgs); err != nil {
+		log.Printf("store packages for %s: %v", srv.ServerID, err)
+		return nil, status.Errorf(codes.Internal, "failed to store packages")
+	}
+	return &pb.ReportPackagesResponse{NeedFull: false}, nil
+}
+
 func (s *vantageServer) ReportInventory(ctx context.Context, req *pb.InventoryReport) (*pb.InventoryReportResponse, error) {
 	srv, err := services.ValidateAgentToken(req.ServerId, req.AgentToken)
 	if err != nil {
@@ -0,0 +1,122 @@
+package services
+
+import (
+	"context"
+	"time"
+
+	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/db"
+	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/models"
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+	"go.mongodb.org/mongo-driver/v2/mongo/options"
+)
+
+// FeatureVulnScanning is the licence feature name gating package collection.
+const FeatureVulnScanning = "vuln_scanning"
+
+// VulnScanningEnabled reports whether this instance may collect packages.
+//
+// It reads the feature by name and never switches on tier, so changing what a
+// tier includes needs no server release. A lapsed licence collects nothing:
+// there is no point accumulating inventory an instance cannot act on.
+func VulnScanningEnabled(instanceID string) bool {
+	st := GetLicenseState(instanceID)
+	return st.Active() && st.Feature(FeatureVulnScanning)
+}
+
+// HasPackageHash reports whether we already hold this exact package set, which
+// is what lets the agent skip sending ~150KB it has already sent.
+func HasPackageHash(instanceID, serverID, hash string) (bool, error) {
+	err := db.Col("server_packages").FindOne(context.Background(), bson.M{
+		"instance_id": instanceID,
+		"server_id":   serverID,
+		"hash":        hash,
+	}, options.FindOne().SetProjection(bson.M{"_id": 1})).Err()
+	if err == mongo.ErrNoDocuments {
+		return false, nil
+	}
+	if err != nil {
+		return false, err
+	}
+	return true, nil
+}
+
+// StorePackages replaces a server's package set and marks it for scanning.
+//
+// It deliberately does NOT match against the vulnerability database. Matching
+// happens in vulnsched, on the leader, for two reasons: every replica would
+// otherwise need the ~50MB database resident, and a database refresh would have
+// N replicas racing to rescan the same fleet and sending N digests.
+func StorePackages(instanceID, serverID string, os models.OSRelease, hash string, pkgs []models.InstalledPackage) error {
+	now := time.Now()
+	_, err := db.Col("server_packages").UpdateOne(context.Background(),
+		bson.M{"instance_id": instanceID, "server_id": serverID},
+		bson.M{"$set": bson.M{
+			"os":           os,
+			"hash":         hash,
+			"packages":     pkgs,
+			"collected_at": now,
+			"scan_pending": true,
+		}},
+		options.UpdateOne().SetUpsert(true),
+	)
+	return err
+}
+
+// ListPackages returns a server's stored package set, or nil when the agent has
+// not reported yet. A missing document is not an error: an agent that has never
+// reported is the normal state for the first hour after install.
+func ListPackages(instanceID, serverID string) (*models.ServerPackages, error) {
+	var sp models.ServerPackages
+	err := db.Col("server_packages").FindOne(context.Background(), bson.M{
+		"instance_id": instanceID,
+		"server_id":   serverID,
+	}).Decode(&sp)
+	if err == mongo.ErrNoDocuments {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	return &sp, nil
+}
+
+// PackageHit is one server running one package.
+type PackageHit struct {
+	ServerID string `json:"server_id"`
+	Name     string `json:"name"`
+	Version  string `json:"version"`
+}
+
+// SearchPackages answers "which servers run package X" across the fleet — the
+// question people actually ask during an incident.
+//
+// The Mongo filter narrows to documents containing the name; the second pass is
+// needed because a multikey match returns the whole document, not the matching
+// array element.
+func SearchPackages(instanceID, name string) ([]PackageHit, error) {
+	ctx := context.Background()
+	cur, err := db.Col("server_packages").Find(ctx, bson.M{
+		"instance_id":   instanceID,
+		"packages.name": name,
+	}, options.Find().SetProjection(bson.M{"server_id": 1, "packages": 1}))
+	if err != nil {
+		return nil, err
+	}
+	defer cur.Close(ctx)
+
+	var docs []models.ServerPackages
+	if err := cur.All(ctx, &docs); err != nil {
+		return nil, err
+	}
+
+	hits := []PackageHit{}
+	for _, d := range docs {
+		for _, p := range d.Packages {
+			if p.Name == name {
+				hits = append(hits, PackageHit{ServerID: d.ServerID, Name: p.Name, Version: p.Version})
+			}
+		}
+	}
+	return hits, nil
+}