#12: merge feature/12-antispam-antivirus-pipeline into main

services/protocols/cmd/lambdamail-protocols/main.go

@@ -12,9 +12,12 @@ import (
 	"github.com/emersion/go-imap/v2/imapserver"
 	gosmtp "github.com/emersion/go-smtp"
 
+	"lambdamail/protocols/internal/application/port"
 	"lambdamail/protocols/internal/application/usecase"
+	"lambdamail/protocols/internal/infrastructure/clamav"
 	"lambdamail/protocols/internal/infrastructure/diskstorage"
 	"lambdamail/protocols/internal/infrastructure/postgres"
+	"lambdamail/protocols/internal/infrastructure/rspamd"
 	tlsprovider "lambdamail/protocols/internal/infrastructure/tls"
 	imappresentation "lambdamail/protocols/internal/presentation/imap"
 	httppresentation "lambdamail/protocols/internal/presentation/http"
@@ -42,6 +45,21 @@ func main() {
 	messages := postgres.NewInboundMessageRepository(pool)
 	useCase := usecase.NewProcessInboundEmailUseCase(mailboxes, blobs, messages)
 
+	clamavAddr := os.Getenv("CLAMAV_ADDR")
+	rspamdURL := os.Getenv("RSPAMD_URL")
+	if clamavAddr != "" || rspamdURL != "" {
+		var scanners []port.ContentScanner
+		if clamavAddr != "" {
+			scanners = append(scanners, clamav.NewClamAVAdapter(clamavAddr))
+		}
+		if rspamdURL != "" {
+			scanners = append(scanners, rspamd.NewRspamdAdapter(rspamdURL))
+		}
+		pipeline := usecase.NewScanningPipeline(scanners...)
+		useCase.SetScanner(pipeline)
+		log.Printf("lambdamail-protocols antispam/antivirus scanner enabled (clamav: %s, rspamd: %s)", clamavAddr, rspamdURL)
+	}
+
 	certProvider, err := tlsprovider.NewLocalFileCertProvider(
 		os.Getenv("PROTOCOLS_CERT_PATH"),
 		os.Getenv("PROTOCOLS_KEY_PATH"),

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)
+	}
+}

services/protocols/internal/domain/valueobject/scan_result.go

@@ -0,0 +1,38 @@
+package valueobject
+
+type ScanVerdict string
+
+const (
+	ScanVerdictClean       ScanVerdict = "CLEAN"
+	ScanVerdictSpamJunk    ScanVerdict = "SPAM_JUNK"
+	ScanVerdictSpamReject  ScanVerdict = "SPAM_REJECT"
+	ScanVerdictGreylist    ScanVerdict = "GREYLIST"
+	ScanVerdictVirusReject ScanVerdict = "VIRUS_REJECT"
+)
+
+type ScanResult struct {
+	Verdict       ScanVerdict
+	Score         float64
+	RequiredScore float64
+	Action        string
+	VirusName     string
+	Symbols       map[string]float64
+	HeadersToAdd  map[string]string
+}
+
+func NewCleanScanResult() *ScanResult {
+	return &ScanResult{
+		Verdict:      ScanVerdictClean,
+		Symbols:      make(map[string]float64),
+		HeadersToAdd: make(map[string]string),
+	}
+}
+
+func NewVirusScanResult(virusName string) *ScanResult {
+	return &ScanResult{
+		Verdict:      ScanVerdictVirusReject,
+		VirusName:    virusName,
+		Symbols:      make(map[string]float64),
+		HeadersToAdd: make(map[string]string),
+	}
+}

services/protocols/internal/domain/valueobject/scan_result_test.go

@@ -0,0 +1,19 @@
+package valueobject
+
+import (
+	"testing"
+)
+
+func TestNewCleanScanResult(t *testing.T) {
+	res := NewCleanScanResult()
+	if res.Verdict != ScanVerdictClean {
+		t.Errorf("verdict = %s, want CLEAN", res.Verdict)
+	}
+}
+
+func TestNewVirusScanResult(t *testing.T) {
+	res := NewVirusScanResult("EICAR-Test-Signature")
+	if res.Verdict != ScanVerdictVirusReject || res.VirusName != "EICAR-Test-Signature" {
+		t.Errorf("unexpected virus scan result: %+v", res)
+	}
+}

services/protocols/internal/e2e/antispam_antivirus_e2e_test.go

