syz-cluster: basic support for finding invalidation

syz-cluster/dashboard/templates/series.html

@@ -148,6 +148,7 @@ <h3>Session {{.CreatedAt.Format "2006-01-02"}}</h3>
                       {{range .Findings}}
                       <tr>
                         <td>
+                          {{if not .InvalidatedAt.IsNull}}<b>[invalidated]</b>{{end}}
                           {{if .ReportURI}}
                             <a href="/findings/{{.ID}}/report" class="modal-link-raw">{{.Title}}</a>
                           {{else}}

syz-cluster/email-reporter/handler.go

@@ -126,19 +126,23 @@ func (h *Handler) IncomingEmail(ctx context.Context, msg *email.Email) error {
 
 	var reply string
 	for _, command := range msg.Commands {
+		var err error
 		switch command.Command {
 		case email.CmdUpstream:
-			err := h.apiClient.UpstreamReport(ctx, reportID, &api.UpstreamReportReq{
+			// Reply nothing on success.
+			err = h.apiClient.UpstreamReport(ctx, reportID, &api.UpstreamReportReq{
 				User: msg.Author,
 			})
-			if err != nil {
-				reply = fmt.Sprintf("Failed to process the command. Contact %s.",
-					h.emailConfig.SupportEmail)
-			}
+		case email.CmdInvalid:
 			// Reply nothing on success.
+			err = h.apiClient.InvalidateReport(ctx, reportID)
 		default:
 			reply = "Unknown command"
 		}
+		if err != nil {
+			reply = fmt.Sprintf("Failed to process the command. Contact %s.",
+				h.emailConfig.SupportEmail)
+		}
 	}
 
 	if reply == "" {

syz-cluster/email-reporter/handler_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/google/syzkaller/syz-cluster/pkg/emailclient"
 	"github.com/google/syzkaller/syz-cluster/pkg/reporter"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 var testEmailConfig = emailclient.TestEmailConfig()
@@ -64,6 +65,38 @@ func TestModerationReportFlow(t *testing.T) {
 	}, receivedEmail)
 }
 
+func TestReportInvalidationFlow(t *testing.T) {
+	env, ctx := app.TestEnvironment(t)
+	testSeries := controller.DummySeries()
+	handler, _, emailServer := setupHandlerTest(t, env, ctx, testSeries)
+
+	report, err := handler.PollAndReport(ctx)
+	require.NoError(t, err)
+
+	receivedEmail := emailServer.email()
+	require.NotNil(t, receivedEmail, "a moderation email must be sent")
+	receivedEmail.Body = nil // for now don't validate the body
+
+	// Emulate an "upstream" command.
+	err = handler.IncomingEmail(ctx, &email.Email{
+		BugIDs: []string{report.ID},
+		Commands: []*email.SingleCommand{
+			{
+				Command: email.CmdInvalid,
+			},
+		},
+	})
+	require.NoError(t, err)
+
+	// The report must be not sent upstream.
+	report, err = handler.PollAndReport(ctx)
+	require.NoError(t, err)
+	assert.Nil(t, report)
+
+	receivedEmail = emailServer.email()
+	assert.Nil(t, receivedEmail, "an email must not be sent upstream")
+}
+
 func TestInvalidReply(t *testing.T) {
 	env, ctx := app.TestEnvironment(t)
 	testSeries := controller.DummySeries()

syz-cluster/pkg/api/api.go

@@ -174,6 +174,7 @@ type Finding struct {
 	Build        BuildInfo `json:"build"`
 	LinkCRepro   string    `json:"c_repro"`
 	LinkSyzRepro string    `json:"syz_repro"`
+	Invalidated  bool      `json:"invalidated"`
 }
 
 type BuildInfo struct {

syz-cluster/pkg/api/reporter.go

@@ -45,6 +45,11 @@ func (client ReporterClient) UpstreamReport(ctx context.Context, id string, req
 	return err
 }
 
+func (client ReporterClient) InvalidateReport(ctx context.Context, id string) error {
+	_, err := postJSON[any, any](ctx, client.baseURL+"/reports/"+id+"/invalidate", nil)
+	return err
+}
+
 type RecordReplyReq struct {
 	MessageID string `json:"message_id"`
 	ReportID  string `json:"report_id"`

syz-cluster/pkg/db/entities.go

@@ -133,15 +133,20 @@ type SessionTest struct {
 }
 
 type Finding struct {
-	ID              string `spanner:"ID"`
-	SessionID       string `spanner:"SessionID"`
-	TestName        string `spanner:"TestName"`
-	Title           string `spanner:"Title"`
-	ReportURI       string `spanner:"ReportURI"`
-	LogURI          string `spanner:"LogURI"`
-	SyzReproURI     string `spanner:"SyzReproURI"`
-	SyzReproOptsURI string `spanner:"SyzReproOptsURI"`
-	CReproURI       string `spanner:"CReproURI"`
+	ID              string           `spanner:"ID"`
+	SessionID       string           `spanner:"SessionID"`
+	TestName        string           `spanner:"TestName"`
+	Title           string           `spanner:"Title"`
+	ReportURI       string           `spanner:"ReportURI"`
+	LogURI          string           `spanner:"LogURI"`
+	SyzReproURI     string           `spanner:"SyzReproURI"`
+	SyzReproOptsURI string           `spanner:"SyzReproOptsURI"`
+	CReproURI       string           `spanner:"CReproURI"`
+	InvalidatedAt   spanner.NullTime `spanner:"InvalidatedAt"`
+}
+
+func (f *Finding) SetInvalidatedAt(t time.Time) {
+	f.InvalidatedAt = spanner.NullTime{Time: t, Valid: true}
 }
 
 type SessionReport struct {

syz-cluster/pkg/db/migrations/7_invalidated_findings.down.sql

@@ -0,0 +1 @@
+ALTER TABLE Findings DROP COLUMN InvalidatedAt;

syz-cluster/pkg/db/migrations/7_invalidated_findings.up.sql

@@ -0,0 +1 @@
+ALTER TABLE Findings ADD COLUMN InvalidatedAt TIMESTAMP DEFAULT(NULL);

syz-cluster/pkg/db/series_repo.go

@@ -173,8 +173,9 @@ func (repo *SeriesRepository) ListLatest(ctx context.Context, filter SeriesFilte
 		stmt.SQL += ")"
 	}
 	if filter.WithFindings {
-		stmt.SQL += " AND Series.LatestSessionID IS NOT NULL " +
-			"AND EXISTS(SELECT 1 FROM Findings WHERE Findings.SessionID = Series.LatestSessionID)"
+		stmt.SQL += " AND Series.LatestSessionID IS NOT NULL AND EXISTS(" +
+			"SELECT 1 FROM Findings WHERE " +
+			"Findings.SessionID = Series.LatestSessionID AND Findings.InvalidatedAt IS NULL)"
 	}
 	stmt.SQL += " ORDER BY PublishedAt DESC, ID"
 	if filter.Limit > 0 {
@@ -262,7 +263,8 @@ func (repo *SeriesRepository) queryFindingCounts(ctx context.Context, ro *spanne
 	}
 	list, err := readEntities[findingCount](ctx, repo.client.Single(), spanner.Statement{
 		SQL: "SELECT `SessionID`, COUNT(`ID`) as `Count` FROM `Findings` " +
-			"WHERE `SessionID` IN UNNEST(@ids) GROUP BY `SessionID`",
+			"WHERE `SessionID` IN UNNEST(@ids) AND `Findings`.`InvalidatedAt` IS NULL " +
+			"GROUP BY `SessionID`",
 		Params: map[string]interface{}{
 			"ids": keys,
 		},

syz-cluster/pkg/db/series_repo_test.go

@@ -145,7 +145,7 @@ func TestSeriesRepositoryList(t *testing.T) {
 	})
 
 	dtd.addSessionTest(session, "test")
-	dtd.addFinding(session, "title", "test")
+	finding := dtd.addFinding(session, "title", "test")
 	dtd.finishSession(session)
 	t.Run("query_finding_count", func(t *testing.T) {
 		list, err := repo.ListLatest(ctx, SeriesFilter{Status: SessionStatusFinished}, time.Time{})
@@ -160,6 +160,18 @@ func TestSeriesRepositoryList(t *testing.T) {
 		assert.Len(t, list, 1)
 		assert.Equal(t, "Series 2", list[0].Series.Title)
 	})
+
+	dtd.invalidateFinding(finding)
+	t.Run("invalidated_findings", func(t *testing.T) {
+		list, err := repo.ListLatest(ctx, SeriesFilter{WithFindings: true}, time.Time{})
+		assert.NoError(t, err)
+		assert.Len(t, list, 0)
+		// When not filtered, ensure invalidated findings are not counted in.
+		list, err = repo.ListLatest(ctx, SeriesFilter{Status: SessionStatusFinished}, time.Time{})
+		assert.NoError(t, err)
+		assert.Len(t, list, 1)
+		assert.Equal(t, 0, list[0].Findings)
+	})
 }
 
 func TestSeriesRepositoryUpdate(t *testing.T) {

syz-cluster/pkg/db/util_test.go

@@ -65,11 +65,21 @@ func (d *dummyTestData) finishSession(session *Session) {
 	assert.NoError(d.t, err)
 }
 
-func (d *dummyTestData) addFinding(session *Session, title, test string) {
+func (d *dummyTestData) addFinding(session *Session, title, test string) *Finding {
 	findingRepo := NewFindingRepository(d.client)
-	assert.NoError(d.t, findingRepo.mustStore(d.ctx, &Finding{
+	finding := &Finding{
 		SessionID: session.ID,
 		Title:     title,
 		TestName:  test,
+	}
+	assert.NoError(d.t, findingRepo.mustStore(d.ctx, finding))
+	return finding
+}
+
+func (d *dummyTestData) invalidateFinding(f *Finding) {
+	findingRepo := NewFindingRepository(d.client)
+	assert.NoError(d.t, findingRepo.Update(d.ctx, f.ID, func(f *Finding) error {
+		f.SetInvalidatedAt(time.Now())
+		return nil
 	}))
 }

syz-cluster/pkg/report/template.txt

@@ -75,4 +75,7 @@ The email will later be sent to:
 
 If the report looks fine to you, reply with:
 #syz upstream
+
+If the report is a false positive, reply with
+#syz invalid
 {{end}}

syz-cluster/pkg/report/testdata/1.moderation.txt

@@ -62,3 +62,6 @@ The email will later be sent to:
 If the report looks fine to you, reply with:
 #syz upstream
 
+If the report is a false positive, reply with
+#syz invalid
+

syz-cluster/pkg/report/testdata/2.moderation.txt

@@ -45,3 +45,6 @@ The email will later be sent to:
 If the report looks fine to you, reply with:
 #syz upstream
 
+If the report is a false positive, reply with
+#syz invalid
+

syz-cluster/pkg/reporter/api.go

@@ -31,6 +31,7 @@ func (s *APIServer) Mux() *http.ServeMux {
 	mux := http.NewServeMux()
 	mux.HandleFunc("/reports/{report_id}/upstream", s.upstreamReport)
 	mux.HandleFunc("/reports/{report_id}/confirm", s.confirmReport)
+	mux.HandleFunc("/reports/{report_id}/invalidate", s.invalidateReport)
 	mux.HandleFunc("/reports/record_reply", s.recordReply)
 	mux.HandleFunc("/reports/last_reply", s.lastReply)
 	mux.HandleFunc("/reports", s.nextReports)
@@ -48,6 +49,12 @@ func (s *APIServer) upstreamReport(w http.ResponseWriter, r *http.Request) {
 	reply[interface{}](w, nil, err)
 }
 
+func (s *APIServer) invalidateReport(w http.ResponseWriter, r *http.Request) {
+	// TODO: journal the action.
+	err := s.reportService.Invalidate(r.Context(), r.PathValue("report_id"))
+	reply[interface{}](w, nil, err)
+}
+
 func (s *APIServer) nextReports(w http.ResponseWriter, r *http.Request) {
 	resp, err := s.reportService.Next(r.Context(), r.FormValue("reporter"))
 	reply(w, resp, err)

syz-cluster/pkg/reporter/api_test.go

@@ -10,7 +10,9 @@ import (
 	"github.com/google/syzkaller/syz-cluster/pkg/api"
 	"github.com/google/syzkaller/syz-cluster/pkg/app"
 	"github.com/google/syzkaller/syz-cluster/pkg/controller"
+	"github.com/google/syzkaller/syz-cluster/pkg/service"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestAPIReportFlow(t *testing.T) {
@@ -192,3 +194,40 @@ func TestReplyReporting(t *testing.T) {
 		}, resp)
 	})
 }
+
+func TestInvalidate(t *testing.T) {
+	env, ctx := app.TestEnvironment(t)
+	client := controller.TestServer(t, env)
+	testSeries := controller.DummySeries()
+	ids := controller.FakeSeriesWithFindings(t, ctx, env, client, testSeries)
+
+	generator := NewGenerator(env)
+	err := generator.Process(ctx, 1)
+	require.NoError(t, err)
+
+	// Create a report.
+	reportClient := TestServer(t, env)
+	nextResp, err := reportClient.GetNextReport(ctx, api.LKMLReporter)
+	require.NoError(t, err)
+	reportID := nextResp.Report.ID
+	err = reportClient.ConfirmReport(ctx, reportID)
+	require.NoError(t, err)
+
+	// Invalidate the findings.
+	err = reportClient.InvalidateReport(ctx, reportID)
+	require.NoError(t, err)
+
+	// Report should not appear in Next().
+	emptyNext, err := reportClient.GetNextReport(ctx, api.LKMLReporter)
+	require.NoError(t, err)
+	assert.Nil(t, emptyNext.Report)
+
+	// All findings must be invalidated.
+	findingService := service.NewFindingService(env)
+	list, err := findingService.List(ctx, ids.SessionID, 0)
+	require.NoError(t, err)
+	assert.Len(t, list, 2)
+	for i, finding := range list {
+		assert.True(t, finding.Invalidated, "finding %d must be invalidated", i)
+	}
+}

syz-cluster/pkg/service/finding.go

@@ -7,6 +7,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/google/syzkaller/syz-cluster/pkg/api"
 	"github.com/google/syzkaller/syz-cluster/pkg/app"
@@ -87,6 +88,23 @@ func (s *FindingService) saveAssets(finding *db.Finding, req *api.NewFinding) er
 	return nil
 }
 
+func (s *FindingService) InvalidateSession(ctx context.Context, sessionID string) error {
+	findings, err := s.findingRepo.ListForSession(ctx, sessionID, 0)
+	if err != nil {
+		return err
+	}
+	for _, finding := range findings {
+		err := s.findingRepo.Update(ctx, finding.ID, func(finding *db.Finding) error {
+			finding.SetInvalidatedAt(time.Now())
+			return nil
+		})
+		if err != nil {
+			return fmt.Errorf("failed to update finding %s: %w", finding.ID, err)
+		}
+	}
+	return nil
+}
+
 func (s *FindingService) List(ctx context.Context, sessionID string, limit int) ([]*api.Finding, error) {
 	list, err := s.findingRepo.ListForSession(ctx, sessionID, limit)
 	if err != nil {
@@ -112,6 +130,9 @@ func (s *FindingService) List(ctx context.Context, sessionID string, limit int)
 		if item.CReproURI != "" {
 			finding.LinkCRepro = s.urls.FindingCRepro(item.ID)
 		}
+		if !item.InvalidatedAt.IsNull() {
+			finding.Invalidated = true
+		}
 		build := testPerName[item.TestName].PatchedBuild
 		if build != nil {
 			finding.Build = makeBuildInfo(s.urls, build)

syz-cluster/pkg/service/report.go

@@ -51,7 +51,7 @@ var ErrNotOnModeration = errors.New("the report is not on moderation")
 func (rs *ReportService) Upstream(ctx context.Context, id string, req *api.UpstreamReportReq) error {
 	rep, err := rs.query(ctx, id)
 	if err != nil {
-		return nil
+		return err
 	} else if !rep.Moderation {
 		return ErrNotOnModeration
 	}
@@ -68,6 +68,15 @@ func (rs *ReportService) Upstream(ctx context.Context, id string, req *api.Upstr
 	return nil
 }
 
+func (rs *ReportService) Invalidate(ctx context.Context, id string) error {
+	rep, err := rs.query(ctx, id)
+	if err != nil {
+		return err
+	}
+	// For now, invalidate all the findings at once - later we can do it more selectively.
+	return rs.findingService.InvalidateSession(ctx, rep.SessionID)
+}
+
 const maxFindingsPerReport = 5
 
 func (rs *ReportService) Next(ctx context.Context, reporter string) (*api.NextReportResp, error) {