#12: implement Antispam & Antivirus scanning pipeline (Rspamd & ClamAV)

services/protocols/internal/application/port/content_scanner.go

@@ -0,0 +1,19 @@
+package port
+
+import (
+	"context"
+
+	"lambdamail/protocols/internal/domain/valueobject"
+)
+
+type ScanInput struct {
+	ClientIP   string
+	HeloDomain string
+	Sender     string
+	Recipient  string
+	Payload    []byte
+}
+
+type ContentScanner interface {
+	Scan(ctx context.Context, input ScanInput) (*valueobject.ScanResult, error)
+}

services/protocols/internal/application/port/inbound_message_repository.go

@@ -13,6 +13,7 @@ type PersistInboundMessageInput struct {
 	Blob             BlobRef
 	SenderAddress    string
 	RecipientAddress string
+	TargetFolderName string // "INBOX", "Junk", etc. (defaults to "INBOX" if empty)
 	SPFResult        string
 	DKIMResult       string
 	DMARCResult      string

services/protocols/internal/application/usecase/process_inbound_email.go

@@ -3,11 +3,13 @@
 package usecase
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"io"
 
 	"lambdamail/protocols/internal/application/port"
+	"lambdamail/protocols/internal/domain/valueobject"
 )
 
 // ErrRecipientNotFound is returned by ResolveRecipient when no active