@@ -0,0 +1,218 @@
+package e2e
+
+import (
+	"context"
+	"encoding/json"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	embeddedpostgres "github.com/fergusstrange/embedded-postgres"
+	gosmtp "github.com/emersion/go-smtp"
+	"github.com/google/uuid"
+
+	appusecase "lambdamail/protocols/internal/application/usecase"
+	"lambdamail/protocols/internal/infrastructure/clamav"
+	"lambdamail/protocols/internal/infrastructure/diskstorage"
+	"lambdamail/protocols/internal/infrastructure/postgres"
+	"lambdamail/protocols/internal/infrastructure/rspamd"
+	smtppresentation "lambdamail/protocols/internal/presentation/smtp"
+)
+
+func TestAntispamAntivirusEndToEnd(t *testing.T) {
+	if testing.Short() {
+		t.Skip("e2e test skipped in -short mode")
+	}
+	ctx := context.Background()
+
+	runtimeDir := t.TempDir()
+	pg := embeddedpostgres.NewDatabase(embeddedpostgres.DefaultConfig().
+		Port(54338).
+		RuntimePath(runtimeDir).
+		StartTimeout(120 * time.Second))
+	if err := pg.Start(); err != nil {
+		t.Fatalf("start embedded postgres: %v", err)
+	}
+	defer pg.Stop()
+
+	dbURL := "postgres://postgres:postgres@localhost:54338/postgres?sslmode=disable"
+	pool, err := postgres.NewPool(ctx, dbURL)
+	if err != nil {
+		t.Fatalf("connect pool: %v", err)
+	}
+	defer pool.Close()
+
+	root := repoRoot(t)
+	migrations := []string{
+		"0001_init_schema.up.sql",
+		"0002_add_is_system_to_aliases.up.sql",
+		"0003_create_report_tables.up.sql",
+	}
+
+	for _, m := range migrations {
+		sql, err := os.ReadFile(filepath.Join(root, "migrations", m))
+		if err != nil {
+			t.Fatalf("read migration %s: %v", m, err)
+		}
+		if _, err := pool.Exec(ctx, string(sql)); err != nil {
+			t.Fatalf("apply migration %s: %v", m, err)
+		}
+	}
+
+	// Seed domain, mailbox, INBOX and Junk folders
+	domainID := uuid.New()
+	mailboxID := uuid.New()
+	inboxID := uuid.New()
+	junkID := uuid.New()
+
+	pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name) VALUES ($1, 'scan.test', 'scan.test')`, domainID)
+	pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash) VALUES ($1, $2, 'user', 'user@scan.test', 'hash')`, mailboxID, domainID)
+	pool.Exec(ctx, `INSERT INTO folders (id, mailbox_id, name, special_use) VALUES ($1, $2, 'INBOX', 'inbox')`, inboxID, mailboxID)
+	pool.Exec(ctx, `INSERT INTO folders (id, mailbox_id, name, special_use) VALUES ($1, $2, 'Junk', 'junk')`, junkID, mailboxID)
+
+	mailboxRepo := postgres.NewMailboxRepository(pool)
+	messageRepo := postgres.NewInboundMessageRepository(pool)
+	spool := t.TempDir()
+	blobStorage := diskstorage.NewLocalDiskBlobStorage(pool, spool)
+
+	processInboundUC := appusecase.NewProcessInboundEmailUseCase(mailboxRepo, blobStorage, messageRepo)
+
+	// Mock ClamAV server state
+	var clamavInfected bool
+
+	clamavLn, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen mock clamav: %v", err)
+	}
+	defer clamavLn.Close()
+
+	go func() {
+		for {
+			conn, err := clamavLn.Accept()
+			if err != nil {
+				return
+			}
+			buf := make([]byte, 4096)
+			for {
+				_, err := conn.Read(buf)
+				if err != nil {
+					break
+				}
+			}
+			if clamavInfected {
+				conn.Write([]byte("stream: EICAR-Test-Signature FOUND\x00"))
+			} else {
+				conn.Write([]byte("stream: OK\x00"))
+			}
+			conn.Close()
+		}
+	}()
+
+	// Mock Rspamd server state
+	var rspamdAction string = "no action"
+
+	rspamdTS := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		resp := map[string]interface{}{
+			"action":         rspamdAction,
+			"score":          5.0,
+			"required_score": 15.0,
+			"symbols":        map[string]interface{}{},
+		}
+		json.NewEncoder(w).Encode(resp)
+	}))
+	defer rspamdTS.Close()
+
+	clamavAdapter := clamav.NewClamAVAdapter(clamavLn.Addr().String())
+	rspamdAdapter := rspamd.NewRspamdAdapter(rspamdTS.URL)
+
+	pipeline := appusecase.NewScanningPipeline(clamavAdapter, rspamdAdapter)
+	processInboundUC.SetScanner(pipeline)
+
+	smtpServer := gosmtp.NewServer(smtppresentation.NewBackend(processInboundUC))
+	smtpServer.Addr = "127.0.0.1:25026"
+	smtpServer.Domain = "scan.test"
+	smtpServer.AllowInsecureAuth = true
+
+	go smtpServer.ListenAndServe()
+	defer smtpServer.Close()
+
+	time.Sleep(100 * time.Millisecond)
+
+	// 1. Clean message -> delivered to INBOX
+	clamavInfected = false
+	rspamdAction = "no action"
+	err = sendTestSMTPMessage("127.0.0.1:25026", "sender@external.test", "user@scan.test", "Clean Email Body")
+	if err != nil {
+		t.Fatalf("clean email delivery failed: %v", err)
+	}
+
+	var inboxCount int
+	pool.QueryRow(ctx, `SELECT total_count FROM folders WHERE id = $1`, inboxID).Scan(&inboxCount)
+	if inboxCount != 1 {
+		t.Errorf("expected 1 message in INBOX, got %d", inboxCount)
+	}
+
+	// 2. Virus message -> rejected with 554 5.7.1
+	clamavInfected = true
+	err = sendTestSMTPMessage("127.0.0.1:25026", "sender@external.test", "user@scan.test", "X5O!P%@AP[4\\PZX54(P^)7CC)7}$EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*")
+	if err == nil || !strings.Contains(err.Error(), "554 5.7.1 Virus detected") {
+		t.Fatalf("expected 554 virus rejection, got: %v", err)
+	}
+
+	// 3. Spam message with action=add header -> delivered to Junk folder
+	clamavInfected = false
+	rspamdAction = "add header"
+	err = sendTestSMTPMessage("127.0.0.1:25026", "sender@external.test", "user@scan.test", "Spammy Email Body")
+	if err != nil {
+		t.Fatalf("junk email delivery failed: %v", err)
+	}
+
+	var junkCount int
+	pool.QueryRow(ctx, `SELECT total_count FROM folders WHERE id = $1`, junkID).Scan(&junkCount)
+	if junkCount != 1 {
+		t.Errorf("expected 1 message in Junk folder, got %d", junkCount)
+	}
+}
+
+func sendTestSMTPMessage(addr, sender, rcpt, body string) error {
+	conn, err := net.Dial("tcp", addr)
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+
+	buf := make([]byte, 1024)
+	conn.Read(buf) // greeting
+
+	conn.Write([]byte("HELO test.sender\r\n"))
+	conn.Read(buf)
+
+	conn.Write([]byte("MAIL FROM:<" + sender + ">\r\n"))
+	conn.Read(buf)
+
+	conn.Write([]byte("RCPT TO:<" + rcpt + ">\r\n"))
+	conn.Read(buf)
+
+	conn.Write([]byte("DATA\r\n"))
+	conn.Read(buf)
+
+	msg := "From: " + sender + "\r\nTo: " + rcpt + "\r\nSubject: Test\r\n\r\n" + body + "\r\n.\r\n"
+	conn.Write([]byte(msg))
+
+	n, _ := conn.Read(buf)
+	respStr := string(buf[:n])
+
+	if !strings.HasPrefix(respStr, "250") {
+		return fmtError(respStr)
+	}
+	return nil
+}
+
+func fmtError(msg string) error {
+	return net.UnknownNetworkError(strings.TrimSpace(msg))
+}

