Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
114 changes: 80 additions & 34 deletions internal/security/scanner/service.go
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ type Storage interface {
SaveScanJob(job *ScanJob) error
GetScanJob(id string) (*ScanJob, error)
ListScanJobs(serverName string) ([]*ScanJob, error)
ListScanJobMetas(serverName string) ([]*ScanJobMeta, error)
GetLatestScanJob(serverName string) (*ScanJob, error)
DeleteScanJob(id string) error
DeleteServerScanJobs(serverName string) error
Expand Down Expand Up @@ -1083,25 +1084,23 @@ func (s *Service) GetScanReportByJobID(ctx context.Context, jobID string) (*Aggr
agg.Pass1Complete = job.ScanPass == ScanPassSecurityScan && job.Status == ScanJobStatusCompleted
agg.Pass2Complete = job.ScanPass == ScanPassSupplyChainAudit && job.Status == ScanJobStatusCompleted

// If this is a Pass 1 job, try to find and merge companion Pass 2 results
// If this is a Pass 1 job, try to find and merge companion Pass 2 results.
// The companion is resolved via the lightweight scan-job index, so this does
// NOT deserialize the full per-server scan history (MCP-2205).
if job.ScanPass == ScanPassSecurityScan || job.ScanPass == 0 {
allJobs, _ := s.storage.ListScanJobs(job.ServerName)
for _, j := range allJobs {
if j.ScanPass == ScanPassSupplyChainAudit && j.Status == ScanJobStatusCompleted && j.StartedAt.After(job.StartedAt) {
pass2Reports, err := s.storage.ListScanReportsByJob(j.ID)
if err == nil {
for _, r := range pass2Reports {
for i := range r.Findings {
r.Findings[i].ScanPass = ScanPassSupplyChainAudit
}
if companionID := s.findCompanionPass2JobID(job); companionID != "" {
pass2Reports, err := s.storage.ListScanReportsByJob(companionID)
if err == nil {
for _, r := range pass2Reports {
for i := range r.Findings {
r.Findings[i].ScanPass = ScanPassSupplyChainAudit
}
allMerged := append(reports, pass2Reports...)
allMerged = deduplicatePass2Findings(allMerged)
agg = AggregateReportsWithJobStatus(job.ID, job.ServerName, allMerged, job)
agg.Pass1Complete = true
agg.Pass2Complete = true
}
break
allMerged := append(reports, pass2Reports...)
allMerged = deduplicatePass2Findings(allMerged)
agg = AggregateReportsWithJobStatus(job.ID, job.ServerName, allMerged, job)
agg.Pass1Complete = true
agg.Pass2Complete = true
}
}

Expand All @@ -1118,6 +1117,40 @@ func (s *Service) GetScanReportByJobID(ctx context.Context, jobID string) (*Aggr
return agg, nil
}

// findCompanionPass2JobID returns the ID of the Pass-2 (supply-chain audit) job
// that companions the given Pass-1 job: the earliest completed Pass-2 job that
// started after it. It reads the lightweight scan-job metadata index rather than
// the full job records, so its cost is independent of scan-output size and the
// report path no longer slows down as a server accrues scan history (MCP-2205).
// Returns "" when no companion exists.
func (s *Service) findCompanionPass2JobID(pass1 *ScanJob) string {
metas, err := s.storage.ListScanJobMetas(pass1.ServerName)
if err != nil {
s.logger.Warn("failed to list scan job metadata for companion lookup",
zap.String("server", pass1.ServerName),
zap.Error(err),
)
return ""
}

var best *ScanJobMeta
for _, m := range metas {
if m.ScanPass != ScanPassSupplyChainAudit || m.Status != ScanJobStatusCompleted {
continue
}
if !m.StartedAt.After(pass1.StartedAt) {
continue
}
if best == nil || m.StartedAt.Before(best.StartedAt) {
best = m
}
}
if best == nil {
return ""
}
return best.ID
}

// deduplicatePass2Findings removes Pass 2 findings that duplicate Pass 1 findings.
// The dedup key is scanner + rule_id + title (not location, since paths may differ between passes).
func deduplicatePass2Findings(reports []*ScanReport) []*ScanReport {
Expand Down Expand Up @@ -1158,38 +1191,51 @@ func deduplicatePass2Findings(reports []*ScanReport) []*ScanReport {
// findLatestPassJobs finds the latest Pass 1 and Pass 2 jobs for a server.
// Returns (pass1Job, pass2Job, error). At least one must be non-nil on success.
func (s *Service) findLatestPassJobs(serverName string) (*ScanJob, *ScanJob, error) {
jobs, err := s.storage.ListScanJobs(serverName)
// Read lightweight metadata rather than full job records so this scales with
// neither scan-output size nor history depth: we deserialize at most the two
// jobs we actually return (MCP-2205).
metas, err := s.storage.ListScanJobMetas(serverName)
if err != nil {
// Surface the underlying I/O error so the caller can distinguish
// transient failures from "no records found".
return nil, nil, fmt.Errorf("list scan jobs for %s: %w", serverName, err)
return nil, nil, fmt.Errorf("list scan job metadata for %s: %w", serverName, err)
}
if len(jobs) == 0 {
if len(metas) == 0 {
return nil, nil, fmt.Errorf("%w: %s", errNoScans, serverName)
}

// Sort by start time descending (newest first)
sort.Slice(jobs, func(i, j int) bool {
return jobs[i].StartedAt.After(jobs[j].StartedAt)
})

var pass1Job, pass2Job *ScanJob
for _, j := range jobs {
if j.ScanPass == ScanPassSupplyChainAudit && pass2Job == nil {
pass2Job = j
} else if (j.ScanPass == ScanPassSecurityScan || j.ScanPass == 0) && pass1Job == nil {
// Pick the newest Pass-1 and Pass-2 job IDs by start time.
var pass1Meta, pass2Meta *ScanJobMeta
for _, m := range metas {
switch m.ScanPass {
case ScanPassSupplyChainAudit:
if pass2Meta == nil || m.StartedAt.After(pass2Meta.StartedAt) {
pass2Meta = m
}
case ScanPassSecurityScan, 0:
// ScanPass == 0 handles legacy jobs (before two-pass was added)
pass1Job = j
}
if pass1Job != nil && pass2Job != nil {
break
if pass1Meta == nil || m.StartedAt.After(pass1Meta.StartedAt) {
pass1Meta = m
}
}
}

if pass1Job == nil && pass2Job == nil {
if pass1Meta == nil && pass2Meta == nil {
return nil, nil, fmt.Errorf("%w: %s", errNoScans, serverName)
}

var pass1Job, pass2Job *ScanJob
if pass1Meta != nil {
if pass1Job, err = s.storage.GetScanJob(pass1Meta.ID); err != nil {
return nil, nil, fmt.Errorf("load latest pass-1 job %s: %w", pass1Meta.ID, err)
}
}
if pass2Meta != nil {
if pass2Job, err = s.storage.GetScanJob(pass2Meta.ID); err != nil {
return nil, nil, fmt.Errorf("load latest pass-2 job %s: %w", pass2Meta.ID, err)
}
}

return pass1Job, pass2Job, nil
}

Expand Down
140 changes: 132 additions & 8 deletions internal/security/scanner/service_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,26 @@ func (m *mockStorage) ListScanJobs(serverName string) ([]*ScanJob, error) {
return result, nil
}

func (m *mockStorage) ListScanJobMetas(serverName string) ([]*ScanJobMeta, error) {
m.mu.Lock()
defer m.mu.Unlock()
var result []*ScanJobMeta
for _, j := range m.jobs {
if serverName != "" && j.ServerName != serverName {
continue
}
result = append(result, &ScanJobMeta{
ID: j.ID,
ServerName: j.ServerName,
Status: j.Status,
ScanPass: j.ScanPass,
StartedAt: j.StartedAt,
CompletedAt: j.CompletedAt,
})
}
return result, nil
}

func (m *mockStorage) GetLatestScanJob(serverName string) (*ScanJob, error) {
m.mu.Lock()
defer m.mu.Unlock()
Expand Down Expand Up @@ -1545,12 +1565,14 @@ func TestScanFindingScanPassTag(t *testing.T) {
}
}

// countingStorage wraps any Storage and counts ListScanJobs invocations.
// Used for verifying the GetScanSummary negative-cache behavior introduced in
// spec 047.
// countingStorage wraps any Storage and counts storage probes. listCalls counts
// the heavy full-history ListScanJobs path (MCP-2205 asserts this stays 0 on the
// report hot paths); metaCalls counts the lightweight ListScanJobMetas index
// path used by the GetScanSummary negative-cache behavior (spec 047).
type countingStorage struct {
Storage
listCalls atomic.Int64
metaCalls atomic.Int64
}

func newCountingStorage(inner Storage) *countingStorage {
Expand All @@ -1562,6 +1584,11 @@ func (c *countingStorage) ListScanJobs(serverName string) ([]*ScanJob, error) {
return c.Storage.ListScanJobs(serverName)
}

func (c *countingStorage) ListScanJobMetas(serverName string) ([]*ScanJobMeta, error) {
c.metaCalls.Add(1)
return c.Storage.ListScanJobMetas(serverName)
}

// erroringStorage returns a transient error from ListScanJobs while delegating
// other calls to the inner Storage. Used to verify that non-errNoScans errors
// do NOT populate the negative cache.
Expand All @@ -1576,6 +1603,11 @@ func (e *erroringStorage) ListScanJobs(string) ([]*ScanJob, error) {
return nil, e.err
}

func (e *erroringStorage) ListScanJobMetas(string) ([]*ScanJobMeta, error) {
e.listCalls.Add(1)
return nil, e.err
}

// Spec 047 — Phase 3 (US1): cache the "no scans found" sentinel.

func TestGetScanSummary_CachesNegativeResult(t *testing.T) {
Expand All @@ -1590,8 +1622,8 @@ func TestGetScanSummary_CachesNegativeResult(t *testing.T) {
}

// Without the negative-cache fix, this would be N storage calls.
if got := store.listCalls.Load(); got != 1 {
t.Errorf("expected exactly 1 ListScanJobs call after %d GetScanSummary invocations, got %d", N, got)
if got := store.metaCalls.Load(); got != 1 {
t.Errorf("expected exactly 1 ListScanJobMetas call after %d GetScanSummary invocations, got %d", N, got)
}
}

Expand All @@ -1606,7 +1638,7 @@ func TestGetScanSummary_DoesNotCacheOnTransientError(t *testing.T) {

// Transient error must NOT populate the negative cache: every call retries.
if got := store.listCalls.Load(); got != int64(N) {
t.Errorf("expected %d ListScanJobs calls (no caching of transient errors), got %d", N, got)
t.Errorf("expected %d storage probes (no caching of transient errors), got %d", N, got)
}
}

Expand All @@ -1619,8 +1651,8 @@ func TestGetScanSummary_OverwritesNilSentinelOnRealScan(t *testing.T) {
if got := svc.GetScanSummary(context.Background(), "later-scanned"); got != nil {
t.Fatalf("expected nil summary on first call, got %+v", got)
}
if got := store.listCalls.Load(); got != 1 {
t.Fatalf("expected 1 ListScanJobs call after first GetScanSummary, got %d", got)
if got := store.metaCalls.Load(); got != 1 {
t.Fatalf("expected 1 ListScanJobMetas call after first GetScanSummary, got %d", got)
}

// Simulate a real scan landing for that server: insert a completed Pass-1 job
Expand All @@ -1646,3 +1678,95 @@ func TestGetScanSummary_OverwritesNilSentinelOnRealScan(t *testing.T) {
t.Errorf("expected real summary {Status: clean}, got %+v", got)
}
}

// TestGetScanReportByJobID_LatencyIndependentOfHistory verifies the MCP-2205
// fix: aggregating a Pass-1 report must NOT deserialize the full per-server scan
// history (ListScanJobs), whose job payloads carry large stdout/stderr. The
// companion Pass-2 lookup uses the lightweight metadata index instead, so report
// latency does not grow with how many times a server has been scanned.
func TestGetScanReportByJobID_LatencyIndependentOfHistory(t *testing.T) {
mock := newMockStorage()
store := newCountingStorage(mock)
svc := NewService(store, NewRegistry(t.TempDir(), zap.NewNop()), nil, t.TempDir(), zap.NewNop())

now := time.Now()

// Pass-1 job under inspection + its report.
pass1 := &ScanJob{ID: "p1", ServerName: "srv", Status: ScanJobStatusCompleted, ScanPass: ScanPassSecurityScan, StartedAt: now.Add(-10 * time.Minute)}
_ = mock.SaveScanJob(pass1)
_ = mock.SaveScanReport(&ScanReport{ID: "r1", JobID: "p1", ServerName: "srv", ScannerID: "s",
Findings: []ScanFinding{{RuleID: "T1", Title: "tool poisoning", Scanner: "s", ThreatLevel: ThreatLevelDangerous}}})

// Companion Pass-2 job that started after Pass-1 + its report.
pass2 := &ScanJob{ID: "p2", ServerName: "srv", Status: ScanJobStatusCompleted, ScanPass: ScanPassSupplyChainAudit, StartedAt: now.Add(-8 * time.Minute)}
_ = mock.SaveScanJob(pass2)
_ = mock.SaveScanReport(&ScanReport{ID: "r2", JobID: "p2", ServerName: "srv", ScannerID: "s",
Findings: []ScanFinding{{RuleID: "CVE-1", Title: "known cve", Scanner: "s", ThreatLevel: ThreatLevelWarning}}})

// Large historical backlog for the same server (the symptom: reports get
// slower the more a server is scanned).
for i := 0; i < 100; i++ {
_ = mock.SaveScanJob(&ScanJob{
ID: fmt.Sprintf("noise-%d", i),
ServerName: "srv",
Status: ScanJobStatusCompleted,
ScanPass: ScanPassSecurityScan,
StartedAt: now.Add(time.Duration(-20-i) * time.Minute),
})
}

agg, err := svc.GetScanReportByJobID(context.Background(), "p1")
if err != nil {
t.Fatalf("GetScanReportByJobID: %v", err)
}

// Correctness: companion Pass-2 still merged in.
if !agg.Pass1Complete || !agg.Pass2Complete {
t.Errorf("expected both passes complete, got pass1=%v pass2=%v", agg.Pass1Complete, agg.Pass2Complete)
}
if len(agg.Findings) != 2 {
t.Fatalf("expected 2 merged findings (Pass-1 + companion Pass-2), got %d", len(agg.Findings))
}

// Latency guard: must not scan the full per-server job history.
if got := store.listCalls.Load(); got != 0 {
t.Errorf("expected 0 ListScanJobs (full-history) calls in report path, got %d", got)
}
}

// TestGetScanReport_LatestLatencyIndependentOfHistory verifies the MCP-2205 fix
// also covers the "latest report" path (GetScanReport by server name, used by the
// Web UI server-detail view). Resolving the latest Pass-1/Pass-2 jobs must use the
// lightweight metadata index plus targeted GetScanJob loads (at most two full job
// deserializations), not a full ListScanJobs over the server's scan history.
func TestGetScanReport_LatestLatencyIndependentOfHistory(t *testing.T) {
mock := newMockStorage()
store := newCountingStorage(mock)
svc := NewService(store, NewRegistry(t.TempDir(), zap.NewNop()), nil, t.TempDir(), zap.NewNop())

now := time.Now()

// Latest Pass-1 + Pass-2 with reports.
_ = mock.SaveScanJob(&ScanJob{ID: "latest-p1", ServerName: "srv", Status: ScanJobStatusCompleted, ScanPass: ScanPassSecurityScan, StartedAt: now.Add(-2 * time.Minute)})
_ = mock.SaveScanReport(&ScanReport{ID: "lr1", JobID: "latest-p1", ServerName: "srv", ScannerID: "s",
Findings: []ScanFinding{{RuleID: "T1", Title: "tp", Scanner: "s", ThreatLevel: ThreatLevelDangerous}}})
_ = mock.SaveScanJob(&ScanJob{ID: "latest-p2", ServerName: "srv", Status: ScanJobStatusCompleted, ScanPass: ScanPassSupplyChainAudit, StartedAt: now.Add(-1 * time.Minute)})
_ = mock.SaveScanReport(&ScanReport{ID: "lr2", JobID: "latest-p2", ServerName: "srv", ScannerID: "s",
Findings: []ScanFinding{{RuleID: "CVE-9", Title: "cve", Scanner: "s", ThreatLevel: ThreatLevelWarning}}})

// Large historical backlog.
for i := 0; i < 100; i++ {
_ = mock.SaveScanJob(&ScanJob{ID: fmt.Sprintf("old-%d", i), ServerName: "srv", Status: ScanJobStatusCompleted, ScanPass: ScanPassSecurityScan, StartedAt: now.Add(time.Duration(-10-i) * time.Minute)})
}

report, err := svc.GetScanReport(context.Background(), "srv")
if err != nil {
t.Fatalf("GetScanReport: %v", err)
}
if len(report.Findings) != 2 {
t.Fatalf("expected 2 merged findings, got %d", len(report.Findings))
}
if got := store.listCalls.Load(); got != 0 {
t.Errorf("expected 0 ListScanJobs (full-history) calls in latest-report path, got %d", got)
}
}
14 changes: 14 additions & 0 deletions internal/security/scanner/types.go
Original file line number Diff line number Diff line change
Expand Up @@ -120,6 +120,20 @@ type ScanJob struct {
ScanContext *ScanContext `json:"scan_context,omitempty"`
}

// ScanJobMeta is a lightweight projection of a scan job, persisted in a
// dedicated index bucket so that companion-job lookups during report
// aggregation never deserialize the full job payload (whose ScannerStatuses can
// carry large stdout/stderr blobs). This keeps report latency independent of a
// server's scan history. See MCP-2205.
type ScanJobMeta struct {
ID string `json:"id"`
ServerName string `json:"server_name"`
Status string `json:"status"`
ScanPass int `json:"scan_pass"`
StartedAt time.Time `json:"started_at"`
CompletedAt time.Time `json:"completed_at,omitempty"`
}

// ScanJobSummary is a lightweight view of a scan job for history listing
type ScanJobSummary struct {
ID string `json:"id"`
Expand Down
7 changes: 7 additions & 0 deletions internal/storage/bbolt.go
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,7 @@ func (b *BoltDB) initBuckets() error {
ActivityStatsBucket,
ScannersBucket,
ScanJobsBucket,
ScanJobIndexBucket,
ScanReportsBucket,
IntegrityBaselinesBucket,
OnboardingBucket,
Expand All @@ -104,6 +105,12 @@ func (b *BoltDB) initBuckets() error {
}
}

// Backfill the scan-job index for databases created before MCP-2205.
// Idempotent: only runs when the index is empty but jobs exist.
if err := backfillScanJobIndex(tx); err != nil {
return fmt.Errorf("failed to backfill scan job index: %w", err)
}

// Set schema version only for new databases. Existing databases keep their
// stored version so migrations can observe and upgrade them.
metaBucket := tx.Bucket([]byte(MetaBucket))
Expand Down
Loading
Loading