From 54d1ea5810ac0bcd059eca788e77549d60308f15 Mon Sep 17 00:00:00 2001 From: Lucas Rayan Date: Sat, 25 Jul 2026 10:25:52 -0300 Subject: [PATCH] #12: implement Antispam & Antivirus scanning pipeline (Rspamd & ClamAV) --- .../cmd/lambdamail-protocols/main.go | 18 ++ .../application/port/content_scanner.go | 19 ++ .../port/inbound_message_repository.go | 1 + .../usecase/process_inbound_email.go | 46 +++- .../application/usecase/scanning_pipeline.go | 55 +++++ .../usecase/scanning_pipeline_test.go | 51 ++++ .../domain/valueobject/scan_result.go | 38 +++ .../domain/valueobject/scan_result_test.go | 19 ++ .../e2e/antispam_antivirus_e2e_test.go | 218 ++++++++++++++++++ .../infrastructure/clamav/clamav_adapter.go | 106 +++++++++ .../clamav/clamav_adapter_test.go | 70 ++++++ .../postgres/inbound_message_repository.go | 13 +- .../infrastructure/rspamd/rspamd_adapter.go | 102 ++++++++ .../rspamd/rspamd_adapter_test.go | 54 +++++ .../internal/presentation/smtp/backend.go | 20 ++ 15 files changed, 825 insertions(+), 5 deletions(-) create mode 100644 services/protocols/internal/application/port/content_scanner.go create mode 100644 services/protocols/internal/application/usecase/scanning_pipeline.go create mode 100644 services/protocols/internal/application/usecase/scanning_pipeline_test.go create mode 100644 services/protocols/internal/domain/valueobject/scan_result.go create mode 100644 services/protocols/internal/domain/valueobject/scan_result_test.go create mode 100644 services/protocols/internal/e2e/antispam_antivirus_e2e_test.go create mode 100644 services/protocols/internal/infrastructure/clamav/clamav_adapter.go create mode 100644 services/protocols/internal/infrastructure/clamav/clamav_adapter_test.go create mode 100644 services/protocols/internal/infrastructure/rspamd/rspamd_adapter.go create mode 100644 services/protocols/internal/infrastructure/rspamd/rspamd_adapter_test.go diff --git a/services/protocols/cmd/lambdamail-protocols/main.go b/services/protocols/cmd/lambdamail-protocols/main.go index a37d2a2..9affdc1 100644 --- a/services/protocols/cmd/lambdamail-protocols/main.go +++ b/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"), diff --git a/services/protocols/internal/application/port/content_scanner.go b/services/protocols/internal/application/port/content_scanner.go new file mode 100644 index 0000000..33398e5 --- /dev/null +++ b/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) +} diff --git a/services/protocols/internal/application/port/inbound_message_repository.go b/services/protocols/internal/application/port/inbound_message_repository.go index 2dbf721..408e0bd 100644 --- a/services/protocols/internal/application/port/inbound_message_repository.go +++ b/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 diff --git a/services/protocols/internal/application/usecase/process_inbound_email.go b/services/protocols/internal/application/usecase/process_inbound_email.go index 4ef68c7..33c55fc 100644 --- a/services/protocols/internal/application/usecase/process_inbound_email.go +++ b/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) } diff --git a/services/protocols/internal/application/usecase/scanning_pipeline.go b/services/protocols/internal/application/usecase/scanning_pipeline.go new file mode 100644 index 0000000..7abf288 --- /dev/null +++ b/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 +} diff --git a/services/protocols/internal/application/usecase/scanning_pipeline_test.go b/services/protocols/internal/application/usecase/scanning_pipeline_test.go new file mode 100644 index 0000000..47fe3e0 --- /dev/null +++ b/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) + } +} diff --git a/services/protocols/internal/domain/valueobject/scan_result.go b/services/protocols/internal/domain/valueobject/scan_result.go new file mode 100644 index 0000000..2ee6924 --- /dev/null +++ b/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), + } +} diff --git a/services/protocols/internal/domain/valueobject/scan_result_test.go b/services/protocols/internal/domain/valueobject/scan_result_test.go new file mode 100644 index 0000000..7a59407 --- /dev/null +++ b/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) + } +} diff --git a/services/protocols/internal/e2e/antispam_antivirus_e2e_test.go b/services/protocols/internal/e2e/antispam_antivirus_e2e_test.go new file mode 100644 index 0000000..25915b0 --- /dev/null +++ b/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)) +} diff --git a/services/protocols/internal/infrastructure/clamav/clamav_adapter.go b/services/protocols/internal/infrastructure/clamav/clamav_adapter.go new file mode 100644 index 0000000..6f306d4 --- /dev/null +++ b/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) +} diff --git a/services/protocols/internal/infrastructure/clamav/clamav_adapter_test.go b/services/protocols/internal/infrastructure/clamav/clamav_adapter_test.go new file mode 100644 index 0000000..e67c8de --- /dev/null +++ b/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) + } +} diff --git a/services/protocols/internal/infrastructure/postgres/inbound_message_repository.go b/services/protocols/internal/infrastructure/postgres/inbound_message_repository.go index 9ed7761..b32f37a 100644 --- a/services/protocols/internal/infrastructure/postgres/inbound_message_repository.go +++ b/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 diff --git a/services/protocols/internal/infrastructure/rspamd/rspamd_adapter.go b/services/protocols/internal/infrastructure/rspamd/rspamd_adapter.go new file mode 100644 index 0000000..de968ea --- /dev/null +++ b/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 +} diff --git a/services/protocols/internal/infrastructure/rspamd/rspamd_adapter_test.go b/services/protocols/internal/infrastructure/rspamd/rspamd_adapter_test.go new file mode 100644 index 0000000..e411937 --- /dev/null +++ b/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) + } + }) + } +} diff --git a/services/protocols/internal/presentation/smtp/backend.go b/services/protocols/internal/presentation/smtp/backend.go index c7e2712..e229442 100644 --- a/services/protocols/internal/presentation/smtp/backend.go +++ b/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},