@@ -36,6 +38,7 @@ type ProcessInboundEmailInput struct {
 // resolution + durable persistence; SPF/DKIM/DMARC/spam/virus checks land in
 // later sub-projects).
 type ProcessInboundEmailUseCase struct {
+	scanner        port.ContentScanner
 	mailboxes      port.MailboxRepository
 	blobs          port.BlobStorage
 	messages       port.InboundMessageRepository
@@ -47,6 +50,10 @@ func NewProcessInboundEmailUseCase(mailboxes port.MailboxRepository, blobs port.
 	return &ProcessInboundEmailUseCase{mailboxes: mailboxes, blobs: blobs, messages: messages}
 }
 
+func (uc *ProcessInboundEmailUseCase) SetScanner(scanner port.ContentScanner) {
+	uc.scanner = scanner
+}
+
 func (uc *ProcessInboundEmailUseCase) SetTrackerManager(tm *MailboxTrackerManager, folders port.ImapFolderRepository) {
 	uc.trackerManager = tm
 	uc.folders = folders
@@ -77,7 +84,41 @@ func (uc *ProcessInboundEmailUseCase) ResolveRecipient(ctx context.Context, addr
 // the same blob (PLAN.md section 9's dedup design: 1 email to 50
 // recipients = 1 file).
 func (uc *ProcessInboundEmailUseCase) Handle(ctx context.Context, input ProcessInboundEmailInput) error {
-	blob, err := uc.blobs.Store(ctx, input.Body)
+	payload, err := io.ReadAll(input.Body)
+	if err != nil {
+		return err
+	}
+
+	targetFolder := "INBOX"
+
+	if uc.scanner != nil {
+		recipientAddr := ""
+		if len(input.RecipientAddresses) > 0 {
+			recipientAddr = input.RecipientAddresses[0]
+		}
+		scanRes, err := uc.scanner.Scan(ctx, port.ScanInput{
+			Sender:    input.Sender,
+			Recipient: recipientAddr,
+			Payload:   payload,
+		})
+		if err != nil {
+			return err
+		}
+		if scanRes != nil {
+			switch scanRes.Verdict {
+			case valueobject.ScanVerdictVirusReject:
+				return errors.New("554 5.7.1 Virus detected: " + scanRes.VirusName)
+			case valueobject.ScanVerdictSpamReject:
+				return errors.New("554 5.7.1 Spam threshold exceeded")
+			case valueobject.ScanVerdictGreylist:
+				return errors.New("451 4.7.1 Greylisted, please try again later")
+			case valueobject.ScanVerdictSpamJunk:
+				targetFolder = "Junk"
+			}
+		}
+	}
+
+	blob, err := uc.blobs.Store(ctx, bytes.NewReader(payload))
 	if err != nil {
 		return err
 	}
@@ -88,6 +129,7 @@ func (uc *ProcessInboundEmailUseCase) Handle(ctx context.Context, input ProcessI
 			Blob:             blob,
 			SenderAddress:    input.Sender,
 			RecipientAddress: input.RecipientAddresses[i],
+			TargetFolderName: targetFolder,
 			SPFResult:        "none",
 			DKIMResult:       "none",
 			DMARCResult:      "none",
@@ -97,7 +139,7 @@ func (uc *ProcessInboundEmailUseCase) Handle(ctx context.Context, input ProcessI
 		}
 
 		if uc.trackerManager != nil && uc.folders != nil {
-			folderRec, err := uc.folders.FindByName(ctx, recipient.ID.String(), "INBOX")
+			folderRec, err := uc.folders.FindByName(ctx, recipient.ID.String(), targetFolder)
 			if err == nil && folderRec != nil {
 				uc.trackerManager.NotifyNumMessages(folderRec.ID, folderRec.NumMessages)
 			}

services/protocols/internal/application/usecase/scanning_pipeline.go

@@ -0,0 +1,55 @@
+package usecase
+
+import (
+	"context"
+	"fmt"
+
+	"lambdamail/protocols/internal/application/port"
+	"lambdamail/protocols/internal/domain/valueobject"
+)
+
+type ScanningPipeline struct {
+	scanners []port.ContentScanner
+}
+
+func NewScanningPipeline(scanners ...port.ContentScanner) *ScanningPipeline {
+	return &ScanningPipeline{scanners: scanners}
+}
+
+func (p *ScanningPipeline) Scan(ctx context.Context, input port.ScanInput) (*valueobject.ScanResult, error) {
+	finalResult := valueobject.NewCleanScanResult()
+
+	for _, scanner := range p.scanners {
+		if scanner == nil {
+			continue
+		}
+		res, err := scanner.Scan(ctx, input)
+		if err != nil {
+			return nil, fmt.Errorf("scanner failed: %w", err)
+		}
+		if res == nil {
+			continue
+		}
+
+		// Fail fast on Virus Reject, Spam Reject, or Greylist
+		if res.Verdict == valueobject.ScanVerdictVirusReject ||
+			res.Verdict == valueobject.ScanVerdictSpamReject ||
+			res.Verdict == valueobject.ScanVerdictGreylist {
+			return res, nil
+		}
+
+		if res.Verdict == valueobject.ScanVerdictSpamJunk {
+			finalResult.Verdict = valueobject.ScanVerdictSpamJunk
+		}
+
+		if res.Score != 0 {
+			finalResult.Score = res.Score
+			finalResult.RequiredScore = res.RequiredScore
+		}
+		for k, v := range res.HeadersToAdd {
+			finalResult.HeadersToAdd[k] = v
+		}
+	}
+
+	return finalResult, nil
+}

services/protocols/internal/application/usecase/scanning_pipeline_test.go

@@ -0,0 +1,51 @@
+package usecase
+
+import (
+	"context"
+	"testing"
+
+	"lambdamail/protocols/internal/application/port"
+	"lambdamail/protocols/internal/domain/valueobject"
+)
+
+type mockScanner struct {
+	res *valueobject.ScanResult
+	err error
+}
+
+func (m *mockScanner) Scan(_ context.Context, _ port.ScanInput) (*valueobject.ScanResult, error) {
+	return m.res, m.err
+}
+
+func TestScanningPipeline_VirusRejectsImmediately(t *testing.T) {
+	s1 := &mockScanner{res: valueobject.NewVirusScanResult("EICAR-Test-Signature")}
+	s2 := &mockScanner{res: valueobject.NewCleanScanResult()}
+
+	pipeline := NewScanningPipeline(s1, s2)
+	res, err := pipeline.Scan(context.Background(), port.ScanInput{})
+	if err != nil {
+		t.Fatalf("Scan: %v", err)
+	}
+
+	if res.Verdict != valueobject.ScanVerdictVirusReject || res.VirusName != "EICAR-Test-Signature" {
+		t.Errorf("unexpected verdict: %+v", res)
+	}
+}
+
+func TestScanningPipeline_AggregatesJunkVerdict(t *testing.T) {
+	s1 := &mockScanner{res: valueobject.NewCleanScanResult()}
+	junkRes := valueobject.NewCleanScanResult()
+	junkRes.Verdict = valueobject.ScanVerdictSpamJunk
+	junkRes.Score = 7.5
+	s2 := &mockScanner{res: junkRes}
+
+	pipeline := NewScanningPipeline(s1, s2)
+	res, err := pipeline.Scan(context.Background(), port.ScanInput{})
+	if err != nil {
+		t.Fatalf("Scan: %v", err)
+	}
+
+	if res.Verdict != valueobject.ScanVerdictSpamJunk || res.Score != 7.5 {
+		t.Errorf("unexpected junk result: %+v", res)
+	}
+}