Add end-to-end tests for TLS RPT report ingestion and WebSocket event handling

- Implemented tests for ingesting TLS RPT reports from emails and ensuring they are stored correctly in the database.
- Added tests to verify that the same report can be ingested multiple times without duplication.
- Created tests for backfilling reports that were previously delivered to the inbox.
- Developed WebSocket event handling tests to ensure events are correctly routed to subscribers based on mailbox ownership.
- Included tests for the mail API to validate session handling, mailbox isolation, and error mapping.
- Ensured that the API rejects unauthorized requests and handles malformed JSON gracefully.
This commit is contained in:
Lucas Rayan committed 2026-08-21 09:55:37 -03:00
1 parent 880dced9e4
commit f330c4324e
28 files changed
+2438 -6

No files matched your search

@@ -0,0 +1,156 @@
package postgres
import (
"context"
"fmt"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgxpool"
"lambdamail/protocols/internal/application/port"
)
// DeliveredReportStore reaches reports that were delivered before automatic
// ingestion existed and are still filed in a mailbox.
type DeliveredReportStore struct {
pool *pgxpool.Pool
}
func NewDeliveredReportStore(pool *pgxpool.Pool) *DeliveredReportStore {
return &DeliveredReportStore{pool: pool}
}
// reportRecipientPattern matches the local parts the report aliases use. It
// mirrors reportLocalParts in the use case; both exist because this one has to
// be expressible in SQL.
const reportRecipientPattern = `^(dmarc|tlsrpt)@`
// ListUningestedReports returns stored messages addressed to a report alias
// that are not already filed in Reports.
//
// Being outside Reports is what "not yet ingested" means here: the delivery
// path files every report it reads into that folder, so anything addressed to
// a report alias and still sitting elsewhere predates ingestion.
func (s *DeliveredReportStore) ListUningestedReports(
ctx context.Context, limit int,
) ([]port.StoredReportMessage, error) {
if limit <= 0 || limit > 5000 {
limit = 5000
}
rows, err := s.pool.Query(ctx, `
SELECT e.id, e.blob_id, recipient
FROM email_messages e
JOIN folders f ON f.id = e.folder_id
CROSS JOIN LATERAL unnest(e.recipient_addresses) AS recipient
WHERE e.expunged_at IS NULL
AND LOWER(f.name) <> LOWER($1)
AND recipient ~* $2
ORDER BY e.received_at
LIMIT $3
`, reportsFolderName, reportRecipientPattern, limit)
if err != nil {
return nil, fmt.Errorf("list delivered reports: %w", err)
}
defer rows.Close()
out := []port.StoredReportMessage{}
for rows.Next() {
var msg port.StoredReportMessage
if err := rows.Scan(&msg.MessageID, &msg.BlobID, &msg.RecipientAddress); err != nil {
return nil, err
}
out = append(out, msg)
}
return out, rows.Err()
}
// reportsFolderName is duplicated from the use case rather than imported: the
// infrastructure layer must not depend on the application layer.
const reportsFolderName = "Reports"
// MoveToReportsFolder files a backfilled message out of the inbox, keeping the
// counters on both folders in step.
//
// The message keeps its identity and its blob; only the folder and the UID
// change, since a UID means nothing outside one folder.
func (s *DeliveredReportStore) MoveToReportsFolder(ctx context.Context, messageID uuid.UUID) error {
tx, err := s.pool.Begin(ctx)
if err != nil {
return err
}
defer tx.Rollback(ctx)
var mailboxID uuid.UUID
var sourceFolderID string
var wasUnread bool
err = tx.QueryRow(ctx, `
SELECT e.mailbox_id, e.folder_id::text,
NOT EXISTS (SELECT 1 FROM message_flags mf
WHERE mf.message_id = e.id
AND mf.received_at = e.received_at
AND mf.flag = '\Seen')
FROM email_messages e
WHERE e.id = $1 AND e.expunged_at IS NULL
`, messageID).Scan(&mailboxID, &sourceFolderID, &wasUnread)
if err != nil {
return fmt.Errorf("find message %s: %w", messageID, err)
}
var targetFolderID string
var targetUID int64
err = tx.QueryRow(ctx, `
SELECT id::text, uid_next FROM folders
WHERE mailbox_id = $1 AND LOWER(name) = LOWER($2)
LIMIT 1
FOR UPDATE
`, mailboxID, reportsFolderName).Scan(&targetFolderID, &targetUID)
if err != nil {
return fmt.Errorf("find the %s folder for mailbox %s: %w", reportsFolderName, mailboxID, err)
}
if targetFolderID == sourceFolderID {
return nil
}
var modseq int64
if err := tx.QueryRow(ctx, `
UPDATE folders SET uid_next = uid_next + 1, highest_modseq = highest_modseq + 1,
total_count = total_count + 1
WHERE id = $1
RETURNING highest_modseq
`, targetFolderID).Scan(&modseq); err != nil {
return err
}
if _, err := tx.Exec(ctx, `
UPDATE email_messages SET folder_id = $2, uid = $3, modseq = $4 WHERE id = $1
`, messageID, targetFolderID, targetUID, modseq); err != nil {
return err
}
// A backfilled report is machine-read mail: marking it seen is what stops
// it raising an unread badge in its new home.
if _, err := tx.Exec(ctx, `
INSERT INTO message_flags (message_id, received_at, flag)
SELECT e.id, e.received_at, '\Seen' FROM email_messages e WHERE e.id = $1
ON CONFLICT DO NOTHING
`, messageID); err != nil {
return err
}
unreadDelta := 0
if wasUnread {
unreadDelta = -1
}
if _, err := tx.Exec(ctx, `
UPDATE folders
SET total_count = GREATEST(0, total_count - 1),
unread_count = GREATEST(0, unread_count + $2),
highest_modseq = highest_modseq + 1
WHERE id = $1
`, sourceFolderID, unreadDelta); err != nil {
return err
}
return tx.Commit(ctx)
}
@@ -0,0 +1,355 @@
package postgres
import (
"bytes"
"compress/gzip"
"context"
"crypto/sha256"
"encoding/base64"
"encoding/hex"
"io"
"os"
"path/filepath"
"strings"
"testing"
"time"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgxpool"
"lambdamail/protocols/internal/application/port"
"lambdamail/protocols/internal/application/usecase"
"lambdamail/protocols/internal/domain/entity"
"lambdamail/protocols/internal/infrastructure/diskstorage"
)
// End to end over a real database: a report arrives as mail, is parsed into
// the report tables, and is filed into Reports without raising an unread badge.
// The report Google actually sends, with the domain changed.
const liveTlsRptReport = `{"organization-name":"Google Inc.",` +
`"date-range":{"start-datetime":"2026-08-19T00:00:00Z","end-datetime":"2026-08-19T23:59:59Z"},` +
`"contact-info":"smtp-tls-reporting@google.com","report-id":"2026-08-19T00:00:00Z_example.test",` +
`"policies":[{"policy":{"policy-type":"no-policy-found","policy-domain":"example.test"},` +
`"summary":{"total-successful-session-count":2,"total-failure-session-count":0}}]}`
func reportMail(t *testing.T, filename string, content []byte) []byte {
t.Helper()
var gzBuf bytes.Buffer
gz := gzip.NewWriter(&gzBuf)
if _, err := gz.Write(content); err != nil {
t.Fatal(err)
}
if err := gz.Close(); err != nil {
t.Fatal(err)
}
encoded := base64.StdEncoding.EncodeToString(gzBuf.Bytes())
var b bytes.Buffer
b.WriteString("From: smtp-tls-reporting@google.com\r\n")
b.WriteString("Subject: Report Domain: example.test\r\n")
b.WriteString("MIME-Version: 1.0\r\n")
b.WriteString("Content-Type: multipart/mixed; boundary=\"sep\"\r\n\r\n")
b.WriteString("--sep\r\n")
b.WriteString("Content-Type: text/plain; charset=utf-8\r\n\r\nReport attached.\r\n")
b.WriteString("--sep\r\n")
b.WriteString("Content-Type: application/gzip; name=\"" + filename + "\"\r\n")
b.WriteString("Content-Disposition: attachment; filename=\"" + filename + "\"\r\n")
b.WriteString("Content-Transfer-Encoding: base64\r\n\r\n")
b.WriteString(encoded + "\r\n")
b.WriteString("--sep--\r\n")
return b.Bytes()
}
func TestDeliveredTlsRptReportIsIngestedAndFiled(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
mailboxID := seedMailbox(t, pool)
// The Reports folder migration 0008 adds for every mailbox.
if _, err := pool.Exec(ctx,
`INSERT INTO folders (mailbox_id, name, special_use) VALUES ($1, 'Reports', NULL)`,
uuid.MustParse(mailboxID)); err != nil {
t.Fatalf("seed Reports folder: %v", err)
}
reportRepo := NewReportRepository(pool)
inbound := usecase.NewProcessInboundEmailUseCase(
nil, &dbBlobStore{pool: pool, dir: t.TempDir()}, NewInboundMessageRepository(pool))
inbound.SetReportIngestor(
usecase.NewIngestDeliveredReportsUseCase(usecase.NewIngestReportsUseCase(reportRepo)))
reportID := "delivered-" + uuid.NewString()
payload := reportMail(t,
"google.com!example.test!1787097600!1787183999!001.json.gz",
[]byte(strings.Replace(liveTlsRptReport,
"2026-08-19T00:00:00Z_example.test", reportID, 1)))
t.Cleanup(func() {
_, _ = pool.Exec(context.Background(),
`DELETE FROM tls_rpt_reports WHERE report_id = $1`, reportID)
})
if err := inbound.Handle(ctx, usecase.ProcessInboundEmailInput{
Sender: "smtp-tls-reporting@google.com",
Recipients: []port.MailboxRecord{{ID: uuid.MustParse(mailboxID)}},
RecipientAddresses: []string{"tlsrpt@example.test"},
Body: bytes.NewReader(payload),
}); err != nil {
t.Fatalf("Handle: %v", err)
}
// The report reached the table the admin console reads.
var stored int
if err := pool.QueryRow(ctx,
`SELECT COUNT(*) FROM tls_rpt_reports WHERE report_id = $1`, reportID).Scan(&stored); err != nil {
t.Fatalf("count stored reports: %v", err)
}
if stored != 1 {
t.Errorf("the report was not stored: %d rows", stored)
}
// And the message went to Reports, silently.
webmail := NewWebmailRepository(pool)
folders, err := webmail.ListFolders(ctx, mailboxID)
if err != nil {
t.Fatalf("list folders: %v", err)
}
for _, f := range folders {
switch f.Name {
case "Reports":
if f.TotalCount != 1 {
t.Errorf("Reports holds %d messages, want 1", f.TotalCount)
}
if f.UnreadCount != 0 {
t.Errorf("Reports shows %d unread; a machine-read report raises no badge", f.UnreadCount)
}
case "INBOX":
if f.TotalCount != 0 {
t.Errorf("the report landed in the inbox: %d messages", f.TotalCount)
}
}
}
}
// dbBlobStore writes a real message_blobs row so the delivery's foreign keys
// hold, without needing the disk storage driver in a database test.
type dbBlobStore struct {
pool *pgxpool.Pool
// dir is where the bytes actually land, so the real disk reader can read
// them back. Pointing storage_path at a file that does not exist would
// leave the read path untested.
dir string
}
func (s *dbBlobStore) Store(ctx context.Context, r io.Reader) (port.BlobRef, error) {
body, err := io.ReadAll(r)
if err != nil {
return port.BlobRef{}, err
}
sum := sha256.Sum256(body)
digest := hex.EncodeToString(sum[:])
// Deduplicated on the digest, as the real storage drivers are: identical
// bytes are one blob, and a second delivery of the same report reuses it.
id := uuid.New()
path := filepath.Join(s.dir, id.String()+".eml")
if err := os.WriteFile(path, body, 0o600); err != nil {
return port.BlobRef{}, err
}
if err := s.pool.QueryRow(ctx, `
INSERT INTO message_blobs (id, content_sha256, storage_driver, storage_path, size_bytes, ref_count)
VALUES ($1, $2, 'local', $3, $4, 0)
ON CONFLICT (content_sha256) DO UPDATE SET storage_path = EXCLUDED.storage_path
RETURNING id`,
id, digest, path, len(body)).Scan(&id); err != nil {
return port.BlobRef{}, err
}
return port.BlobRef{ID: id, SHA256: digest, SizeBytes: int64(len(body))}, nil
}
// A reporter that retries - or an SMTP retry after a slow commit - delivers
// the same report twice. Automatic ingestion makes that routine, where before
// reports were only ever written by hand through the API, so the same report
// arriving again must not double every number the console shows.
func TestIngestingTheSameReportTwiceStoresItOnce(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
repo := NewReportRepository(pool)
reportID := "retry-" + uuid.NewString()
report := func() *entity.TlsRptReport {
return &entity.TlsRptReport{
OrganizationName: "Google Inc.",
ReportID: reportID,
Domain: "example.test",
DateRangeBegin: time.Now().Add(-24 * time.Hour),
DateRangeEnd: time.Now(),
Policies: []entity.TlsRptPolicy{
{PolicyType: "no-policy-found", SuccessCount: 2, FailureCount: 0},
},
}
}
t.Cleanup(func() {
_, _ = pool.Exec(context.Background(),
`DELETE FROM tls_rpt_reports WHERE report_id = $1`, reportID)
})
if err := repo.SaveTlsRptReport(ctx, report()); err != nil {
t.Fatalf("first save: %v", err)
}
if err := repo.SaveTlsRptReport(ctx, report()); err != nil {
t.Fatalf("a redelivered report must not error: %v", err)
}
var headers, policies int
if err := pool.QueryRow(ctx,
`SELECT COUNT(*) FROM tls_rpt_reports WHERE report_id = $1`, reportID).Scan(&headers); err != nil {
t.Fatal(err)
}
if headers != 1 {
t.Errorf("the same report is stored %d times", headers)
}
if err := pool.QueryRow(ctx, `
SELECT COUNT(*) FROM tls_rpt_report_policies p
JOIN tls_rpt_reports r ON r.id = p.report_id
WHERE r.report_id = $1`, reportID).Scan(&policies); err != nil {
t.Fatal(err)
}
if policies != 1 {
t.Errorf("the report's policies are stored %d times", policies)
}
}
// The same, for DMARC: its unique key includes domain_id, which this path
// never fills, and in Postgres a NULL never conflicts - so the key it had did
// not in fact prevent a duplicate.
func TestIngestingTheSameDmarcReportTwiceStoresItOnce(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
repo := NewReportRepository(pool)
reportID := "retry-" + uuid.NewString()
report := func() *entity.DmarcReport {
return &entity.DmarcReport{
OrgName: "google.com",
ReportID: reportID,
Domain: "example.test",
DateRangeBegin: time.Now().Add(-24 * time.Hour),
DateRangeEnd: time.Now(),
Records: []entity.DmarcRecord{
{SourceIP: "203.0.113.10", Count: 3, Disposition: "none", DKIMResult: "pass", SPFResult: "pass"},
},
}
}
t.Cleanup(func() {
_, _ = pool.Exec(context.Background(),
`DELETE FROM dmarc_reports WHERE report_id = $1`, reportID)
})
if err := repo.SaveDmarcReport(ctx, report()); err != nil {
t.Fatalf("first save: %v", err)
}
if err := repo.SaveDmarcReport(ctx, report()); err != nil {
t.Fatalf("a redelivered report must not error: %v", err)
}
var headers int
if err := pool.QueryRow(ctx,
`SELECT COUNT(*) FROM dmarc_reports WHERE report_id = $1`, reportID).Scan(&headers); err != nil {
t.Fatal(err)
}
if headers != 1 {
t.Errorf("the same DMARC report is stored %d times", headers)
}
}
// The backfill, over a real database: a report that was delivered to the
// inbox before ingestion existed is parsed after the fact and filed away.
func TestBackfillParsesAReportAlreadySittingInTheInbox(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
mailboxID := seedMailbox(t, pool)
if _, err := pool.Exec(ctx,
`INSERT INTO folders (mailbox_id, name, special_use) VALUES ($1, 'Reports', NULL)`,
uuid.MustParse(mailboxID)); err != nil {
t.Fatalf("seed Reports folder: %v", err)
}
// Delivered the old way: straight into the inbox, unread, never parsed.
reportID := "backfill-" + uuid.NewString()
body := `{"organization-name":"Google Inc.",` +
`"date-range":{"start-datetime":"2026-08-19T00:00:00Z","end-datetime":"2026-08-19T23:59:59Z"},` +
`"report-id":"` + reportID + `",` +
`"policies":[{"policy":{"policy-type":"sts","policy-domain":"example.test"},` +
`"summary":{"total-successful-session-count":5,"total-failure-session-count":1}}]}`
payload := reportMail(t, "google.com!example.test!1!2!001.json.gz", []byte(body))
blobs := &dbBlobStore{pool: pool, dir: t.TempDir()}
blob, err := blobs.Store(ctx, bytes.NewReader(payload))
if err != nil {
t.Fatalf("store blob: %v", err)
}
uids, err := NewInboundMessageRepository(pool).PersistAll(ctx, []port.PersistInboundMessageInput{{
MailboxID: uuid.MustParse(mailboxID),
Blob: blob,
SenderAddress: "smtp-tls-reporting@google.com",
RecipientAddress: "tlsrpt@example.test",
TargetFolderName: "INBOX",
Subject: "Report Domain: example.test",
}})
if err != nil || len(uids) != 1 {
t.Fatalf("seed delivered report: %v", err)
}
t.Cleanup(func() {
_, _ = pool.Exec(context.Background(),
`DELETE FROM tls_rpt_reports WHERE report_id = $1`, reportID)
})
store := NewDeliveredReportStore(pool)
uc := usecase.NewBackfillReportsUseCase(store, diskstorage.NewLocalDiskBlobReader(pool),
usecase.NewIngestDeliveredReportsUseCase(
usecase.NewIngestReportsUseCase(NewReportRepository(pool))))
summary, err := uc.Run(ctx, 0)
if err != nil {
t.Fatalf("backfill: %v", err)
}
if summary.Ingested < 1 {
t.Fatalf("the backlogged report was not ingested: %+v", summary)
}
var stored int
if err := pool.QueryRow(ctx,
`SELECT COUNT(*) FROM tls_rpt_reports WHERE report_id = $1`, reportID).Scan(&stored); err != nil {
t.Fatal(err)
}
if stored != 1 {
t.Errorf("the report reached the table %d times, want 1", stored)
}
// And it left the inbox.
webmail := NewWebmailRepository(pool)
folders, _ := webmail.ListFolders(ctx, mailboxID)
for _, f := range folders {
if f.Name == "INBOX" && f.TotalCount != 0 {
t.Errorf("the backfilled report is still in the inbox: %d messages", f.TotalCount)
}
if f.Name == "Reports" {
if f.TotalCount != 1 {
t.Errorf("Reports holds %d messages, want 1", f.TotalCount)
}
if f.UnreadCount != 0 {
t.Errorf("the backfilled report raised an unread badge: %d", f.UnreadCount)
}
}
}
// Running it again finds nothing left to do.
second, err := uc.Run(ctx, 0)
if err != nil {
t.Fatalf("second run: %v", err)
}
if second.Ingested != 0 {
t.Errorf("a second pass re-ingested %d reports; it should find none", second.Ingested)
}
}
@@ -25,16 +25,35 @@ func (r *ReportRepository) SaveDmarcReport(ctx context.Context, report *entity.D
}
defer tx.Rollback(ctx)
// Idempotent on (org_name, report_id): the reporter's own identifier for
// the report. Reports are now parsed out of inbound mail automatically, so
// a redelivery - an SMTP retry, or a reporter simply sending again - is
// routine, and storing the same report twice would double every number the
// admin console shows.
//
// DO UPDATE rather than DO NOTHING so the existing row's id comes back and
// the records below can be reconciled against it; DO NOTHING returns no
// row at all when it conflicts.
var reportID uuid.UUID
err = tx.QueryRow(ctx, `
INSERT INTO dmarc_reports (org_name, report_id, domain, date_range_begin, date_range_end)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (org_name, report_id)
DO UPDATE SET domain = EXCLUDED.domain
RETURNING id
`, report.OrgName, report.ReportID, report.Domain, report.DateRangeBegin, report.DateRangeEnd).Scan(&reportID)
if err != nil {
return fmt.Errorf("insert dmarc report header: %w", err)
}
// The records are replaced, not appended: on a redelivery the report's
// previous rows are still here, and adding another set would double every
// count derived from them.
if _, err := tx.Exec(ctx,
`DELETE FROM dmarc_report_records WHERE report_id = $1`, reportID); err != nil {
return fmt.Errorf("clear dmarc records: %w", err)
}
for _, rec := range report.Records {
_, err := tx.Exec(ctx, `
INSERT INTO dmarc_report_records (report_id, source_ip, count, disposition, dkim_result, spf_result, header_from)
@@ -55,16 +74,24 @@ func (r *ReportRepository) SaveTlsRptReport(ctx context.Context, report *entity.
}
defer tx.Rollback(ctx)
// Idempotent for the same reason as the DMARC writer above.
var reportID uuid.UUID
err = tx.QueryRow(ctx, `
INSERT INTO tls_rpt_reports (organization_name, report_id, domain, date_range_begin, date_range_end)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (organization_name, report_id)
DO UPDATE SET domain = EXCLUDED.domain
RETURNING id
`, report.OrganizationName, report.ReportID, report.Domain, report.DateRangeBegin, report.DateRangeEnd).Scan(&reportID)
if err != nil {
return fmt.Errorf("insert tls-rpt report header: %w", err)
}
if _, err := tx.Exec(ctx,
`DELETE FROM tls_rpt_report_policies WHERE report_id = $1`, reportID); err != nil {
return fmt.Errorf("clear tls-rpt policies: %w", err)
}
for _, pol := range report.Policies {
_, err := tx.Exec(ctx, `
INSERT INTO tls_rpt_report_policies (report_id, policy_type, success_count, failure_count)