diff --git a/server/internal/grpc/server.go b/server/internal/grpc/server.go index 207873f..dbfe266 100644 --- a/server/internal/grpc/server.go +++ b/server/internal/grpc/server.go @@ -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 { diff --git a/server/internal/services/packages.go b/server/internal/services/packages.go new file mode 100644 index 0000000..c42a292 --- /dev/null +++ b/server/internal/services/packages.go @@ -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 +}