syz-cluster: add email-reporter component

Add the high-level logic for the email-reporter component that will
be responsible for sending bug reports and reacting to incoming emails.

12 files changed, 494 insertions, 1 deletions

diff --git a/syz-cluster/Makefile b/syz-cluster/Makefile
index 4473e9552..af203c1bd 100644
--- a/syz-cluster/Makefile
+++ b/syz-cluster/Makefile
@@ -24,6 +24,7 @@ $(eval $(call build_image_rules,./dashboard,web-dashboard))
 $(eval $(call build_image_rules,./reporter-server,reporter-server))
 $(eval $(call build_image_rules,./series-tracker,series-tracker))
 $(eval $(call build_image_rules,./db-mgmt,db-mgmt))
+$(eval $(call build_image_rules,./email-reporter,email-reporter))
 $(eval $(call build_image_rules,./workflow/triage-step,triage-step))
 $(eval $(call build_image_rules,./workflow/build-step,build-step))
 $(eval $(call build_image_rules,./workflow/fuzz-step,fuzz-step))
diff --git a/syz-cluster/email-reporter/Dockerfile b/syz-cluster/email-reporter/Dockerfile
new file mode 100644
index 000000000..779544fe3
--- /dev/null
+++ b/syz-cluster/email-reporter/Dockerfile
@@ -0,0 +1,22 @@
+FROM golang:1.23-alpine AS email-reporter-builder
+
+WORKDIR /build
+
+# Prepare the dependencies.
+COPY go.mod ./
+COPY go.sum ./
+RUN go mod download
+COPY pkg/gcs/ pkg/gcs/
+
+# Build the tool.
+COPY syz-cluster/email-reporter/ syz-cluster/email-reporter/
+COPY syz-cluster/pkg/ syz-cluster/pkg/
+RUN go build -o /bin/email-reporter /build/syz-cluster/email-reporter
+
+# Build the container.
+FROM alpine:latest
+WORKDIR /app
+
+COPY --from=email-reporter-builder /bin/email-reporter /bin/email-reporter
+
+ENTRYPOINT ["/bin/email-reporter"]
diff --git a/syz-cluster/email-reporter/deployment.yaml b/syz-cluster/email-reporter/deployment.yaml
new file mode 100644
index 000000000..35def6721
--- /dev/null
+++ b/syz-cluster/email-reporter/deployment.yaml
@@ -0,0 +1,35 @@
+# Copyright 2025 syzkaller project authors. All rights reserved.
+# Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.
+
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: email-reporter-deployment
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: email-reporter
+  template:
+    metadata:
+      labels:
+        app: email-reporter
+    spec:
+      serviceAccountName: gke-service-ksa
+      containers:
+      - name: email-reporter
+        image: ${IMAGE_PREFIX}email-reporter:${IMAGE_TAG}
+        volumeMounts:
+          - name: config-volume
+            mountPath: /config
+        resources:
+          requests:
+            cpu: 2
+            memory: 8G
+          limits:
+            cpu: 4
+            memory: 16G
+      volumes:
+      - name: config-volume
+        configMap:
+          name: global-config
diff --git a/syz-cluster/email-reporter/handler.go b/syz-cluster/email-reporter/handler.go
new file mode 100644
index 000000000..a956b77f9
--- /dev/null
+++ b/syz-cluster/email-reporter/handler.go
@@ -0,0 +1,143 @@
+// Copyright 2025 syzkaller project authors. All rights reserved.
+// Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/google/syzkaller/pkg/email"
+	"github.com/google/syzkaller/syz-cluster/pkg/api"
+	"github.com/google/syzkaller/syz-cluster/pkg/app"
+	"github.com/google/syzkaller/syz-cluster/pkg/report"
+)
+
+type EmailToSend struct {
+	To        []string
+	Cc        []string
+	Subject   string
+	InReplyTo string
+	Body      []byte
+}
+
+type SendEmailCb func(context.Context, *EmailToSend) (string, error)
+
+type Handler struct {
+	apiClient   *api.ReporterClient
+	emailConfig *app.EmailConfig
+	sender      SendEmailCb
+}
+
+func (h *Handler) PollReportsLoop(ctx context.Context, pollPeriod time.Duration) {
+	for {
+		_, err := h.PollAndReport(ctx)
+		if err != nil {
+			app.Errorf("%v", err)
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(pollPeriod):
+		}
+	}
+}
+
+func (h *Handler) PollAndReport(ctx context.Context) (*api.SessionReport, error) {
+	reply, err := h.apiClient.GetNextReport(ctx, api.EmailReporter)
+	if err != nil {
+		return nil, fmt.Errorf("failed to poll the next report: %w", err)
+	} else if reply == nil || reply.Report == nil {
+		return nil, nil
+	}
+	report := reply.Report
+	log.Printf("report %q is to be sent", report.ID)
+	if err := h.report(ctx, report); err != nil {
+		// TODO: consider retrying if the error happened before we attempted
+		// to actually send the message.
+		return nil, fmt.Errorf("failed to report %q: %w", report.ID, err)
+	}
+	return report, nil
+}
+
+func (h *Handler) report(ctx context.Context, rep *api.SessionReport) error {
+	// Start by confirming the report - it's better to not send an email at all than to send it multiple times.
+	err := h.apiClient.ConfirmReport(ctx, rep.ID)
+	if err != nil {
+		return fmt.Errorf("failed to confirm: %w", err)
+	}
+
+	// Construct and send the message.
+	body, err := report.Render(rep, h.emailConfig)
+	if err != nil {
+		// This should never be happening..
+		return fmt.Errorf("failed to render the template: %w", err)
+	}
+	toSend := &EmailToSend{
+		Subject: "Re: " + rep.Series.Title, // TODO: use the original rather than the stripped title.
+		To:      rep.Cc,
+		Body:    body,
+	}
+	if rep.Moderation {
+		toSend.To = []string{h.emailConfig.ModerationList}
+	} else {
+		// We assume that email reporting is used for series received over emails.
+		toSend.InReplyTo = rep.Series.ExtID
+		toSend.To = rep.Cc
+		toSend.Cc = []string{h.emailConfig.ArchiveList}
+	}
+	msgID, err := h.sender(ctx, toSend)
+	if err != nil {
+		return fmt.Errorf("failed to send: %w", err)
+	}
+
+	// Now that the report is sent, update the link to the email discussion.
+	err = h.apiClient.UpdateReport(ctx, rep.ID, &api.UpdateReportReq{
+		// TODO: for Lore emails, set Link = lore.Link(msgID).
+		MessageID: msgID,
+	})
+	if err != nil {
+		return fmt.Errorf("failed to update: %w", err)
+	}
+	return nil
+}
+
+// IncomingEmail assumes that the related report ID is already extracted and resides in msg.BugIDs.
+func (h *Handler) IncomingEmail(ctx context.Context, msg *email.Email) error {
+	if len(msg.BugIDs) == 0 {
+		// Unrelated email.
+		return nil
+	}
+	reportID := msg.BugIDs[0]
+
+	var reply string
+	for _, command := range msg.Commands {
+		switch command.Command {
+		case email.CmdUpstream:
+			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)
+			}
+			// Reply nothing on success.
+		default:
+			reply = "Unknown command"
+		}
+	}
+
+	if reply == "" {
+		return nil
+	}
+	_, err := h.sender(ctx, &EmailToSend{
+		To:        []string{msg.Author},
+		Cc:        msg.Cc,
+		Subject:   "Re: " + msg.Subject,
+		InReplyTo: msg.MessageID,
+		Body:      []byte(email.FormReply(msg, reply)),
+	})
+	return err
+}
diff --git a/syz-cluster/email-reporter/handler_test.go b/syz-cluster/email-reporter/handler_test.go
new file mode 100644
index 000000000..355a77283
--- /dev/null
+++ b/syz-cluster/email-reporter/handler_test.go
@@ -0,0 +1,167 @@
+// Copyright 2025 syzkaller project authors. All rights reserved.
+// Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.
+
+package main
+
+import (
+	"context"
+	"testing"
+
+	"github.com/google/syzkaller/pkg/email"
+	"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/reporter"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestModerationReportFlow(t *testing.T) {
+	env, ctx := app.TestEnvironment(t)
+	testSeries := controller.DummySeries()
+	handler, emailServer := setupHandlerTest(t, env, ctx, testSeries)
+
+	report, err := handler.PollAndReport(ctx)
+	assert.NoError(t, err)
+
+	receivedEmail := emailServer.email()
+	assert.NotNil(t, receivedEmail, "a moderation email must be sent")
+	receivedEmail.Body = nil // for now don't validate the body
+	assert.Equal(t, &EmailToSend{
+		To:      []string{testEmailConfig.ModerationList},
+		Subject: "Re: " + testSeries.Title,
+		// Note that InReplyTo and Cc are nil.
+	}, receivedEmail)
+
+	// Emulate an "upstream" command.
+	err = handler.IncomingEmail(ctx, &email.Email{
+		BugIDs: []string{report.ID},
+		Commands: []*email.SingleCommand{
+			{
+				Command: email.CmdUpstream,
+			},
+		},
+	})
+	assert.NoError(t, err)
+
+	// The report must be sent upstream.
+	_, err = handler.PollAndReport(ctx)
+	assert.NoError(t, err)
+
+	receivedEmail = emailServer.email()
+	assert.NotNil(t, receivedEmail, "an email must be sent upstream")
+	receivedEmail.Body = nil
+	assert.Equal(t, &EmailToSend{
+		To:        testSeries.Cc,
+		Cc:        []string{testEmailConfig.ArchiveList},
+		Subject:   "Re: " + testSeries.Title,
+		InReplyTo: testSeries.ExtID,
+	}, receivedEmail)
+}
+
+func TestInvalidReply(t *testing.T) {
+	env, ctx := app.TestEnvironment(t)
+	testSeries := controller.DummySeries()
+	handler, emailServer := setupHandlerTest(t, env, ctx, testSeries)
+
+	report, err := handler.PollAndReport(ctx)
+	assert.NoError(t, err)
+
+	receivedEmail := emailServer.email()
+	assert.NotNil(t, receivedEmail, "a moderation email must be sent")
+	receivedEmail.Body = nil
+
+	t.Run("unrelated email", func(t *testing.T) {
+		err = handler.IncomingEmail(ctx, &email.Email{
+			Commands: []*email.SingleCommand{
+				{
+					Command: email.CmdUpstream,
+				},
+			},
+		})
+		assert.NoError(t, err)
+		_, err = handler.PollAndReport(ctx)
+		assert.NoError(t, err)
+		// No email must be sent in reply.
+		assert.Nil(t, emailServer.email())
+	})
+
+	t.Run("unsupported command", func(t *testing.T) {
+		err := handler.IncomingEmail(ctx, &email.Email{
+			Author:    "user@email.com",
+			Subject:   "Command",
+			BugIDs:    []string{report.ID},
+			Cc:        []string{"a@a.com", "b@b.com"},
+			MessageID: "user-reply-msg-id",
+			Commands: []*email.SingleCommand{
+				{
+					Command: email.CmdFix,
+				},
+			},
+			Body: `#syz fix: abcd`,
+		})
+		assert.NoError(t, err)
+		reply := emailServer.email()
+		assert.NotNil(t, reply)
+		assert.Equal(t, &EmailToSend{
+			To:        []string{"user@email.com"},
+			Cc:        []string{"a@a.com", "b@b.com"},
+			Subject:   "Re: Command",
+			InReplyTo: "user-reply-msg-id",
+			Body: []byte(`> #syz fix: abcd
+
+Unknown command
+
+`),
+		}, reply)
+	})
+}
+
+func setupHandlerTest(t *testing.T, env *app.AppEnvironment, ctx context.Context,
+	series *api.Series) (*Handler, *fakeSender) {
+	client := controller.TestServer(t, env)
+	controller.FakeSeriesWithFindings(t, ctx, env, client, series)
+
+	generator := reporter.NewGenerator(env)
+	err := generator.Process(ctx, 1)
+	assert.NoError(t, err)
+
+	emailServer := makeFakeSender()
+	handler := &Handler{
+		apiClient:   reporter.TestServer(t, env),
+		emailConfig: testEmailConfig,
+		sender:      emailServer.send,
+	}
+	return handler, emailServer
+}
+
+type fakeSender struct {
+	ch chan *EmailToSend
+}
+
+func makeFakeSender() *fakeSender {
+	return &fakeSender{
+		ch: make(chan *EmailToSend, 16),
+	}
+}
+
+func (f *fakeSender) send(ctx context.Context, e *EmailToSend) (string, error) {
+	f.ch <- e
+	return "email-id", nil
+}
+
+func (f *fakeSender) email() *EmailToSend {
+	select {
+	case e := <-f.ch:
+		return e
+	default:
+		return nil
+	}
+}
+
+var testEmailConfig = &app.EmailConfig{
+	Name:           "name",
+	DocsLink:       "docs",
+	Sender:         "a@b.com",
+	ModerationList: "moderation@list.com",
+	ArchiveList:    "archive@list.com",
+}
diff --git a/syz-cluster/email-reporter/kustomization.yaml b/syz-cluster/email-reporter/kustomization.yaml
new file mode 100644
index 000000000..1a385170b
--- /dev/null
+++ b/syz-cluster/email-reporter/kustomization.yaml
@@ -0,0 +1,5 @@
+# Copyright 2025 syzkaller project authors. All rights reserved.
+# Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.
+
+resources:
+- deployment.yaml
diff --git a/syz-cluster/email-reporter/main.go b/syz-cluster/email-reporter/main.go
new file mode 100644
index 000000000..52bc3f651
--- /dev/null
+++ b/syz-cluster/email-reporter/main.go
@@ -0,0 +1,55 @@
+// Copyright 2025 syzkaller project authors. All rights reserved.
+// Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.
+
+// NOTE: This app assumes that only one copy of it is runnning at the same time.
+
+package main
+
+import (
+	"context"
+	"log"
+	"time"
+
+	"github.com/google/syzkaller/pkg/email"
+	"github.com/google/syzkaller/syz-cluster/pkg/app"
+)
+
+// TODO: add extra sanity checks that would prevent flooding the mailing lists:
+// - this pod may crash and be restarted by K8S: this complicates accounting,
+// - the send operation might return an error, yet an email would be actually sent: back off on errors?
+
+// How often to check whether there are new emails to be sent.
+const pollPeriod = 30 * time.Second
+
+func main() {
+	ctx := context.Background()
+	cfg, err := app.Config()
+	if err != nil {
+		app.Fatalf("failed to load config: %v", err)
+	}
+	if cfg.EmailReporting == nil {
+		app.Fatalf("reporting is not configured: %v", err)
+	}
+	sender := &smtpSender{}
+	handler := &Handler{
+		apiClient:   app.DefaultReporterClient(),
+		emailConfig: cfg.EmailReporting,
+		sender:      sender.Send,
+	}
+	emailStream := NewLoreEmailStreamer()
+	ch := make(chan *email.Email, 16)
+	go func() {
+		for newEmail := range ch {
+			log.Printf("received email %q", newEmail.MessageID)
+			err := handler.IncomingEmail(ctx, newEmail)
+			if err != nil {
+				// Note that we just print an error and go on instead of retrying.
+				// Some retrying may be reasonable, but it also comes with a risk of flooding
+				// the mailing lists.
+				app.Errorf("email %q: failed to process: %v", newEmail.MessageID, err)
+			}
+		}
+	}()
+	go handler.PollReportsLoop(ctx, pollPeriod)
+	go emailStream.Loop(ctx, ch)
+}
diff --git a/syz-cluster/email-reporter/sender.go b/syz-cluster/email-reporter/sender.go
new file mode 100644
index 000000000..31a1b3358
--- /dev/null
+++ b/syz-cluster/email-reporter/sender.go
@@ -0,0 +1,22 @@
+// Copyright 2025 syzkaller project authors. All rights reserved.
+// Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.
+
+package main
+
+import "context"
+
+// TODO: how can we test it?
+// Using some STMP server library is probably an overkill?
+
+type smtpSender struct {
+}
+
+// Send constructs a raw email from EmailToSend and sends it over SMTP.
+func (sender *smtpSender) Send(ctx context.Context, item *EmailToSend) (string, error) {
+	// TODO:
+	// 1) Fill in email headers, including the Message ID.
+	// https://pkg.go.dev/github.com/emersion/go-message/mail#Header.GenerateMessageIDWithHostname
+	// 2) Send over STMP:
+	// https://pkg.go.dev/net/smtp
+	return "", nil
+}
diff --git a/syz-cluster/email-reporter/stream.go b/syz-cluster/email-reporter/stream.go
new file mode 100644
index 000000000..9937e645a
--- /dev/null
+++ b/syz-cluster/email-reporter/stream.go
@@ -0,0 +1,32 @@
+// Copyright 2025 syzkaller project authors. All rights reserved.
+// Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.
+
+package main
+
+import (
+	"context"
+
+	"github.com/google/syzkaller/pkg/email"
+)
+
+// TODO: there's a lot of common code with series-tracker.
+// TODO: alternatively, we could parse the whole archive and track each email via In-Reply-To to the original email.
+
+type LoreEmailStreamer struct {
+}
+
+func NewLoreEmailStreamer() *LoreEmailStreamer {
+	return &LoreEmailStreamer{}
+}
+
+func (s *LoreEmailStreamer) Loop(ctx context.Context, writeTo chan *email.Email) {
+	<-ctx.Done()
+
+	// !! We assume that the archive mostly consists of relevant emails.
+	// 1. Query the last recorded discussion via API.
+	// 2. Poll the lore archive and query the emails starting from the date returned in (1).
+	// 3. Parse the email using email.Parse().
+	// 4. Report the new email via API, figure out which report was involved, save report ID to msg's BugIDs.
+	// 5. Push to the channel only if the message has not been seen before.
+	//    Also, we probablty don't want to react to old messages (e.g. > 1 day from now).
+}
diff --git a/syz-cluster/overlays/gke/global-config.yaml b/syz-cluster/overlays/gke/global-config.yaml
index 85cfe4c07..07e1cb5f0 100644
--- a/syz-cluster/overlays/gke/global-config.yaml
+++ b/syz-cluster/overlays/gke/global-config.yaml
@@ -11,3 +11,7 @@ data:
     loreArchives:
       - netdev
       - linux-ext4
+    emailReporting:
+      name: test-name
+      docs: http://docs/
+      supportEmail: name@email.com
diff --git a/syz-cluster/overlays/minikube/global-config.yaml b/syz-cluster/overlays/minikube/global-config.yaml
index e0c126082..d4887b0ab 100644
--- a/syz-cluster/overlays/minikube/global-config.yaml
+++ b/syz-cluster/overlays/minikube/global-config.yaml
@@ -11,3 +11,7 @@ data:
     # Whatever, it's just for debugging.
     loreArchives:
       - linux-wireless
+    emailReporting:
+      name: test-name
+      docs: http://docs/
+      supportEmail: name@email.com
diff --git a/syz-cluster/pkg/app/env.go b/syz-cluster/pkg/app/env.go
index e54cb30b2..7a5b8cca5 100644
--- a/syz-cluster/pkg/app/env.go
+++ b/syz-cluster/pkg/app/env.go
@@ -70,6 +70,9 @@ func DefaultStorage(ctx context.Context) (blob.Storage, error) {
 }
 
 func DefaultClient() *api.Client {
-	// TODO: take it from some env variable.
 	return api.NewClient(`http://controller-service:8080`)
 }
+
+func DefaultReporterClient() *api.ReporterClient {
+	return api.NewReporterClient(`http://reporter-server-service:8080`)
+}