services/protocols/internal/infrastructure/clamav/clamav_adapter.go

@@ -0,0 +1,106 @@
+package clamav
+
+import (
+	"bufio"
+	"context"
+	"encoding/binary"
+	"fmt"
+	"io"
+	"net"
+	"strings"
+	"time"
+
+	"lambdamail/protocols/internal/application/port"
+	"lambdamail/protocols/internal/domain/valueobject"
+)
+
+type ClamAVAdapter struct {
+	address string
+	timeout time.Duration
+}
+
+func NewClamAVAdapter(address string) *ClamAVAdapter {
+	if address == "" {
+		address = "localhost:3310"
+	}
+	return &ClamAVAdapter{
+		address: address,
+		timeout: 10 * time.Second,
+	}
+}
+
+func (a *ClamAVAdapter) Scan(ctx context.Context, input port.ScanInput) (*valueobject.ScanResult, error) {
+	d := net.Dialer{Timeout: a.timeout}
+	conn, err := d.DialContext(ctx, "tcp", a.address)
+	if err != nil {
+		return nil, fmt.Errorf("connect to clamd at %s: %w", a.address, err)
+	}
+	defer conn.Close()
+
+	if deadline, ok := ctx.Deadline(); ok {
+		conn.SetDeadline(deadline)
+	} else {
+		conn.SetDeadline(time.Now().Add(a.timeout))
+	}
+
+	// Send zINSTREAM command
+	if _, err := conn.Write([]byte("zINSTREAM\x00")); err != nil {
+		return nil, fmt.Errorf("write zINSTREAM: %w", err)
+	}
+
+	// Stream payload in chunks of up to 4096 bytes
+	chunkSize := 4096
+	data := input.Payload
+	for i := 0; i < len(data); i += chunkSize {
+		end := i + chunkSize
+		if end > len(data) {
+			end = len(data)
+		}
+		chunk := data[i:end]
+
+		lenBuf := make([]byte, 4)
+		binary.BigEndian.PutUint32(lenBuf, uint32(len(chunk)))
+
+		if _, err := conn.Write(lenBuf); err != nil {
+			return nil, fmt.Errorf("write chunk length: %w", err)
+		}
+		if _, err := conn.Write(chunk); err != nil {
+			return nil, fmt.Errorf("write chunk data: %w", err)
+		}
+	}
+
+	// Write zero-length terminator chunk
+	if _, err := conn.Write([]byte{0, 0, 0, 0}); err != nil {
+		return nil, fmt.Errorf("write eof chunk: %w", err)
+	}
+
+	if tcpConn, ok := conn.(*net.TCPConn); ok {
+		tcpConn.CloseWrite()
+	}
+
+	// Read clamd response
+	reader := bufio.NewReader(conn)
+	respStr, err := reader.ReadString('\x00')
+	if err != nil && err != io.EOF {
+		return nil, fmt.Errorf("read clamd response: %w", err)
+	}
+	respStr = strings.TrimRight(respStr, "\x00\r\n")
+
+	if strings.HasSuffix(respStr, "OK") {
+		res := valueobject.NewCleanScanResult()
+		res.HeadersToAdd["X-Virus-Scanned"] = "ClamAV clean"
+		return res, nil
+	}
+
+	if strings.HasSuffix(respStr, "FOUND") {
+		// Response format: "stream: EICAR-Test-Signature FOUND"
+		parts := strings.Split(respStr, ":")
+		virusName := "Unknown"
+		if len(parts) >= 2 {
+			virusName = strings.TrimSuffix(strings.TrimSpace(parts[1]), " FOUND")
+		}
+		return valueobject.NewVirusScanResult(virusName), nil
+	}
+
+	return nil, fmt.Errorf("unexpected clamd response: %q", respStr)
+}

services/protocols/internal/infrastructure/clamav/clamav_adapter_test.go

@@ -0,0 +1,70 @@
+package clamav
+
+import (
+	"context"
+	"net"
+	"testing"
+
+	"lambdamail/protocols/internal/application/port"
+	"lambdamail/protocols/internal/domain/valueobject"
+)
+
+func TestClamAVAdapter_Scan_CleanAndVirus(t *testing.T) {
+	// Clean mock server
+	cleanLn, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen clean server: %v", err)
+	}
+	defer cleanLn.Close()
+
+	go func() {
+		conn, err := cleanLn.Accept()
+		if err != nil {
+			return
+		}
+		defer conn.Close()
+		buf := make([]byte, 4096)
+		for {
+			_, err := conn.Read(buf)
+			if err != nil {
+				break
+			}
+		}
+		conn.Write([]byte("stream: OK\x00"))
+	}()
+
+	adapter := NewClamAVAdapter(cleanLn.Addr().String())
+	res, err := adapter.Scan(context.Background(), port.ScanInput{Payload: []byte("clean content")})
+	if err != nil || res.Verdict != valueobject.ScanVerdictClean {
+		t.Fatalf("expected clean scan, got res=%+v, err=%v", res, err)
+	}
+
+	// Virus mock server
+	virusLn, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen virus server: %v", err)
+	}
+	defer virusLn.Close()
+
+	go func() {
+		conn, err := virusLn.Accept()
+		if err != nil {
+			return
+		}
+		defer conn.Close()
+		buf := make([]byte, 4096)
+		for {
+			_, err := conn.Read(buf)
+			if err != nil {
+				break
+			}
+		}
+		conn.Write([]byte("stream: EICAR-Test-Signature FOUND\x00"))
+	}()
+
+	virusAdapter := NewClamAVAdapter(virusLn.Addr().String())
+	res, err = virusAdapter.Scan(context.Background(), port.ScanInput{Payload: []byte("virus content")})
+	if err != nil || res.Verdict != valueobject.ScanVerdictVirusReject || res.VirusName != "EICAR-Test-Signature" {
+		t.Fatalf("expected virus scan, got res=%+v, err=%v", res, err)
+	}
+}

services/protocols/internal/infrastructure/postgres/inbound_message_repository.go

@@ -28,15 +28,22 @@ func (r *InboundMessageRepository) Persist(ctx context.Context, input port.Persi
 	}
 	defer tx.Rollback(ctx) // no-op if Commit succeeds
 
+	targetFolder := input.TargetFolderName
+	if targetFolder == "" {
+		targetFolder = "INBOX"
+	}
+
 	var folderID string
 	var uid int64
 	err = tx.QueryRow(ctx, `
 		SELECT id, uid_next FROM folders
-		WHERE mailbox_id = $1 AND special_use = 'inbox'
+		WHERE mailbox_id = $1 AND (special_use = LOWER($2) OR LOWER(name) = LOWER($2))
+		ORDER BY special_use IS NOT NULL DESC
+		LIMIT 1
 		FOR UPDATE
-	`, input.MailboxID).Scan(&folderID, &uid)
+	`, input.MailboxID, targetFolder).Scan(&folderID, &uid)
 	if err != nil {
-		return 0, fmt.Errorf("find inbox folder for mailbox %s: %w", input.MailboxID, err)
+		return 0, fmt.Errorf("find folder %q for mailbox %s: %w", targetFolder, input.MailboxID, err)
 	}
 
 	var modseq int64

services/protocols/internal/infrastructure/rspamd/rspamd_adapter.go

@@ -0,0 +1,102 @@
+package rspamd
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strings"
+
+	"lambdamail/protocols/internal/application/port"
+	"lambdamail/protocols/internal/domain/valueobject"
+)
+
+type RspamdResponse struct {
+	Action        string                         `json:"action"`
+	Score         float64                        `json:"score"`
+	RequiredScore float64                        `json:"required_score"`
+	Symbols       map[string]map[string]interface{} `json:"symbols"`
+}
+
+type RspamdAdapter struct {
+	baseURL    string
+	httpClient *http.Client
+}
+
+func NewRspamdAdapter(baseURL string) *RspamdAdapter {
+	if baseURL == "" {
+		baseURL = "http://localhost:11333"
+	}
+	return &RspamdAdapter{
+		baseURL:    strings.TrimSuffix(baseURL, "/"),
+		httpClient: &http.Client{},
+	}
+}
+
+func (a *RspamdAdapter) Scan(ctx context.Context, input port.ScanInput) (*valueobject.ScanResult, error) {
+	url := fmt.Sprintf("%s/checkv2", a.baseURL)
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(input.Payload))
+	if err != nil {
+		return nil, fmt.Errorf("create rspamd request: %w", err)
+	}
+
+	if input.ClientIP != "" {
+		req.Header.Set("IP", input.ClientIP)
+	}
+	if input.HeloDomain != "" {
+		req.Header.Set("Helo", input.HeloDomain)
+	}
+	if input.Sender != "" {
+		req.Header.Set("From", input.Sender)
+	}
+	if input.Recipient != "" {
+		req.Header.Set("Deliver-To", input.Recipient)
+	}
+
+	resp, err := a.httpClient.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("rspamd http post: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("rspamd returned HTTP %d", resp.StatusCode)
+	}
+
+	var parsed RspamdResponse
+	if err := json.NewDecoder(resp.Body).Decode(&parsed); err != nil {
+		return nil, fmt.Errorf("decode rspamd json: %w", err)
+	}
+
+	res := &valueobject.ScanResult{
+		Score:         parsed.Score,
+		RequiredScore: parsed.RequiredScore,
+		Action:        parsed.Action,
+		Symbols:       make(map[string]float64),
+		HeadersToAdd:  make(map[string]string),
+	}
+
+	for name, obj := range parsed.Symbols {
+		if scoreVal, ok := obj["score"].(float64); ok {
+			res.Symbols[name] = scoreVal
+		}
+	}
+
+	switch parsed.Action {
+	case "reject":
+		res.Verdict = valueobject.ScanVerdictSpamReject
+	case "greylist":
+		res.Verdict = valueobject.ScanVerdictGreylist
+	case "add header", "rewrite subject":
+		res.Verdict = valueobject.ScanVerdictSpamJunk
+		res.HeadersToAdd["X-Spam-Flag"] = "YES"
+		res.HeadersToAdd["X-Spam-Score"] = fmt.Sprintf("%.2f", parsed.Score)
+	default:
+		res.Verdict = valueobject.ScanVerdictClean
+		res.HeadersToAdd["X-Spam-Flag"] = "NO"
+		res.HeadersToAdd["X-Spam-Score"] = fmt.Sprintf("%.2f", parsed.Score)
+	}
+
+	return res, nil
+}

services/protocols/internal/infrastructure/rspamd/rspamd_adapter_test.go

@@ -0,0 +1,54 @@
+package rspamd
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"lambdamail/protocols/internal/application/port"
+	"lambdamail/protocols/internal/domain/valueobject"
+)
+
+func TestRspamdAdapter_Scan_MapsActions(t *testing.T) {
+	tests := []struct {
+		rspamdAction string
+		wantVerdict  valueobject.ScanVerdict
+	}{
+		{"no action", valueobject.ScanVerdictClean},
+		{"add header", valueobject.ScanVerdictSpamJunk},
+		{"greylist", valueobject.ScanVerdictGreylist},
+		{"reject", valueobject.ScanVerdictSpamReject},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.rspamdAction, func(t *testing.T) {
+			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				if r.URL.Path != "/checkv2" {
+					t.Fatalf("unexpected path: %s", r.URL.Path)
+				}
+				resp := RspamdResponse{
+					Action:        tt.rspamdAction,
+					Score:         5.0,
+					RequiredScore: 15.0,
+					Symbols: map[string]map[string]interface{}{
+						"SPF_PASS": {"score": -1.0},
+					},
+				}
+				json.NewEncoder(w).Encode(resp)
+			}))
+			defer ts.Close()
+
+			adapter := NewRspamdAdapter(ts.URL)
+			res, err := adapter.Scan(context.Background(), port.ScanInput{Payload: []byte("test email")})
+			if err != nil {
+				t.Fatalf("Scan: %v", err)
+			}
+
+			if res.Verdict != tt.wantVerdict {
+				t.Errorf("verdict = %s, want %s", res.Verdict, tt.wantVerdict)
+			}
+		})
+	}
+}

services/protocols/internal/presentation/smtp/backend.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"errors"
 	"io"
+	"strings"
 
 	gosmtp "github.com/emersion/go-smtp"
 
@@ -108,6 +109,25 @@ func (s *session) Data(r io.Reader) error {
 		Body:               r,
 	})
 	if err != nil {
+		var smtpErr *gosmtp.SMTPError
+		if errors.As(err, &smtpErr) {
+			return smtpErr
+		}
+		errMsg := err.Error()
+		if strings.HasPrefix(errMsg, "554 ") {
+			return &gosmtp.SMTPError{
+				Code:         554,
+				EnhancedCode: gosmtp.EnhancedCode{5, 7, 1},
+				Message:      strings.TrimPrefix(errMsg, "554 5.7.1 "),
+			}
+		}
+		if strings.HasPrefix(errMsg, "451 ") {
+			return &gosmtp.SMTPError{
+				Code:         451,
+				EnhancedCode: gosmtp.EnhancedCode{4, 7, 1},
+				Message:      strings.TrimPrefix(errMsg, "451 4.7.1 "),
+			}
+		}
 		return &gosmtp.SMTPError{
 			Code:         451,
 			EnhancedCode: gosmtp.EnhancedCode{4, 3, 0},