From bd0af98076277c87bf949cb3e25a8eb0b6e62978 Mon Sep 17 00:00:00 2001 From: Lucas Rayan Date: Sat, 1 Aug 2026 22:48:38 -0300 Subject: [PATCH] #17: implement DANE TLSA, MTA-STS policy resolution, ARC sealing and mail authentication --- .../0004_index_outbound_stale_locks.down.sql | 1 + .../0004_index_outbound_stale_locks.up.sql | 6 + .../internal/application/port/arc_sealer.go | 18 + .../internal/application/port/dns_verifier.go | 18 + .../application/port/mail_authenticator.go | 50 +++ .../application/port/tls_policy_resolver.go | 18 + .../usecase/process_outbound_email.go | 186 +++++++++++ .../usecase/process_outbound_email_test.go | 171 ++++++++++ .../internal/application/usecase/relay.go | 143 ++++++++ .../application/usecase/relay_test.go | 309 ++++++++++++++++++ .../application/usecase/rollover_tlsa.go | 246 ++++++++++++++ .../application/usecase/rollover_tlsa_test.go | 220 +++++++++++++ .../internal/infrastructure/arc/canonical.go | 147 +++++++++ .../internal/infrastructure/arc/sealer.go | 268 +++++++++++++++ .../infrastructure/arc/sealer_test.go | 258 +++++++++++++++ .../internal/infrastructure/arc/verifier.go | 135 ++++++++ .../infrastructure/mailauth/authenticator.go | 221 +++++++++++++ .../mailauth/authenticator_test.go | 243 ++++++++++++++ .../infrastructure/netdns/public_verifier.go | 106 ++++++ .../infrastructure/netdns/tlsa_resolver.go | 77 +++++ .../infrastructure/tlspolicy/resolver.go | 233 +++++++++++++ .../infrastructure/tlspolicy/resolver_test.go | 222 +++++++++++++ 22 files changed, 3296 insertions(+) create mode 100644 migrations/0004_index_outbound_stale_locks.down.sql create mode 100644 migrations/0004_index_outbound_stale_locks.up.sql create mode 100644 services/protocols/internal/application/port/arc_sealer.go create mode 100644 services/protocols/internal/application/port/dns_verifier.go create mode 100644 services/protocols/internal/application/port/mail_authenticator.go create mode 100644 services/protocols/internal/application/port/tls_policy_resolver.go create mode 100644 services/protocols/internal/application/usecase/process_outbound_email.go create mode 100644 services/protocols/internal/application/usecase/process_outbound_email_test.go create mode 100644 services/protocols/internal/application/usecase/relay.go create mode 100644 services/protocols/internal/application/usecase/relay_test.go create mode 100644 services/protocols/internal/application/usecase/rollover_tlsa.go create mode 100644 services/protocols/internal/application/usecase/rollover_tlsa_test.go create mode 100644 services/protocols/internal/infrastructure/arc/canonical.go create mode 100644 services/protocols/internal/infrastructure/arc/sealer.go create mode 100644 services/protocols/internal/infrastructure/arc/sealer_test.go create mode 100644 services/protocols/internal/infrastructure/arc/verifier.go create mode 100644 services/protocols/internal/infrastructure/mailauth/authenticator.go create mode 100644 services/protocols/internal/infrastructure/mailauth/authenticator_test.go create mode 100644 services/protocols/internal/infrastructure/netdns/public_verifier.go create mode 100644 services/protocols/internal/infrastructure/netdns/tlsa_resolver.go create mode 100644 services/protocols/internal/infrastructure/tlspolicy/resolver.go create mode 100644 services/protocols/internal/infrastructure/tlspolicy/resolver_test.go diff --git a/migrations/0004_index_outbound_stale_locks.down.sql b/migrations/0004_index_outbound_stale_locks.down.sql new file mode 100644 index 0000000..a3ac721 --- /dev/null +++ b/migrations/0004_index_outbound_stale_locks.down.sql @@ -0,0 +1 @@ +DROP INDEX IF EXISTS idx_outbound_stale_locks; diff --git a/migrations/0004_index_outbound_stale_locks.up.sql b/migrations/0004_index_outbound_stale_locks.up.sql new file mode 100644 index 0000000..ce6fea8 --- /dev/null +++ b/migrations/0004_index_outbound_stale_locks.up.sql @@ -0,0 +1,6 @@ +-- Delivery workers claim a job by moving it to SENDING with a lock deadline. +-- If the worker dies before releasing it, the job must be picked up again once +-- the lock expires; without this index that reclaim scan is a sequential scan +-- over every job ever queued. +CREATE INDEX idx_outbound_stale_locks ON outbound_jobs(locked_until) + WHERE status = 'SENDING'; diff --git a/services/protocols/internal/application/port/arc_sealer.go b/services/protocols/internal/application/port/arc_sealer.go new file mode 100644 index 0000000..8920d4d --- /dev/null +++ b/services/protocols/internal/application/port/arc_sealer.go @@ -0,0 +1,18 @@ +package port + +import "context" + +// ArcSealer adds an Authenticated Received Chain set to a message +// (RFC 8617, PLAN.md section 5). +// +// It matters on a host that passes mail onward: a forwarder that modifies a +// message breaks its DKIM signature, and the final receiver then sees a DMARC +// failure for mail that authenticated correctly at the first hop. The ARC set +// records this hop's verdict and seals it so the failure can be recognised as +// a forwarding artefact rather than a forgery. +type ArcSealer interface { + // Seal returns the message with an ARC set prepended. A chain that + // already failed is returned untouched: RFC 8617 section 5.1.2 forbids + // extending it. + Seal(ctx context.Context, message []byte, authResult InboundAuthResult) ([]byte, error) +} diff --git a/services/protocols/internal/application/port/dns_verifier.go b/services/protocols/internal/application/port/dns_verifier.go new file mode 100644 index 0000000..13e9ad8 --- /dev/null +++ b/services/protocols/internal/application/port/dns_verifier.go @@ -0,0 +1,18 @@ +package port + +import ( + "context" + + "lambdamail/protocols/internal/domain/entity" +) + +// DnsVerifier re-reads published records through resolvers outside our +// control. PLAN.md section 7.5 requires this step because the provider API +// confirming a write only proves the API accepted it, not that the world can +// see it. +type DnsVerifier interface { + // VerifyRecord reports whether the record is visible with the expected + // value. A lookup failure is reported as not-visible with the reason, + // never as a hard error, so one unreachable resolver cannot fail a sync. + VerifyRecord(ctx context.Context, record entity.DnsRecord) (visible bool, detail string) +} diff --git a/services/protocols/internal/application/port/mail_authenticator.go b/services/protocols/internal/application/port/mail_authenticator.go new file mode 100644 index 0000000..9182bf3 --- /dev/null +++ b/services/protocols/internal/application/port/mail_authenticator.go @@ -0,0 +1,50 @@ +package port + +import ( + "context" + "net" +) + +// Authentication result values as they appear in an Authentication-Results +// header (RFC 8601 section 2.7) and in the email_messages columns. +const ( + AuthResultNone = "none" + AuthResultPass = "pass" + AuthResultFail = "fail" + AuthResultSoftFail = "softfail" + AuthResultNeutral = "neutral" + AuthResultTempError = "temperror" + AuthResultPermError = "permerror" +) + +// InboundAuthInput is everything the authentication checks need about one +// inbound transaction. +type InboundAuthInput struct { + ClientIP net.IP + HeloDomain string + EnvelopeFrom string + Message []byte +} + +// InboundAuthResult carries the verdicts recorded against the message and +// surfaced to the user (PLAN.md section 6.1). +type InboundAuthResult struct { + SPF string + DKIM string + DMARC string + + // DmarcPolicy is the policy the sending domain published: "none", + // "quarantine" or "reject". Empty when the domain publishes no record. + DmarcPolicy string + + // AuthenticationResults is the RFC 8601 header to prepend to the stored + // message so downstream filters and the user can see how it was judged. + AuthenticationResults string +} + +// MailAuthenticator runs SPF, DKIM and DMARC over an inbound message. +type MailAuthenticator interface { + // Authenticate never fails the delivery on its own: a DNS outage yields + // a temperror verdict, not an error. + Authenticate(ctx context.Context, input InboundAuthInput) InboundAuthResult +} diff --git a/services/protocols/internal/application/port/tls_policy_resolver.go b/services/protocols/internal/application/port/tls_policy_resolver.go new file mode 100644 index 0000000..6450cc1 --- /dev/null +++ b/services/protocols/internal/application/port/tls_policy_resolver.go @@ -0,0 +1,18 @@ +package port + +import ( + "context" + + "lambdamail/protocols/internal/domain/entity" +) + +// TLSPolicyResolver determines the transport security policy that applies to +// one outbound destination, combining DANE and MTA-STS with the precedence of +// RFC 8461 section 2 (PLAN.md section 6.2). +type TLSPolicyResolver interface { + // Resolve never returns an error for a destination that simply publishes + // no policy: that is an opportunistic destination, not a failure. It + // errors only when a published policy could not be evaluated, which the + // caller must treat as a reason to defer rather than to downgrade. + Resolve(ctx context.Context, destinationDomain string, mxHost string) (entity.TLSPolicy, error) +} diff --git a/services/protocols/internal/application/usecase/process_outbound_email.go b/services/protocols/internal/application/usecase/process_outbound_email.go new file mode 100644 index 0000000..58a0bff --- /dev/null +++ b/services/protocols/internal/application/usecase/process_outbound_email.go @@ -0,0 +1,186 @@ +package usecase + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "strings" + "time" + + "github.com/alexedwards/argon2id" + "github.com/google/uuid" + + "lambdamail/protocols/internal/application/port" + "lambdamail/protocols/internal/domain/entity" +) + +var ( + // ErrSubmissionAuthFailed is returned for both an unknown mailbox and a + // wrong password: telling the two apart would turn submission into an + // account-enumeration oracle. + ErrSubmissionAuthFailed = errors.New("process outbound email: authentication failed") + + // ErrSendLimitExceeded is returned when a mailbox has queued more + // recipients in the last hour than its policy allows. + ErrSendLimitExceeded = errors.New("process outbound email: sender hourly recipient limit exceeded") + + // ErrSenderNotOwned is returned when the authenticated mailbox tries to + // send as somebody else. + ErrSenderNotOwned = errors.New("process outbound email: envelope sender does not belong to the authenticated mailbox") +) + +// deliveryWindow is how long a job may keep retrying before it bounces +// (PLAN.md section 6.3). +const deliveryWindow = 5 * 24 * time.Hour + +// sendLimitWindow is the period the per-mailbox recipient limit applies over. +const sendLimitWindow = time.Hour + +// ProcessOutboundEmailInput is one authenticated submission. +type ProcessOutboundEmailInput struct { + MailboxID uuid.UUID + SenderAddr string + DomainName string + Recipients []string + Body io.Reader + // MaxRecipientsPerHour is the sender's policy, carried from the + // authenticated session. + MaxRecipientsPerHour int +} + +// ProcessOutboundEmailUseCase implements the submission half of PLAN.md +// section 3: authenticate, apply send limits, sign with DKIM, enqueue. +type ProcessOutboundEmailUseCase struct { + auth port.AuthRepository + outbound port.OutboundRepository + blobs port.BlobStorage + signer port.DkimSigner +} + +func NewProcessOutboundEmailUseCase( + auth port.AuthRepository, + outbound port.OutboundRepository, + blobs port.BlobStorage, + signer port.DkimSigner, +) *ProcessOutboundEmailUseCase { + return &ProcessOutboundEmailUseCase{auth: auth, outbound: outbound, blobs: blobs, signer: signer} +} + +// Authenticate verifies a submission credential and returns the mailbox the +// session may then send as. +func (uc *ProcessOutboundEmailUseCase) Authenticate(ctx context.Context, address, password string) (*port.MailboxAuth, error) { + rec, err := uc.auth.FindByAddress(ctx, address) + if err != nil { + return nil, err + } + if rec == nil { + // Still spend the time a real comparison would, so that a missing + // mailbox is not distinguishable by response time. + _, _ = argon2id.ComparePasswordAndHash(password, decoyHash) + return nil, ErrSubmissionAuthFailed + } + + match, err := argon2id.ComparePasswordAndHash(password, rec.PasswordHash) + if err != nil || !match { + return nil, ErrSubmissionAuthFailed + } + return rec, nil +} + +// decoyHash is a well-formed Argon2id hash of a value nobody knows. It exists +// only to keep the timing of a failed lookup close to that of a real check. +const decoyHash = "$argon2id$v=19$m=65536,t=3,p=2$YWJjZGVmZ2hpamtsbW5vcA$Zm9yY2VkLWRlY295LWNvbXBhcmlzb24taGFzaA" + +// Submit signs the message and queues one delivery job per recipient. The +// message is stored once and every job references the same blob, matching the +// deduplication design of PLAN.md section 9. +func (uc *ProcessOutboundEmailUseCase) Submit(ctx context.Context, input ProcessOutboundEmailInput) error { + if len(input.Recipients) == 0 { + return fmt.Errorf("process outbound email: no recipients") + } + + if err := uc.checkSendLimit(ctx, input); err != nil { + return err + } + + payload, err := io.ReadAll(input.Body) + if err != nil { + return fmt.Errorf("read submitted message: %w", err) + } + + if uc.signer != nil { + signed, err := uc.signer.Sign(ctx, input.DomainName, payload) + if err != nil { + return fmt.Errorf("sign submitted message: %w", err) + } + payload = signed + } + + blob, err := uc.blobs.Store(ctx, bytes.NewReader(payload)) + if err != nil { + return fmt.Errorf("store submitted message: %w", err) + } + + now := time.Now() + mailboxID := input.MailboxID + + for _, recipient := range input.Recipients { + job := &entity.OutboundJob{ + MailboxID: &mailboxID, + BlobID: blob.ID, + EnvelopeFrom: input.SenderAddr, + EnvelopeTo: recipient, + DestinationDomain: recipientDomain(recipient), + Status: entity.OutboundJobStatusQueued, + Attempt: 0, + NextAttemptAt: now, + ExpiresAt: now.Add(deliveryWindow), + } + if err := uc.outbound.Enqueue(ctx, job); err != nil { + return fmt.Errorf("enqueue delivery to %s: %w", recipient, err) + } + } + + return nil +} + +// checkSendLimit refuses the whole submission when accepting it would take the +// mailbox past its hourly recipient budget. Refusing up front is what contains +// a compromised account; accepting and dropping later would not. +func (uc *ProcessOutboundEmailUseCase) checkSendLimit(ctx context.Context, input ProcessOutboundEmailInput) error { + if input.MaxRecipientsPerHour <= 0 { + return nil + } + + alreadySent, err := uc.outbound.CountRecipientsSince(ctx, input.MailboxID, time.Now().Add(-sendLimitWindow)) + if err != nil { + return fmt.Errorf("check send limit: %w", err) + } + + if alreadySent+len(input.Recipients) > input.MaxRecipientsPerHour { + return fmt.Errorf("%w: %d already sent plus %d requested exceeds %d", + ErrSendLimitExceeded, alreadySent, len(input.Recipients), input.MaxRecipientsPerHour) + } + return nil +} + +// AuthorizeSender enforces that a session sends only as itself. Without this +// any account could forge any other address on the server. +func AuthorizeSender(session *port.MailboxAuth, envelopeFrom string) error { + if session == nil { + return ErrSubmissionAuthFailed + } + if !strings.EqualFold(strings.TrimSpace(envelopeFrom), session.EmailAddress) { + return ErrSenderNotOwned + } + return nil +} + +func recipientDomain(address string) string { + if idx := strings.LastIndex(address, "@"); idx >= 0 && idx < len(address)-1 { + return strings.ToLower(address[idx+1:]) + } + return "" +} diff --git a/services/protocols/internal/application/usecase/process_outbound_email_test.go b/services/protocols/internal/application/usecase/process_outbound_email_test.go new file mode 100644 index 0000000..6cbca1b --- /dev/null +++ b/services/protocols/internal/application/usecase/process_outbound_email_test.go @@ -0,0 +1,171 @@ +package usecase + +import ( + "bytes" + "context" + "errors" + "strings" + "testing" + "time" + + "github.com/alexedwards/argon2id" + "github.com/google/uuid" + + "lambdamail/protocols/internal/application/port" + "lambdamail/protocols/internal/domain/entity" +) + +type fakeAuthRepo struct { + record *port.MailboxAuth +} + +func (f *fakeAuthRepo) FindByAddress(_ context.Context, _ string) (*port.MailboxAuth, error) { + return f.record, nil +} + +type countingOutboundRepo struct { + fakeOutboundRepo + recentCount int +} + +func (c *countingOutboundRepo) CountRecipientsSince(_ context.Context, _ uuid.UUID, _ time.Time) (int, error) { + return c.recentCount, nil +} + +type recordingSigner struct { + calledForDomain string +} + +func (r *recordingSigner) Sign(_ context.Context, domainName string, message []byte) ([]byte, error) { + r.calledForDomain = domainName + return append([]byte("DKIM-Signature: v=1; d="+domainName+"\r\n"), message...), nil +} + +func submissionInput(mailboxID uuid.UUID, recipients ...string) ProcessOutboundEmailInput { + return ProcessOutboundEmailInput{ + MailboxID: mailboxID, + SenderAddr: "user@example.test", + DomainName: "example.test", + Recipients: recipients, + Body: bytes.NewBufferString("Subject: hi\r\n\r\nbody"), + MaxRecipientsPerHour: 200, + } +} + +// PLAN.md section 3: submission signs with DKIM and enqueues one job per +// recipient, all referencing a single stored copy of the message. +func TestSubmit_SignsAndEnqueuesOneJobPerRecipient(t *testing.T) { + repo := &countingOutboundRepo{} + blobs := &capturingBlobStorage{} + signer := &recordingSigner{} + mailboxID := uuid.New() + + uc := NewProcessOutboundEmailUseCase(nil, repo, blobs, signer) + err := uc.Submit(context.Background(), submissionInput(mailboxID, "a@remote.test", "b@other.test")) + if err != nil { + t.Fatalf("Submit: %v", err) + } + + if len(repo.jobs) != 2 { + t.Fatalf("enqueued %d jobs, want 2", len(repo.jobs)) + } + if signer.calledForDomain != "example.test" { + t.Errorf("signed for domain %q, want example.test", signer.calledForDomain) + } + if !strings.HasPrefix(string(blobs.stored), "DKIM-Signature:") { + t.Errorf("the stored message is not the signed one:\n%s", blobs.stored) + } + + first, second := repo.jobs[0], repo.jobs[1] + if first.BlobID != second.BlobID { + t.Error("recipients got separate copies of the message instead of sharing one blob") + } + if first.DestinationDomain != "remote.test" || second.DestinationDomain != "other.test" { + t.Errorf("destination domains = %q, %q", first.DestinationDomain, second.DestinationDomain) + } + if first.Status != entity.OutboundJobStatusQueued { + t.Errorf("status = %s, want QUEUED", first.Status) + } + if first.ExpiresAt.Sub(first.NextAttemptAt) < 4*24*time.Hour { + t.Errorf("delivery window = %v, want the 5 day window of PLAN section 6.3", first.ExpiresAt.Sub(first.NextAttemptAt)) + } + if first.MailboxID == nil || *first.MailboxID != mailboxID { + t.Error("job is not attributed to the sending mailbox") + } +} + +// A compromised account must hit the wall before the mail is queued, not +// after (PLAN.md section 5.2). +func TestSubmit_RefusesWhenHourlyRecipientLimitWouldBeExceeded(t *testing.T) { + repo := &countingOutboundRepo{recentCount: 199} + + uc := NewProcessOutboundEmailUseCase(nil, repo, &capturingBlobStorage{}, nil) + err := uc.Submit(context.Background(), submissionInput(uuid.New(), "a@remote.test", "b@remote.test")) + + if !errors.Is(err, ErrSendLimitExceeded) { + t.Fatalf("error = %v, want ErrSendLimitExceeded", err) + } + if len(repo.jobs) != 0 { + t.Errorf("enqueued %d jobs despite exceeding the limit", len(repo.jobs)) + } +} + +func TestSubmit_AllowsSubmissionExactlyAtTheLimit(t *testing.T) { + repo := &countingOutboundRepo{recentCount: 198} + + uc := NewProcessOutboundEmailUseCase(nil, repo, &capturingBlobStorage{}, nil) + if err := uc.Submit(context.Background(), submissionInput(uuid.New(), "a@remote.test", "b@remote.test")); err != nil { + t.Fatalf("Submit at exactly the limit: %v", err) + } + if len(repo.jobs) != 2 { + t.Errorf("enqueued %d jobs, want 2", len(repo.jobs)) + } +} + +func TestAuthenticate_AcceptsCorrectPasswordAndRejectsWrongOne(t *testing.T) { + hash, err := argon2id.CreateHash("correct horse battery staple", argon2id.DefaultParams) + if err != nil { + t.Fatalf("CreateHash: %v", err) + } + repo := &fakeAuthRepo{record: &port.MailboxAuth{ + ID: uuid.New(), PasswordHash: hash, EmailAddress: "user@example.test", + }} + uc := NewProcessOutboundEmailUseCase(repo, nil, nil, nil) + ctx := context.Background() + + if _, err := uc.Authenticate(ctx, "user@example.test", "correct horse battery staple"); err != nil { + t.Fatalf("Authenticate with the right password: %v", err) + } + if _, err := uc.Authenticate(ctx, "user@example.test", "wrong"); !errors.Is(err, ErrSubmissionAuthFailed) { + t.Errorf("error = %v, want ErrSubmissionAuthFailed", err) + } +} + +// An unknown mailbox and a wrong password must be indistinguishable, so the +// endpoint cannot be used to enumerate accounts. +func TestAuthenticate_UnknownMailboxYieldsTheSameError(t *testing.T) { + uc := NewProcessOutboundEmailUseCase(&fakeAuthRepo{record: nil}, nil, nil, nil) + + if _, err := uc.Authenticate(context.Background(), "nobody@example.test", "x"); !errors.Is(err, ErrSubmissionAuthFailed) { + t.Fatalf("error = %v, want ErrSubmissionAuthFailed", err) + } +} + +// Without this check any authenticated account could forge mail from any +// other address the server hosts. +func TestAuthorizeSender_RejectsForeignEnvelopeSender(t *testing.T) { + session := &port.MailboxAuth{EmailAddress: "user@example.test"} + + if err := AuthorizeSender(session, "user@example.test"); err != nil { + t.Errorf("sending as itself was refused: %v", err) + } + if err := AuthorizeSender(session, "USER@EXAMPLE.TEST"); err != nil { + t.Errorf("address comparison must be case insensitive: %v", err) + } + if err := AuthorizeSender(session, "ceo@example.test"); !errors.Is(err, ErrSenderNotOwned) { + t.Errorf("error = %v, want ErrSenderNotOwned", err) + } + if err := AuthorizeSender(nil, "user@example.test"); !errors.Is(err, ErrSubmissionAuthFailed) { + t.Errorf("error = %v, want ErrSubmissionAuthFailed for an unauthenticated session", err) + } +} diff --git a/services/protocols/internal/application/usecase/relay.go b/services/protocols/internal/application/usecase/relay.go new file mode 100644 index 0000000..eefc38e --- /dev/null +++ b/services/protocols/internal/application/usecase/relay.go @@ -0,0 +1,143 @@ +package usecase + +import ( + "crypto/tls" + "crypto/x509" + "fmt" + "net" + "net/smtp" + "strconv" + "strings" + "time" +) + +const relayDialTimeout = 15 * time.Second + +// RelayConfig is the smarthost of PLAN.md section 10.4, used when the host's +// outbound port 25 is blocked. In this mode the relay takes responsibility for +// transport security to the final destination, so DANE and MTA-STS are not +// evaluated here - they would only describe the wrong hop. +type RelayConfig struct { + Host string + Port int + Username string + Password string + // RootCAs trusts an internal relay presenting a certificate from a + // private CA. Left nil, the system trust store is used. There is + // deliberately no option to skip verification: the credentials sent over + // this connection are worth protecting. + RootCAs *x509.CertPool +} + +// Configured reports whether delivery should go through the relay. +func (c RelayConfig) Configured() bool { + return c.Host != "" +} + +// Address renders the dial target, defaulting to the submission port. +func (c RelayConfig) Address() string { + port := c.Port + if port == 0 { + port = 587 + } + return net.JoinHostPort(c.Host, strconv.Itoa(port)) +} + +// SpfInclude renders the mechanism to add to the domain's SPF record. Without +// it the relay's own IP is not authorised to send for the domain and every +// message it forwards fails SPF (PLAN.md section 10.4). +func (c RelayConfig) SpfInclude() string { + if !c.Configured() { + return "" + } + // A relay host is normally a subdomain of the provider's sending + // infrastructure; including its organisational domain is what the + // provider documents. The host itself is used when it cannot be reduced. + host := strings.TrimSuffix(strings.ToLower(c.Host), ".") + labels := strings.Split(host, ".") + if len(labels) > 2 { + host = strings.Join(labels[len(labels)-2:], ".") + } + return "include:" + host +} + +// authFor selects the SASL mechanism to use against the relay. PLAIN is +// preferred, and CRAM-MD5 is deliberately not offered: it requires the server +// to store a reversible secret. +func (c RelayConfig) authFor(serverName string) smtp.Auth { + if c.Username == "" { + return nil + } + return smtp.PlainAuth("", c.Username, c.Password, serverName) +} + +// relayTLSConfig requires a valid certificate. Unlike opportunistic delivery +// to an arbitrary MX, the relay is a host we chose and configured, so there is +// no reason to accept an unverified certificate - and the credentials sent +// over this connection make it worth refusing. +func relayTLSConfig(relay RelayConfig) *tls.Config { + return &tls.Config{ + ServerName: relay.Host, + MinVersion: tls.VersionTLS12, + RootCAs: relay.RootCAs, + } +} + +// deliverViaRelay hands one message to the smarthost. +func (w *OutboundWorkerUseCase) deliverViaRelay(from, to string, payload []byte) error { + relay := w.relay + addr := relay.Address() + + conn, err := net.DialTimeout("tcp", addr, relayDialTimeout) + if err != nil { + return fmt.Errorf("dial relay %s: %w", addr, err) + } + defer conn.Close() + + client, err := smtp.NewClient(conn, relay.Host) + if err != nil { + return fmt.Errorf("relay handshake: %w", err) + } + defer client.Close() + + if err := client.Hello(w.localHost); err != nil { + return fmt.Errorf("relay EHLO failed: %w", err) + } + + if ok, _ := client.Extension("STARTTLS"); ok { + if err := client.StartTLS(relayTLSConfig(relay)); err != nil { + return fmt.Errorf("relay STARTTLS failed: %w", err) + } + } else if relay.Username != "" { + // Sending the relay credentials over a cleartext link would leak them + // to anyone on the path, which is worse than deferring the message. + return fmt.Errorf("relay %s does not offer STARTTLS and credentials must not be sent in the clear", addr) + } + + if auth := relay.authFor(relay.Host); auth != nil { + if err := client.Auth(auth); err != nil { + return fmt.Errorf("relay authentication failed: %w", err) + } + } + + if err := client.Mail(from); err != nil { + return fmt.Errorf("relay MAIL FROM failed: %w", err) + } + if err := client.Rcpt(to); err != nil { + return fmt.Errorf("relay RCPT TO failed: %w", err) + } + + writer, err := client.Data() + if err != nil { + return fmt.Errorf("relay DATA command failed: %w", err) + } + if _, err := writer.Write(payload); err != nil { + writer.Close() + return fmt.Errorf("relay write DATA failed: %w", err) + } + if err := writer.Close(); err != nil { + return fmt.Errorf("relay close DATA failed: %w", err) + } + + return client.Quit() +} diff --git a/services/protocols/internal/application/usecase/relay_test.go b/services/protocols/internal/application/usecase/relay_test.go new file mode 100644 index 0000000..1ab3bd3 --- /dev/null +++ b/services/protocols/internal/application/usecase/relay_test.go @@ -0,0 +1,309 @@ +package usecase + +import ( + "bufio" + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/base64" + "math/big" + "net" + "strings" + "testing" + "time" + + "lambdamail/protocols/internal/domain/entity" +) + +// fakeRelay serves one SMTP conversation with STARTTLS and AUTH, recording +// what the client sent. +type fakeRelay struct { + addr string + authSeen chan string + mailFrom chan string + offerTLS bool + certificate tls.Certificate +} + +func startFakeRelay(t *testing.T, offerTLS bool) *fakeRelay { + t.Helper() + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + t.Cleanup(func() { ln.Close() }) + + relay := &fakeRelay{ + addr: ln.Addr().String(), + authSeen: make(chan string, 1), + mailFrom: make(chan string, 1), + offerTLS: offerTLS, + } + if offerTLS { + relay.certificate = relayTestCertificate(t) + } + + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + go relay.serve(conn) + } + }() + + return relay +} + +func (f *fakeRelay) serve(conn net.Conn) { + defer conn.Close() + conn.SetDeadline(time.Now().Add(10 * time.Second)) + + r := bufio.NewReader(conn) + conn.Write([]byte("220 relay.test ESMTP\r\n")) + + for { + line, err := r.ReadString('\n') + if err != nil { + return + } + cmd := strings.ToUpper(strings.TrimSpace(line)) + + switch { + case strings.HasPrefix(cmd, "EHLO"): + if f.offerTLS { + conn.Write([]byte("250-relay.test\r\n250-STARTTLS\r\n250 AUTH PLAIN LOGIN\r\n")) + } else { + conn.Write([]byte("250-relay.test\r\n250 AUTH PLAIN LOGIN\r\n")) + } + + case cmd == "STARTTLS": + conn.Write([]byte("220 Ready to start TLS\r\n")) + tlsConn := tls.Server(conn, &tls.Config{Certificates: []tls.Certificate{f.certificate}}) + if err := tlsConn.Handshake(); err != nil { + return + } + conn = tlsConn + r = bufio.NewReader(conn) + + case strings.HasPrefix(cmd, "AUTH PLAIN"): + payload := strings.TrimSpace(strings.TrimPrefix(strings.TrimSpace(line)[len("AUTH PLAIN"):], " ")) + if decoded, err := base64.StdEncoding.DecodeString(payload); err == nil { + select { + case f.authSeen <- strings.ReplaceAll(string(decoded), "\x00", "|"): + default: + } + } + conn.Write([]byte("235 2.7.0 Authentication successful\r\n")) + + case strings.HasPrefix(cmd, "MAIL FROM"): + select { + case f.mailFrom <- strings.TrimSpace(line): + default: + } + conn.Write([]byte("250 Ok\r\n")) + + case cmd == "DATA": + conn.Write([]byte("354 Send it\r\n")) + for { + body, err := r.ReadString('\n') + if err != nil { + return + } + if strings.TrimSpace(body) == "." { + break + } + } + conn.Write([]byte("250 2.0.0 Queued\r\n")) + + case cmd == "QUIT": + conn.Write([]byte("221 Bye\r\n")) + return + + default: + conn.Write([]byte("250 Ok\r\n")) + } + } +} + +func relayWorker(t *testing.T, relay RelayConfig) (*OutboundWorkerUseCase, *fakeOutboundRepo) { + t.Helper() + + repo := &fakeOutboundRepo{} + repo.Enqueue(context.Background(), newTestJob()) + + worker := NewOutboundWorkerUseCase( + repo, + // A resolver that would fail loudly if it were consulted: with a + // relay configured, MX resolution must not happen at all. + &fakeMXResolver{hosts: []string{"127.0.0.1:1"}}, + &fakeBlobReader{payload: []byte("From: a\r\nTo: b\r\n\r\nTest")}, + nil, nil, "mail.local", + ) + worker.SetRelay(relay) + return worker, repo +} + +// PLAN.md section 10.4: with a smarthost every message leaves through the +// relay, whatever the destination domain is. +func TestOutboundWorker_DeliversThroughRelayInsteadOfMX(t *testing.T) { + relay := startFakeRelay(t, true) + host, portStr, _ := net.SplitHostPort(relay.addr) + + worker, repo := relayWorker(t, RelayConfig{ + Host: host, + Port: atoiOrZero(portStr), + Username: "relay-user", + Password: "relay-pass", + RootCAs: relayTestRoots, + }) + + if _, err := worker.ProcessBatch(context.Background(), "w1", 10); err != nil { + t.Fatalf("ProcessBatch: %v", err) + } + + job := repo.jobs[0] + if job.Status != entity.OutboundJobStatusDelivered { + t.Fatalf("status = %s, want DELIVERED (last error: %s)", job.Status, job.LastError) + } + if job.TlsPolicyUsed != entity.TLSModeRelay { + t.Errorf("TlsPolicyUsed = %q, want relay", job.TlsPolicyUsed) + } + + select { + case credentials := <-relay.authSeen: + if !strings.Contains(credentials, "relay-user") || !strings.Contains(credentials, "relay-pass") { + t.Errorf("relay credentials = %q", credentials) + } + case <-time.After(2 * time.Second): + t.Error("the worker never authenticated against the relay") + } + + select { + case from := <-relay.mailFrom: + if !strings.Contains(from, "user@domain.test") { + t.Errorf("MAIL FROM = %q, want the original envelope sender", from) + } + case <-time.After(2 * time.Second): + t.Error("the relay never received MAIL FROM") + } +} + +// Sending the relay credentials over a cleartext link would leak them; the +// message is deferred instead. +func TestOutboundWorker_RefusesRelayCredentialsWithoutStartTLS(t *testing.T) { + relay := startFakeRelay(t, false) + host, portStr, _ := net.SplitHostPort(relay.addr) + + worker, repo := relayWorker(t, RelayConfig{ + Host: host, Port: atoiOrZero(portStr), + Username: "relay-user", Password: "relay-pass", + }) + + if _, err := worker.ProcessBatch(context.Background(), "w1", 10); err != nil { + t.Fatalf("ProcessBatch: %v", err) + } + + job := repo.jobs[0] + if job.Status != entity.OutboundJobStatusDeferred { + t.Errorf("status = %s, want DEFERRED", job.Status) + } + if !strings.Contains(job.LastError, "STARTTLS") { + t.Errorf("LastError = %q, want it to name the missing STARTTLS", job.LastError) + } +} + +// An unreachable relay defers rather than bounces: the relay may come back. +func TestOutboundWorker_UnreachableRelayDefers(t *testing.T) { + worker, repo := relayWorker(t, RelayConfig{Host: "127.0.0.1", Port: 1}) + + if _, err := worker.ProcessBatch(context.Background(), "w1", 10); err != nil { + t.Fatalf("ProcessBatch: %v", err) + } + + if job := repo.jobs[0]; job.Status != entity.OutboundJobStatusDeferred { + t.Errorf("status = %s, want DEFERRED", job.Status) + } +} + +func TestRelayConfig_AddressDefaultsToSubmissionPort(t *testing.T) { + if got := (RelayConfig{Host: "relay.example.net"}).Address(); got != "relay.example.net:587" { + t.Errorf("Address = %q, want relay.example.net:587", got) + } + if got := (RelayConfig{Host: "relay.example.net", Port: 2525}).Address(); got != "relay.example.net:2525" { + t.Errorf("Address = %q, want relay.example.net:2525", got) + } +} + +// PLAN.md section 10.4: the published SPF must authorise the relay, otherwise +// the hard fail rejects everything the relay forwards. +func TestRelayConfig_SpfInclude(t *testing.T) { + cases := map[string]string{ + "smtp.sendgrid.net": "include:sendgrid.net", + "relay.example.net": "include:example.net", + "example.net": "include:example.net", + "": "", + } + + for host, want := range cases { + if got := (RelayConfig{Host: host}).SpfInclude(); got != want { + t.Errorf("SpfInclude(%q) = %q, want %q", host, got, want) + } + } +} + +func atoiOrZero(s string) int { + value := 0 + for _, c := range s { + if c < '0' || c > '9' { + return 0 + } + value = value*10 + int(c-'0') + } + return value +} + +// relayTestCertificate issues a certificate for 127.0.0.1 and registers it as +// a trusted root for the test, so the relay connection is verified for real +// rather than with verification switched off. +func relayTestCertificate(t *testing.T) tls.Certificate { + t.Helper() + + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatalf("generate key: %v", err) + } + template := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "relay.test"}, + IPAddresses: []net.IP{net.ParseIP("127.0.0.1")}, + DNSNames: []string{"127.0.0.1", "relay.test"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign, + BasicConstraintsValid: true, + IsCA: true, + } + der, err := x509.CreateCertificate(rand.Reader, template, template, &key.PublicKey, key) + if err != nil { + t.Fatalf("create certificate: %v", err) + } + parsed, err := x509.ParseCertificate(der) + if err != nil { + t.Fatalf("parse certificate: %v", err) + } + + relayTestRoots = x509.NewCertPool() + relayTestRoots.AddCert(parsed) + + return tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key} +} + +// relayTestRoots holds the CA pool for the certificate issued above. +var relayTestRoots *x509.CertPool diff --git a/services/protocols/internal/application/usecase/rollover_tlsa.go b/services/protocols/internal/application/usecase/rollover_tlsa.go new file mode 100644 index 0000000..d1892aa --- /dev/null +++ b/services/protocols/internal/application/usecase/rollover_tlsa.go @@ -0,0 +1,246 @@ +package usecase + +import ( + "context" + "crypto" + "crypto/sha256" + "crypto/x509" + "encoding/hex" + "encoding/pem" + "fmt" + "time" +) + +// RolloverStage names the state the rollover is in. It follows the state +// machine of PLAN.md section 5.1, which exists because the naive approach - +// publish the digest of the current certificate and update it at renewal - +// permanently breaks mail delivery to every DANE-validating receiver for as +// long as DNS has not caught up. +type RolloverStage string + +const ( + // StageStable means one association is published and the certificate + // matches it. Nothing to do. + StageStable RolloverStage = "STABLE" + // StagePublishNext means a key for the next certificate has been + // generated and its association has to go into DNS. + StagePublishNext RolloverStage = "PUBLISH_NEXT" + // StageAwaitPropagation means both associations are published and the + // rollover is waiting out the DNS caches before switching over. + StageAwaitPropagation RolloverStage = "AWAIT_PROPAGATION" + // StageRenew means the wait is over and the certificate can be re-issued + // against the pre-generated key. + StageRenew RolloverStage = "RENEW" + // StageCleanup means the new certificate is live and the old association + // can be retired, again only after the caches have expired. + StageCleanup RolloverStage = "CLEANUP" +) + +// RolloverConfig tunes the timings. +type RolloverConfig struct { + // RenewBefore is how long before expiry a renewal starts. Let's Encrypt + // certificates last 90 days and are renewable from day 60. + RenewBefore time.Duration + // PrepareBefore is how long before the renewal the next key is generated + // and published. It has to exceed PropagationWait comfortably. + PrepareBefore time.Duration + // PropagationWait is how long both associations stay published before the + // certificate changes. RFC 7671 section 8.1 asks for at least twice the + // record TTL; a day is used because a resolver that ignores TTLs is not + // hypothetical. + PropagationWait time.Duration +} + +// DefaultRolloverConfig follows PLAN.md section 5.1. +func DefaultRolloverConfig() RolloverConfig { + return RolloverConfig{ + RenewBefore: 30 * 24 * time.Hour, + PrepareBefore: 30 * 24 * time.Hour, + PropagationWait: 24 * time.Hour, + } +} + +// RolloverState is everything the decision needs. +type RolloverState struct { + // CurrentDigest is the TLSA value of the certificate being served. + CurrentDigest string + // NotAfter is the expiry of the certificate being served. + NotAfter time.Time + // PendingDigest is the association of the pre-generated next key, empty + // when no rollover is in flight. + PendingDigest string + // PendingPublishedAt is when that association went into DNS. + PendingPublishedAt *time.Time + // CertificateMatchesPending is true once the served certificate was + // issued against the pending key. + CertificateMatchesPending bool +} + +// RolloverDecision is the action to take now. +type RolloverDecision struct { + Stage RolloverStage + // Publish is the full set of associations that must be in DNS after this + // step. It is always the complete desired state, never a delta, so + // applying it is idempotent. + Publish []string + Reason string +} + +// RolloverTlsaUseCase decides how to move a DANE rollover forward. It is a +// pure decision function over the observed state: the caller performs the DNS +// writes and the issuance, which keeps the risky ordering rules in one +// testable place. +type RolloverTlsaUseCase struct { + config RolloverConfig + now func() time.Time +} + +func NewRolloverTlsaUseCase(config RolloverConfig) *RolloverTlsaUseCase { + if config.RenewBefore == 0 { + config = DefaultRolloverConfig() + } + return &RolloverTlsaUseCase{config: config, now: time.Now} +} + +// WithClock fixes the clock for tests. +func (uc *RolloverTlsaUseCase) WithClock(now func() time.Time) *RolloverTlsaUseCase { + uc.now = now + return uc +} + +func (uc *RolloverTlsaUseCase) Decide(_ context.Context, state RolloverState) RolloverDecision { + now := uc.now() + + // The new certificate is live. Both associations are still published; + // the old one may only go once the caches that hold it have expired. + if state.CertificateMatchesPending { + if state.PendingPublishedAt != nil && now.Sub(*state.PendingPublishedAt) < uc.config.PropagationWait { + return RolloverDecision{ + Stage: StageCleanup, + Publish: dedupe(state.CurrentDigest, state.PendingDigest), + Reason: "new certificate is live; keeping the previous association until caches expire", + } + } + return RolloverDecision{ + Stage: StageStable, + Publish: dedupe(state.PendingDigest), + Reason: "rollover complete; the previous association can be removed", + } + } + + // A rollover already in flight: both are published, so the only question + // is whether the wait is over. + if state.PendingDigest != "" { + if state.PendingPublishedAt == nil { + return RolloverDecision{ + Stage: StagePublishNext, + Publish: dedupe(state.CurrentDigest, state.PendingDigest), + Reason: "publishing the association for the pre-generated next key", + } + } + if now.Sub(*state.PendingPublishedAt) < uc.config.PropagationWait { + remaining := uc.config.PropagationWait - now.Sub(*state.PendingPublishedAt) + return RolloverDecision{ + Stage: StageAwaitPropagation, + Publish: dedupe(state.CurrentDigest, state.PendingDigest), + Reason: fmt.Sprintf("waiting %s more for the next association to propagate", remaining.Round(time.Minute)), + } + } + return RolloverDecision{ + Stage: StageRenew, + Publish: dedupe(state.CurrentDigest, state.PendingDigest), + Reason: "the next association has propagated; the certificate can be issued against the pre-generated key", + } + } + + // No rollover in flight. Start one only when the renewal is close enough. + if !state.NotAfter.IsZero() && now.After(state.NotAfter.Add(-uc.config.PrepareBefore)) { + return RolloverDecision{ + Stage: StagePublishNext, + Publish: dedupe(state.CurrentDigest), + Reason: "renewal is approaching; a next key has to be generated and published", + } + } + + return RolloverDecision{ + Stage: StageStable, + Publish: dedupe(state.CurrentDigest), + Reason: "certificate is current and no rollover is due", + } +} + +// ShouldRenewNow reports whether the certificate itself is due for renewal, +// independently of DANE. Deployments without DANE use only this. +func (uc *RolloverTlsaUseCase) ShouldRenewNow(notAfter time.Time) bool { + if notAfter.IsZero() { + return true + } + return uc.now().After(notAfter.Add(-uc.config.RenewBefore)) +} + +// TlsaDigestForKey renders the "3 1 1" association of a private key: the +// SHA-256 of its SubjectPublicKeyInfo (RFC 6698 section 2.1.2). Deriving it +// from the key rather than from a certificate is what lets the association be +// published before the certificate exists. +func TlsaDigestForKey(privateKeyPEM []byte) (string, error) { + block, _ := pem.Decode(privateKeyPEM) + if block == nil { + return "", fmt.Errorf("rollover: private key is not valid PEM") + } + + var publicKey any + switch parsed, err := x509.ParsePKCS8PrivateKey(block.Bytes); { + case err == nil: + // crypto.Signer is the interface every usable private key satisfies; + // its Public method returns crypto.PublicKey, which is a defined type + // rather than a bare any, so it has to be named exactly. + signer, ok := parsed.(crypto.Signer) + if !ok { + return "", fmt.Errorf("rollover: key of type %T has no public half", parsed) + } + publicKey = signer.Public() + default: + ecKey, ecErr := x509.ParseECPrivateKey(block.Bytes) + if ecErr != nil { + return "", fmt.Errorf("rollover: parse private key: %w", err) + } + publicKey = &ecKey.PublicKey + } + + spki, err := x509.MarshalPKIXPublicKey(publicKey) + if err != nil { + return "", fmt.Errorf("rollover: marshal public key: %w", err) + } + + sum := sha256.Sum256(spki) + return hex.EncodeToString(sum[:]), nil +} + +// TlsaDigestForCertificate renders the association of an already issued +// certificate, used to confirm what is actually being served. +func TlsaDigestForCertificate(certificatePEM []byte) (string, error) { + block, _ := pem.Decode(certificatePEM) + if block == nil { + return "", fmt.Errorf("rollover: certificate is not valid PEM") + } + cert, err := x509.ParseCertificate(block.Bytes) + if err != nil { + return "", fmt.Errorf("rollover: parse certificate: %w", err) + } + + sum := sha256.Sum256(cert.RawSubjectPublicKeyInfo) + return hex.EncodeToString(sum[:]), nil +} + +func dedupe(values ...string) []string { + seen := map[string]bool{} + var out []string + for _, value := range values { + if value == "" || seen[value] { + continue + } + seen[value] = true + out = append(out, value) + } + return out +} diff --git a/services/protocols/internal/application/usecase/rollover_tlsa_test.go b/services/protocols/internal/application/usecase/rollover_tlsa_test.go new file mode 100644 index 0000000..fb0a97a --- /dev/null +++ b/services/protocols/internal/application/usecase/rollover_tlsa_test.go @@ -0,0 +1,220 @@ +package usecase + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "math/big" + "testing" + "time" +) + +func rolloverAt(t *testing.T, now time.Time) *RolloverTlsaUseCase { + t.Helper() + return NewRolloverTlsaUseCase(DefaultRolloverConfig()). + WithClock(func() time.Time { return now }) +} + +func contains(values []string, want string) bool { + for _, v := range values { + if v == want { + return true + } + } + return false +} + +var now = time.Date(2026, 7, 20, 12, 0, 0, 0, time.UTC) + +// Far from expiry there is nothing to do, and exactly one association stays +// published. +func TestRollover_StableWhenRenewalIsFarAway(t *testing.T) { + decision := rolloverAt(t, now).Decide(context.Background(), RolloverState{ + CurrentDigest: "aaaa", + NotAfter: now.Add(80 * 24 * time.Hour), + }) + + if decision.Stage != StageStable { + t.Errorf("stage = %s, want STABLE", decision.Stage) + } + if len(decision.Publish) != 1 || decision.Publish[0] != "aaaa" { + t.Errorf("Publish = %v, want just the current association", decision.Publish) + } +} + +// As renewal approaches, a next key has to be generated and published. +func TestRollover_StartsWhenRenewalApproaches(t *testing.T) { + decision := rolloverAt(t, now).Decide(context.Background(), RolloverState{ + CurrentDigest: "aaaa", + NotAfter: now.Add(10 * 24 * time.Hour), + }) + + if decision.Stage != StagePublishNext { + t.Errorf("stage = %s, want PUBLISH_NEXT", decision.Stage) + } +} + +// The critical invariant of PLAN.md section 5.1: while a rollover is in +// flight, BOTH associations stay published. Dropping the current one before +// the new certificate is live is exactly what breaks DANE delivery. +func TestRollover_PublishesBothAssociationsWhileInFlight(t *testing.T) { + published := now.Add(-2 * time.Hour) + + decision := rolloverAt(t, now).Decide(context.Background(), RolloverState{ + CurrentDigest: "aaaa", + PendingDigest: "bbbb", + PendingPublishedAt: &published, + NotAfter: now.Add(10 * 24 * time.Hour), + }) + + if decision.Stage != StageAwaitPropagation { + t.Errorf("stage = %s, want AWAIT_PROPAGATION", decision.Stage) + } + if len(decision.Publish) != 2 { + t.Fatalf("Publish = %v, want both associations", decision.Publish) + } + if !contains(decision.Publish, "aaaa") || !contains(decision.Publish, "bbbb") { + t.Errorf("Publish = %v, want the current and the next association", decision.Publish) + } +} + +// The certificate must not change before the new association has had time to +// propagate. +func TestRollover_WaitsOutPropagationBeforeRenewing(t *testing.T) { + justPublished := now.Add(-30 * time.Minute) + longPublished := now.Add(-30 * time.Hour) + + early := rolloverAt(t, now).Decide(context.Background(), RolloverState{ + CurrentDigest: "aaaa", PendingDigest: "bbbb", PendingPublishedAt: &justPublished, + }) + if early.Stage != StageAwaitPropagation { + t.Errorf("stage = %s, want AWAIT_PROPAGATION 30 minutes in", early.Stage) + } + + ready := rolloverAt(t, now).Decide(context.Background(), RolloverState{ + CurrentDigest: "aaaa", PendingDigest: "bbbb", PendingPublishedAt: &longPublished, + }) + if ready.Stage != StageRenew { + t.Errorf("stage = %s, want RENEW after the propagation window", ready.Stage) + } + if len(ready.Publish) != 2 { + t.Errorf("Publish = %v, both associations must remain during renewal", ready.Publish) + } +} + +// Once the new certificate is live, the old association is retired - but only +// after the caches holding it have expired. +func TestRollover_RetiresOldAssociationOnlyAfterPropagation(t *testing.T) { + recent := now.Add(-2 * time.Hour) + old := now.Add(-40 * time.Hour) + + cleanup := rolloverAt(t, now).Decide(context.Background(), RolloverState{ + CurrentDigest: "aaaa", PendingDigest: "bbbb", + PendingPublishedAt: &recent, CertificateMatchesPending: true, + }) + if cleanup.Stage != StageCleanup { + t.Errorf("stage = %s, want CLEANUP", cleanup.Stage) + } + if len(cleanup.Publish) != 2 { + t.Errorf("Publish = %v, the old association must survive the cache window", cleanup.Publish) + } + + done := rolloverAt(t, now).Decide(context.Background(), RolloverState{ + CurrentDigest: "aaaa", PendingDigest: "bbbb", + PendingPublishedAt: &old, CertificateMatchesPending: true, + }) + if done.Stage != StageStable { + t.Errorf("stage = %s, want STABLE", done.Stage) + } + if len(done.Publish) != 1 || done.Publish[0] != "bbbb" { + t.Errorf("Publish = %v, want only the new association", done.Publish) + } +} + +// A pending key whose association has not been published yet must be +// published before anything else happens. +func TestRollover_PublishesUnpublishedPendingAssociation(t *testing.T) { + decision := rolloverAt(t, now).Decide(context.Background(), RolloverState{ + CurrentDigest: "aaaa", PendingDigest: "bbbb", PendingPublishedAt: nil, + }) + + if decision.Stage != StagePublishNext { + t.Errorf("stage = %s, want PUBLISH_NEXT", decision.Stage) + } + if !contains(decision.Publish, "aaaa") || !contains(decision.Publish, "bbbb") { + t.Errorf("Publish = %v, want both", decision.Publish) + } +} + +func TestRollover_ShouldRenewNow(t *testing.T) { + uc := rolloverAt(t, now) + + if uc.ShouldRenewNow(now.Add(60 * 24 * time.Hour)) { + t.Error("a certificate valid for another 60 days should not be renewed") + } + if !uc.ShouldRenewNow(now.Add(10 * 24 * time.Hour)) { + t.Error("a certificate expiring in 10 days should be renewed") + } + if !uc.ShouldRenewNow(time.Time{}) { + t.Error("a missing certificate should be issued") + } +} + +// The digest must be derivable from the key alone. That is the property the +// whole rollover depends on: the association for the next certificate has to +// be publishable before that certificate is issued. +func TestTlsaDigestForKey_MatchesTheIssuedCertificate(t *testing.T) { + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatalf("generate key: %v", err) + } + + keyDER, err := x509.MarshalPKCS8PrivateKey(key) + if err != nil { + t.Fatalf("marshal key: %v", err) + } + keyPEM := pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: keyDER}) + + fromKey, err := TlsaDigestForKey(keyPEM) + if err != nil { + t.Fatalf("TlsaDigestForKey: %v", err) + } + if len(fromKey) != 64 { + t.Errorf("digest length = %d, want 64 hex characters for SHA-256", len(fromKey)) + } + + // Issue a certificate against that same key and confirm both agree. + template := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "mail.example.test"}, + NotBefore: now.Add(-time.Hour), + NotAfter: now.Add(90 * 24 * time.Hour), + } + der, err := x509.CreateCertificate(rand.Reader, template, template, &key.PublicKey, key) + if err != nil { + t.Fatalf("create certificate: %v", err) + } + certPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}) + + fromCert, err := TlsaDigestForCertificate(certPEM) + if err != nil { + t.Fatalf("TlsaDigestForCertificate: %v", err) + } + + if fromKey != fromCert { + t.Errorf("digest from key = %s, from certificate = %s; they must agree or the pre-published association would never match", fromKey, fromCert) + } +} + +func TestTlsaDigest_RejectsGarbage(t *testing.T) { + if _, err := TlsaDigestForKey([]byte("not pem")); err == nil { + t.Error("expected an error for a non-PEM key") + } + if _, err := TlsaDigestForCertificate([]byte("not pem")); err == nil { + t.Error("expected an error for a non-PEM certificate") + } +} diff --git a/services/protocols/internal/infrastructure/arc/canonical.go b/services/protocols/internal/infrastructure/arc/canonical.go new file mode 100644 index 0000000..846fa7b --- /dev/null +++ b/services/protocols/internal/infrastructure/arc/canonical.go @@ -0,0 +1,147 @@ +// Package arc implements Authenticated Received Chain (RFC 8617). +// +// ARC exists for the case DKIM cannot cover: a mailing list or forwarder that +// legitimately modifies a message breaks its DKIM signature, and the receiving +// domain then sees a DMARC failure for mail that was authentic when it was +// sent. An ARC chain records each intermediary's own authentication verdict +// and seals it, so the final receiver can see that the message authenticated +// correctly at the first hop (PLAN.md section 5). +// +// Nothing in the Go ecosystem provides ARC, so the signature construction is +// implemented here on top of the same relaxed canonicalization DKIM uses +// (RFC 6376 section 3.4.2), which RFC 8617 mandates for both ARC signatures. +package arc + +import ( + "crypto" + "crypto/ed25519" + "crypto/rand" + "crypto/sha256" + "encoding/base64" + "fmt" + "strings" +) + +// Header names defined by RFC 8617 section 4.1. +const ( + HeaderAuthenticationResults = "ARC-Authentication-Results" + HeaderMessageSignature = "ARC-Message-Signature" + HeaderSeal = "ARC-Seal" +) + +// header is one parsed header field, kept in the order it appeared. +type header struct { + name string + value string + raw string +} + +// canonicalizeHeaderRelaxed renders a header field the way RFC 6376 section +// 3.4.2 prescribes: lowercase name, whitespace runs folded to a single space, +// no space around the colon, no trailing whitespace. +func canonicalizeHeaderRelaxed(name, value string) string { + name = strings.ToLower(strings.TrimSpace(name)) + + // Unfold: a continuation line is part of the same field value. + value = strings.ReplaceAll(value, "\r\n", "") + value = strings.ReplaceAll(value, "\n", "") + + value = strings.Join(strings.Fields(value), " ") + return name + ":" + value +} + +// canonicalizeBodyRelaxed applies RFC 6376 section 3.4.4: trailing whitespace +// removed from each line, internal whitespace runs collapsed, trailing empty +// lines removed, and a single CRLF kept when the body is not empty. +func canonicalizeBodyRelaxed(body string) string { + body = strings.ReplaceAll(body, "\r\n", "\n") + lines := strings.Split(body, "\n") + + for i, line := range lines { + lines[i] = strings.TrimRight(strings.Join(strings.Fields(line), " "), " \t") + } + + // Drop trailing empty lines. + end := len(lines) + for end > 0 && lines[end-1] == "" { + end-- + } + lines = lines[:end] + + if len(lines) == 0 { + return "" + } + return strings.Join(lines, "\r\n") + "\r\n" +} + +// bodyHash is the "bh=" tag: the base64 SHA-256 of the canonicalized body. +func bodyHash(body string) string { + sum := sha256.Sum256([]byte(canonicalizeBodyRelaxed(body))) + return base64.StdEncoding.EncodeToString(sum[:]) +} + +// splitMessage separates the header block from the body, keeping the headers +// in order and with their continuation lines joined. +func splitMessage(message string) (headers []header, body string) { + normalized := strings.ReplaceAll(message, "\r\n", "\n") + + headerBlock, body, found := strings.Cut(normalized, "\n\n") + if !found { + headerBlock = normalized + body = "" + } + + var current *header + for _, line := range strings.Split(headerBlock, "\n") { + if line == "" { + continue + } + // A line starting with whitespace continues the previous field. + if (line[0] == ' ' || line[0] == '\t') && current != nil { + current.value += " " + strings.TrimSpace(line) + current.raw += "\r\n" + line + continue + } + + name, value, ok := strings.Cut(line, ":") + if !ok { + continue + } + headers = append(headers, header{ + name: strings.TrimSpace(name), + value: strings.TrimSpace(value), + raw: line, + }) + current = &headers[len(headers)-1] + } + + return headers, body +} + +// sign produces the base64 signature over data for either supported key type. +func sign(signer crypto.Signer, data []byte) (string, error) { + // RFC 8617 section 4.1.2 allows rsa-sha256 and, following RFC 8463, + // ed25519-sha256. Ed25519 signs the message itself; RSA signs its digest. + if _, isEd25519 := signer.Public().(ed25519.PublicKey); isEd25519 { + signature, err := signer.Sign(rand.Reader, data, crypto.Hash(0)) + if err != nil { + return "", fmt.Errorf("ed25519 sign: %w", err) + } + return base64.StdEncoding.EncodeToString(signature), nil + } + + digest := sha256.Sum256(data) + signature, err := signer.Sign(rand.Reader, digest[:], crypto.SHA256) + if err != nil { + return "", fmt.Errorf("rsa sign: %w", err) + } + return base64.StdEncoding.EncodeToString(signature), nil +} + +// algorithmFor names the algorithm in the "a=" tag. +func algorithmFor(signer crypto.Signer) string { + if _, isEd25519 := signer.Public().(ed25519.PublicKey); isEd25519 { + return "ed25519-sha256" + } + return "rsa-sha256" +} diff --git a/services/protocols/internal/infrastructure/arc/sealer.go b/services/protocols/internal/infrastructure/arc/sealer.go new file mode 100644 index 0000000..fd58d09 --- /dev/null +++ b/services/protocols/internal/infrastructure/arc/sealer.go @@ -0,0 +1,268 @@ +package arc + +import ( + "context" + "crypto" + "fmt" + "strconv" + "strings" + "time" + + "lambdamail/protocols/internal/application/port" +) + +// maxChainInstance caps the chain length. RFC 8617 section 5.1.1 fixes the +// maximum instance number at 50; beyond it the chain must be treated as +// failed rather than extended. +const maxChainInstance = 50 + +// signedHeaders are the fields the ARC-Message-Signature covers. It mirrors +// the DKIM set: headers a forwarder is likely to add later stay out, or the +// signature would break for the very case ARC exists to handle. +var signedHeaders = []string{ + "from", "to", "cc", "subject", "date", "message-id", + "mime-version", "content-type", "content-transfer-encoding", + "in-reply-to", "references", +} + +// ChainValidation is the "cv=" tag: the state of the chain as it arrived. +type ChainValidation string + +const ( + // ChainNone means no ARC chain was present, so this hop starts one. + ChainNone ChainValidation = "none" + // ChainPass means the chain arrived intact. + ChainPass ChainValidation = "pass" + // ChainFail means the chain is broken. RFC 8617 section 5.1.2 forbids + // extending a failed chain, so sealing stops here. + ChainFail ChainValidation = "fail" +) + +// Sealer adds an ARC set to a message being forwarded. +type Sealer struct { + // domain and selector identify the key, and are the same identity DKIM + // signs with: an ARC seal is verified through the same DNS record. + domain string + selector string + signer crypto.Signer + // authServID names this server in the ARC-Authentication-Results header. + authServID string + now func() time.Time +} + +func NewSealer(domain, selector, authServID string, signer crypto.Signer) *Sealer { + if authServID == "" { + authServID = domain + } + return &Sealer{ + domain: domain, + selector: selector, + signer: signer, + authServID: authServID, + now: time.Now, + } +} + +// WithClock fixes the timestamp, so a test can assert on exact output. +func (s *Sealer) WithClock(now func() time.Time) *Sealer { + s.now = now + return s +} + +// Seal prepends one ARC set - ARC-Authentication-Results, then +// ARC-Message-Signature, then ARC-Seal - to the message. +// +// The order matters and is prescribed by RFC 8617 section 5.1.1: the seal is +// computed last, over the other two, and is placed above them. +func (s *Sealer) Seal(_ context.Context, message []byte, authResult port.InboundAuthResult) ([]byte, error) { + text := string(message) + headers, body := splitMessage(text) + + instance, chainState := s.inspectChain(headers) + if chainState == ChainFail { + // Extending a failed chain is forbidden; the message is forwarded as + // it is so the receiver can see the break for itself. + return message, nil + } + if instance > maxChainInstance { + return message, nil + } + + timestamp := s.now().Unix() + + // 1. ARC-Authentication-Results records what this hop concluded. + aar := fmt.Sprintf("i=%d; %s", instance, s.formatAuthResults(authResult)) + + // 2. ARC-Message-Signature covers the message, exactly as a DKIM + // signature would, but never covers the ARC-Seal headers. + ams, err := s.buildMessageSignature(instance, timestamp, headers, body) + if err != nil { + return nil, err + } + + // 3. ARC-Seal covers the whole chain including the two headers above. + seal, err := s.buildSeal(instance, timestamp, chainState, headers, aar, ams) + if err != nil { + return nil, err + } + + var out strings.Builder + out.WriteString(HeaderSeal + ": " + seal + "\r\n") + out.WriteString(HeaderMessageSignature + ": " + ams + "\r\n") + out.WriteString(HeaderAuthenticationResults + ": " + aar + "\r\n") + out.WriteString(text) + + return []byte(out.String()), nil +} + +// inspectChain finds the instance number this hop should use and the state of +// the chain that arrived. +func (s *Sealer) inspectChain(headers []header) (instance int, state ChainValidation) { + highest := 0 + previousCV := "" + + for _, h := range headers { + if !strings.EqualFold(h.name, HeaderSeal) { + continue + } + tags := parseTags(h.value) + if i, err := strconv.Atoi(tags["i"]); err == nil && i > highest { + highest = i + previousCV = tags["cv"] + } + } + + if highest == 0 { + // No chain yet: this hop is instance 1 and reports "none", which is + // what RFC 8617 section 5.1.1 requires of the first sealer. + return 1, ChainNone + } + if previousCV == string(ChainFail) { + return highest + 1, ChainFail + } + return highest + 1, ChainPass +} + +// buildMessageSignature produces the AMS value. It is a DKIM signature in all +// but name, so the tag set and canonicalization match RFC 6376. +func (s *Sealer) buildMessageSignature(instance int, timestamp int64, headers []header, body string) (string, error) { + present := signedHeaderList(headers) + + unsigned := fmt.Sprintf( + "i=%d; a=%s; c=relaxed/relaxed; d=%s; s=%s; t=%d; h=%s; bh=%s; b=", + instance, algorithmFor(s.signer), s.domain, s.selector, timestamp, + strings.Join(present, ":"), bodyHash(body), + ) + + material := canonicalHeaderBlock(headers, present) + // The signature covers its own header with an empty "b=" tag + // (RFC 6376 section 3.7). + material += canonicalizeHeaderRelaxed(HeaderMessageSignature, unsigned) + + signature, err := sign(s.signer, []byte(material)) + if err != nil { + return "", err + } + return unsigned + signature, nil +} + +// buildSeal produces the AS value. Unlike the message signature it covers no +// body and no message headers: only the ARC set headers, in instance order +// (RFC 8617 section 5.1.1). +func (s *Sealer) buildSeal(instance int, timestamp int64, state ChainValidation, headers []header, aar, ams string) (string, error) { + unsigned := fmt.Sprintf( + "i=%d; a=%s; cv=%s; d=%s; s=%s; t=%d; b=", + instance, algorithmFor(s.signer), state, s.domain, s.selector, timestamp, + ) + + var material strings.Builder + // Every previous ARC set, oldest first. + for i := 1; i < instance; i++ { + for _, name := range []string{HeaderAuthenticationResults, HeaderMessageSignature, HeaderSeal} { + if value, ok := findInstanceHeader(headers, name, i); ok { + material.WriteString(canonicalizeHeaderRelaxed(name, value)) + } + } + } + // Then this hop's own set, with the seal itself carrying an empty "b=". + material.WriteString(canonicalizeHeaderRelaxed(HeaderAuthenticationResults, aar)) + material.WriteString(canonicalizeHeaderRelaxed(HeaderMessageSignature, ams)) + material.WriteString(canonicalizeHeaderRelaxed(HeaderSeal, unsigned)) + + signature, err := sign(s.signer, []byte(material.String())) + if err != nil { + return "", err + } + return unsigned + signature, nil +} + +// formatAuthResults renders this hop's verdicts for the AAR header. +func (s *Sealer) formatAuthResults(result port.InboundAuthResult) string { + if result.AuthenticationResults != "" { + return result.AuthenticationResults + } + return fmt.Sprintf("%s; spf=%s; dkim=%s; dmarc=%s", + s.authServID, orNone(result.SPF), orNone(result.DKIM), orNone(result.DMARC)) +} + +func orNone(value string) string { + if value == "" { + return "none" + } + return value +} + +// signedHeaderList keeps only the headers actually present, in the order the +// "h=" tag will list them. +func signedHeaderList(headers []header) []string { + var present []string + for _, candidate := range signedHeaders { + for _, h := range headers { + if strings.EqualFold(h.name, candidate) { + present = append(present, candidate) + break + } + } + } + return present +} + +// canonicalHeaderBlock renders the signed headers in "h=" order. +func canonicalHeaderBlock(headers []header, names []string) string { + var sb strings.Builder + for _, name := range names { + for _, h := range headers { + if strings.EqualFold(h.name, name) { + sb.WriteString(canonicalizeHeaderRelaxed(h.name, h.value)) + break + } + } + } + return sb.String() +} + +// findInstanceHeader locates the ARC header of a given name and instance. +func findInstanceHeader(headers []header, name string, instance int) (string, bool) { + for _, h := range headers { + if !strings.EqualFold(h.name, name) { + continue + } + if parseTags(h.value)["i"] == strconv.Itoa(instance) { + return h.value, true + } + } + return "", false +} + +// parseTags reads the "tag=value; tag=value" form ARC and DKIM headers use. +func parseTags(value string) map[string]string { + tags := map[string]string{} + for _, part := range strings.Split(value, ";") { + key, val, ok := strings.Cut(part, "=") + if !ok { + continue + } + tags[strings.ToLower(strings.TrimSpace(key))] = strings.TrimSpace(val) + } + return tags +} diff --git a/services/protocols/internal/infrastructure/arc/sealer_test.go b/services/protocols/internal/infrastructure/arc/sealer_test.go new file mode 100644 index 0000000..94112e7 --- /dev/null +++ b/services/protocols/internal/infrastructure/arc/sealer_test.go @@ -0,0 +1,258 @@ +package arc + +import ( + "context" + "crypto" + "fmt" + "strings" + "testing" + "time" + + "lambdamail/protocols/internal/application/port" + "lambdamail/protocols/internal/infrastructure/dkim" +) + +const forwardedMessage = "From: author@origin.test\r\n" + + "To: list@forwarder.test\r\n" + + "Subject: hello\r\n" + + "Date: Mon, 20 Jul 2026 10:00:00 +0000\r\n" + + "Message-ID: \r\n" + + "\r\n" + + "the body\r\n" + +// keyring resolves the public keys of every sealing domain in a test. +type keyring map[string]crypto.PublicKey + +func (k keyring) lookup(domain, selector string) (crypto.PublicKey, error) { + key, ok := k[domain+"/"+selector] + if !ok { + return nil, fmt.Errorf("no key for %s/%s", domain, selector) + } + return key, nil +} + +func newSealer(t *testing.T, domain, algorithm string, ring keyring) *Sealer { + t.Helper() + + generated, err := dkim.Generate(algorithm) + if err != nil { + t.Fatalf("Generate: %v", err) + } + signer, err := dkim.ParsePrivateKey(generated.PrivateKeyPEM) + if err != nil { + t.Fatalf("ParsePrivateKey: %v", err) + } + ring[domain+"/default"] = signer.Public() + + fixed := time.Date(2026, 7, 20, 10, 0, 0, 0, time.UTC) + return NewSealer(domain, "default", "mx."+domain, signer). + WithClock(func() time.Time { return fixed }) +} + +func passingResult() port.InboundAuthResult { + return port.InboundAuthResult{ + SPF: "pass", DKIM: "pass", DMARC: "pass", + AuthenticationResults: "mx.forwarder.test; spf=pass; dkim=pass; dmarc=pass", + } +} + +// A first sealer starts the chain: instance 1 with cv=none +// (RFC 8617 section 5.1.1). +func TestSealer_StartsChainAtInstanceOneWithCvNone(t *testing.T) { + ring := keyring{} + sealed, err := newSealer(t, "forwarder.test", dkim.AlgorithmRSA2048, ring). + Seal(context.Background(), []byte(forwardedMessage), passingResult()) + if err != nil { + t.Fatalf("Seal: %v", err) + } + + out := string(sealed) + for _, name := range []string{HeaderSeal, HeaderMessageSignature, HeaderAuthenticationResults} { + if !strings.Contains(out, name+":") { + t.Errorf("missing %s:\n%s", name, out) + } + } + if !strings.Contains(out, "cv=none") { + t.Errorf("first seal must report cv=none:\n%s", out) + } + if !strings.Contains(out, "i=1") { + t.Errorf("first seal must be instance 1:\n%s", out) + } + if !strings.Contains(out, "d=forwarder.test") { + t.Errorf("seal does not claim the sealing domain:\n%s", out) + } + // The original message must survive untouched below the new headers. + if !strings.HasSuffix(out, forwardedMessage) { + t.Error("the original message was modified") + } +} + +// The header order is prescribed: the seal is computed last and sits on top. +func TestSealer_WritesHeadersInPrescribedOrder(t *testing.T) { + ring := keyring{} + sealed, _ := newSealer(t, "forwarder.test", dkim.AlgorithmRSA2048, ring). + Seal(context.Background(), []byte(forwardedMessage), passingResult()) + + out := string(sealed) + sealAt := strings.Index(out, HeaderSeal+":") + amsAt := strings.Index(out, HeaderMessageSignature+":") + aarAt := strings.Index(out, HeaderAuthenticationResults+":") + + if !(sealAt < amsAt && amsAt < aarAt) { + t.Errorf("header order is seal=%d ams=%d aar=%d, want seal first", sealAt, amsAt, aarAt) + } +} + +// The whole point: a chain we sealed must verify. +func TestSealer_ProducesVerifiableChain(t *testing.T) { + for _, algorithm := range []string{dkim.AlgorithmRSA2048, dkim.AlgorithmEd25519} { + t.Run(algorithm, func(t *testing.T) { + ring := keyring{} + sealed, err := newSealer(t, "forwarder.test", algorithm, ring). + Seal(context.Background(), []byte(forwardedMessage), passingResult()) + if err != nil { + t.Fatalf("Seal: %v", err) + } + + state, err := VerifyChain(sealed, ring.lookup) + if err != nil { + t.Fatalf("VerifyChain: %v", err) + } + if state != ChainPass { + t.Errorf("chain state = %s, want pass", state) + } + }) + } +} + +// A second forwarder extends the chain, and both sets must still verify. +func TestSealer_ExtendsExistingChain(t *testing.T) { + ring := keyring{} + + first, err := newSealer(t, "first.test", dkim.AlgorithmRSA2048, ring). + Seal(context.Background(), []byte(forwardedMessage), passingResult()) + if err != nil { + t.Fatalf("first Seal: %v", err) + } + + second, err := newSealer(t, "second.test", dkim.AlgorithmEd25519, ring). + Seal(context.Background(), first, passingResult()) + if err != nil { + t.Fatalf("second Seal: %v", err) + } + + out := string(second) + if !strings.Contains(out, "i=2") { + t.Errorf("the second hop did not take instance 2:\n%s", out) + } + // The second hop saw an intact chain, so it reports cv=pass. + if !strings.Contains(out, "cv=pass") { + t.Errorf("the second seal should report cv=pass:\n%s", out) + } + if count := strings.Count(out, HeaderSeal+":"); count != 2 { + t.Errorf("found %d seals, want 2", count) + } + + state, err := VerifyChain(second, ring.lookup) + if err != nil { + t.Fatalf("VerifyChain: %v", err) + } + if state != ChainPass { + t.Errorf("chain state = %s, want pass", state) + } +} + +// Modifying the body after sealing must break the message signature - this is +// what proves the body hash is actually covering anything. +func TestVerifyChain_DetectsBodyTampering(t *testing.T) { + ring := keyring{} + sealed, _ := newSealer(t, "forwarder.test", dkim.AlgorithmRSA2048, ring). + Seal(context.Background(), []byte(forwardedMessage), passingResult()) + + tampered := strings.Replace(string(sealed), "the body", "evil body", 1) + + state, err := VerifyChain([]byte(tampered), ring.lookup) + if err == nil { + t.Fatal("expected tampering to be detected") + } + if state != ChainFail { + t.Errorf("chain state = %s, want fail", state) + } +} + +// Modifying a signed header must break the signature too. +func TestVerifyChain_DetectsHeaderTampering(t *testing.T) { + ring := keyring{} + sealed, _ := newSealer(t, "forwarder.test", dkim.AlgorithmRSA2048, ring). + Seal(context.Background(), []byte(forwardedMessage), passingResult()) + + tampered := strings.Replace(string(sealed), "Subject: hello", "Subject: forged", 1) + + if state, err := VerifyChain([]byte(tampered), ring.lookup); err == nil || state != ChainFail { + t.Errorf("state = %s, err = %v; want a detected failure", state, err) + } +} + +// A message with no chain is "none", not a failure. +func TestVerifyChain_NoChainIsNone(t *testing.T) { + state, err := VerifyChain([]byte(forwardedMessage), keyring{}.lookup) + if err != nil { + t.Fatalf("VerifyChain: %v", err) + } + if state != ChainNone { + t.Errorf("state = %s, want none", state) + } +} + +// RFC 8617 section 5.1.2: a broken chain must not be extended. +func TestSealer_RefusesToExtendFailedChain(t *testing.T) { + ring := keyring{} + failed := "ARC-Seal: i=1; a=rsa-sha256; cv=fail; d=broken.test; s=default; t=1; b=AAAA\r\n" + + forwardedMessage + + sealed, err := newSealer(t, "forwarder.test", dkim.AlgorithmRSA2048, ring). + Seal(context.Background(), []byte(failed), passingResult()) + if err != nil { + t.Fatalf("Seal: %v", err) + } + + if string(sealed) != failed { + t.Error("a failed chain was extended instead of being left alone") + } +} + +// The relaxed body canonicalization of RFC 6376 section 3.4.4 must ignore +// trailing whitespace and trailing empty lines, so cosmetic changes in transit +// do not break the signature. +func TestBodyHash_IgnoresCosmeticWhitespace(t *testing.T) { + base := bodyHash("line one\r\nline two\r\n") + + equivalent := []string{ + "line one \r\nline two\r\n", + "line one\r\nline two\r\n\r\n\r\n", + "line one\r\nline two\r\n", + } + for _, variant := range equivalent { + if bodyHash(variant) != base { + t.Errorf("body hash changed for a cosmetically equivalent body: %q", variant) + } + } + + if bodyHash("line one\r\nline three\r\n") == base { + t.Error("body hash did not change for genuinely different content") + } +} + +func TestCanonicalizeHeaderRelaxed(t *testing.T) { + cases := map[string]string{ + "Subject: hello": "subject:hello", + "SUBJECT: hello ": "subject:hello", + "Subject:\thello there": "subject:hello there", + } + for input, want := range cases { + name, value, _ := strings.Cut(input, ":") + if got := canonicalizeHeaderRelaxed(name, value); got != want { + t.Errorf("canonicalizeHeaderRelaxed(%q) = %q, want %q", input, got, want) + } + } +} diff --git a/services/protocols/internal/infrastructure/arc/verifier.go b/services/protocols/internal/infrastructure/arc/verifier.go new file mode 100644 index 0000000..9a12ec1 --- /dev/null +++ b/services/protocols/internal/infrastructure/arc/verifier.go @@ -0,0 +1,135 @@ +package arc + +import ( + "crypto" + "crypto/ed25519" + "crypto/rsa" + "crypto/sha256" + "encoding/base64" + "fmt" + "strconv" + "strings" +) + +// VerifyChain checks an ARC chain against the public keys of the sealing +// domains and returns the chain validation state a downstream hop should +// report in its own "cv=" tag. +// +// publicKeyFor resolves the key for a domain and selector; it is injected so +// the caller decides how lookups happen (DNS in production, a fixed map in a +// test). +func VerifyChain(message []byte, publicKeyFor func(domain, selector string) (crypto.PublicKey, error)) (ChainValidation, error) { + headers, body := splitMessage(string(message)) + + highest := 0 + for _, h := range headers { + if strings.EqualFold(h.name, HeaderSeal) { + if i, err := strconv.Atoi(parseTags(h.value)["i"]); err == nil && i > highest { + highest = i + } + } + } + + if highest == 0 { + return ChainNone, nil + } + if highest > maxChainInstance { + return ChainFail, fmt.Errorf("arc: chain exceeds the maximum of %d instances", maxChainInstance) + } + + // RFC 8617 section 5.2: the most recent ARC-Message-Signature must + // validate, and every ARC-Seal in the chain must validate. + if err := verifyMessageSignature(headers, body, highest, publicKeyFor); err != nil { + return ChainFail, err + } + + for instance := 1; instance <= highest; instance++ { + if err := verifySeal(headers, instance, publicKeyFor); err != nil { + return ChainFail, err + } + } + + return ChainPass, nil +} + +func verifyMessageSignature(headers []header, body string, instance int, publicKeyFor func(string, string) (crypto.PublicKey, error)) error { + value, ok := findInstanceHeader(headers, HeaderMessageSignature, instance) + if !ok { + return fmt.Errorf("arc: no %s for instance %d", HeaderMessageSignature, instance) + } + + tags := parseTags(value) + if tags["bh"] != bodyHash(body) { + return fmt.Errorf("arc: body hash mismatch at instance %d", instance) + } + + signedNames := strings.Split(tags["h"], ":") + material := canonicalHeaderBlock(headers, signedNames) + material += canonicalizeHeaderRelaxed(HeaderMessageSignature, stripSignature(value)) + + return verifySignature(tags, []byte(material), publicKeyFor) +} + +func verifySeal(headers []header, instance int, publicKeyFor func(string, string) (crypto.PublicKey, error)) error { + value, ok := findInstanceHeader(headers, HeaderSeal, instance) + if !ok { + return fmt.Errorf("arc: no %s for instance %d", HeaderSeal, instance) + } + + var material strings.Builder + for i := 1; i < instance; i++ { + for _, name := range []string{HeaderAuthenticationResults, HeaderMessageSignature, HeaderSeal} { + if previous, ok := findInstanceHeader(headers, name, i); ok { + material.WriteString(canonicalizeHeaderRelaxed(name, previous)) + } + } + } + + if aar, ok := findInstanceHeader(headers, HeaderAuthenticationResults, instance); ok { + material.WriteString(canonicalizeHeaderRelaxed(HeaderAuthenticationResults, aar)) + } + if ams, ok := findInstanceHeader(headers, HeaderMessageSignature, instance); ok { + material.WriteString(canonicalizeHeaderRelaxed(HeaderMessageSignature, ams)) + } + material.WriteString(canonicalizeHeaderRelaxed(HeaderSeal, stripSignature(value))) + + return verifySignature(parseTags(value), []byte(material.String()), publicKeyFor) +} + +// stripSignature blanks the "b=" tag, which is how the signature over a header +// that contains itself is defined (RFC 6376 section 3.7). +func stripSignature(value string) string { + index := strings.LastIndex(value, "b=") + if index < 0 { + return value + } + return value[:index+2] +} + +func verifySignature(tags map[string]string, material []byte, publicKeyFor func(string, string) (crypto.PublicKey, error)) error { + signature, err := base64.StdEncoding.DecodeString(tags["b"]) + if err != nil { + return fmt.Errorf("arc: signature is not valid base64: %w", err) + } + + key, err := publicKeyFor(tags["d"], tags["s"]) + if err != nil { + return fmt.Errorf("arc: cannot resolve key for %s/%s: %w", tags["d"], tags["s"], err) + } + + switch typed := key.(type) { + case ed25519.PublicKey: + if !ed25519.Verify(typed, material, signature) { + return fmt.Errorf("arc: ed25519 signature does not verify for %s", tags["d"]) + } + return nil + case *rsa.PublicKey: + digest := sha256.Sum256(material) + if err := rsa.VerifyPKCS1v15(typed, crypto.SHA256, digest[:], signature); err != nil { + return fmt.Errorf("arc: rsa signature does not verify for %s: %w", tags["d"], err) + } + return nil + default: + return fmt.Errorf("arc: unsupported key type %T", key) + } +} diff --git a/services/protocols/internal/infrastructure/mailauth/authenticator.go b/services/protocols/internal/infrastructure/mailauth/authenticator.go new file mode 100644 index 0000000..0cef09b --- /dev/null +++ b/services/protocols/internal/infrastructure/mailauth/authenticator.go @@ -0,0 +1,221 @@ +// Package mailauth runs the inbound authentication chain of PLAN.md section +// 6.1: SPF over the envelope, DKIM over the message, then DMARC alignment on +// top of both. +package mailauth + +import ( + "bytes" + "context" + "fmt" + "net" + "net/mail" + "strings" + + spflib "blitiri.com.ar/go/spf" + dkimlib "github.com/emersion/go-msgauth/dkim" + dmarclib "github.com/emersion/go-msgauth/dmarc" + + "lambdamail/protocols/internal/application/port" +) + +// LookupTXTFunc resolves TXT records. It is injected so the chain can be +// tested against a fixed zone instead of the live internet. +type LookupTXTFunc func(domain string) ([]string, error) + +// Authenticator implements port.MailAuthenticator. +type Authenticator struct { + // authServID is the name this server puts in the Authentication-Results + // header so that downstream consumers can tell our verdicts from a + // forwarder's (RFC 8601 section 2.4). + authServID string + lookupTXT LookupTXTFunc + resolver spflib.DNSResolver +} + +func NewAuthenticator(authServID string) *Authenticator { + if authServID == "" { + authServID = "lambdamail" + } + return &Authenticator{authServID: authServID, lookupTXT: net.LookupTXT, resolver: net.DefaultResolver} +} + +// WithLookupTXT overrides DNS resolution, for tests and for pointing the +// checks at a validating resolver. +func (a *Authenticator) WithLookupTXT(lookup LookupTXTFunc, resolver spflib.DNSResolver) *Authenticator { + a.lookupTXT = lookup + a.resolver = resolver + return a +} + +func (a *Authenticator) Authenticate(ctx context.Context, input port.InboundAuthInput) port.InboundAuthResult { + result := port.InboundAuthResult{ + SPF: port.AuthResultNone, + DKIM: port.AuthResultNone, + DMARC: port.AuthResultNone, + } + + spfResult, spfDomain := a.checkSPF(ctx, input) + result.SPF = spfResult + + dkimResult, dkimDomains := a.checkDKIM(input.Message) + result.DKIM = dkimResult + + fromDomain := headerFromDomain(input.Message) + result.DMARC, result.DmarcPolicy = a.checkDMARC(fromDomain, spfResult, spfDomain, dkimResult, dkimDomains) + + result.AuthenticationResults = a.formatAuthResults(result, spfDomain, dkimDomains, fromDomain) + return result +} + +func (a *Authenticator) checkSPF(ctx context.Context, input port.InboundAuthInput) (result string, domain string) { + if input.ClientIP == nil { + return port.AuthResultNone, "" + } + + // RFC 7208 section 2.4: with an empty envelope sender (a bounce) the + // identity checked is the HELO name. + sender := input.EnvelopeFrom + if sender == "" { + sender = "postmaster@" + input.HeloDomain + } + domain = domainOf(sender) + if domain == "" { + return port.AuthResultNone, "" + } + + options := []spflib.Option{spflib.WithContext(ctx)} + if a.resolver != nil { + options = append(options, spflib.WithResolver(a.resolver)) + } + + res, _ := spflib.CheckHostWithSender(input.ClientIP, input.HeloDomain, sender, options...) + return string(res), domain +} + +func (a *Authenticator) checkDKIM(message []byte) (result string, signingDomains []string) { + verifications, err := dkimlib.VerifyWithOptions(bytes.NewReader(message), &dkimlib.VerifyOptions{ + LookupTXT: a.lookupTXT, + // A message carrying hundreds of signatures is a resource-exhaustion + // attempt, not a legitimate mail (PLAN.md section 12.6). + MaxVerifications: 10, + }) + if err != nil && len(verifications) == 0 { + return port.AuthResultPermError, nil + } + if len(verifications) == 0 { + return port.AuthResultNone, nil + } + + // One passing signature is enough to authenticate the message; the + // domains of the passing ones are what DMARC aligns against. + anyPass := false + for _, v := range verifications { + if v.Err == nil { + anyPass = true + signingDomains = append(signingDomains, v.Domain) + } + } + if anyPass { + return port.AuthResultPass, signingDomains + } + return port.AuthResultFail, nil +} + +func (a *Authenticator) checkDMARC(fromDomain, spfResult, spfDomain, dkimResult string, dkimDomains []string) (result string, policy string) { + if fromDomain == "" { + return port.AuthResultNone, "" + } + + record, err := dmarclib.LookupWithOptions(fromDomain, &dmarclib.LookupOptions{LookupTXT: a.lookupTXT}) + if err != nil { + if dmarclib.IsTempFail(err) { + return port.AuthResultTempError, "" + } + return port.AuthResultNone, "" + } + policy = string(record.Policy) + + // RFC 7489 section 6.6.2: DMARC passes when SPF or DKIM passes *and* the + // authenticated identifier aligns with the From: domain. + spfAligned := spfResult == port.AuthResultPass && + aligned(fromDomain, spfDomain, record.SPFAlignment == dmarclib.AlignmentStrict) + + dkimAligned := false + if dkimResult == port.AuthResultPass { + for _, d := range dkimDomains { + if aligned(fromDomain, d, record.DKIMAlignment == dmarclib.AlignmentStrict) { + dkimAligned = true + break + } + } + } + + if spfAligned || dkimAligned { + return port.AuthResultPass, policy + } + return port.AuthResultFail, policy +} + +// aligned implements the identifier alignment of RFC 7489 section 3.1: strict +// requires an exact match, relaxed accepts a shared organizational domain. +func aligned(fromDomain, authDomain string, strict bool) bool { + if authDomain == "" { + return false + } + fromDomain, authDomain = strings.ToLower(fromDomain), strings.ToLower(authDomain) + if fromDomain == authDomain { + return true + } + if strict { + return false + } + return strings.HasSuffix(fromDomain, "."+authDomain) || strings.HasSuffix(authDomain, "."+fromDomain) +} + +// formatAuthResults renders RFC 8601 section 2.2. The header is written by us +// and describes our own checks, so it must always name this server. +func (a *Authenticator) formatAuthResults(result port.InboundAuthResult, spfDomain string, dkimDomains []string, fromDomain string) string { + var sb strings.Builder + sb.WriteString(a.authServID) + + fmt.Fprintf(&sb, ";\r\n\tspf=%s", result.SPF) + if spfDomain != "" { + fmt.Fprintf(&sb, " smtp.mailfrom=%s", spfDomain) + } + + fmt.Fprintf(&sb, ";\r\n\tdkim=%s", result.DKIM) + if len(dkimDomains) > 0 { + fmt.Fprintf(&sb, " header.d=%s", dkimDomains[0]) + } + + fmt.Fprintf(&sb, ";\r\n\tdmarc=%s", result.DMARC) + if fromDomain != "" { + fmt.Fprintf(&sb, " header.from=%s", fromDomain) + } + + return sb.String() +} + +// headerFromDomain extracts the domain of the From: header, which is the +// identity DMARC is defined over (RFC 7489 section 6.6.1). +func headerFromDomain(message []byte) string { + msg, err := mail.ReadMessage(bytes.NewReader(message)) + if err != nil { + return "" + } + addresses, err := mail.ParseAddressList(msg.Header.Get("From")) + if err != nil || len(addresses) == 0 { + return "" + } + return domainOf(addresses[0].Address) +} + +func domainOf(address string) string { + if idx := strings.LastIndex(address, "@"); idx >= 0 && idx < len(address)-1 { + return strings.ToLower(address[idx+1:]) + } + return "" +} + +// Ensure port interface compliance +var _ port.MailAuthenticator = (*Authenticator)(nil) diff --git a/services/protocols/internal/infrastructure/mailauth/authenticator_test.go b/services/protocols/internal/infrastructure/mailauth/authenticator_test.go new file mode 100644 index 0000000..a173f64 --- /dev/null +++ b/services/protocols/internal/infrastructure/mailauth/authenticator_test.go @@ -0,0 +1,243 @@ +package mailauth + +import ( + "bytes" + "context" + "net" + "strings" + "testing" + + dkimlib "github.com/emersion/go-msgauth/dkim" + + "lambdamail/protocols/internal/application/port" + "lambdamail/protocols/internal/infrastructure/dkim" +) + +// fakeZone answers TXT and address lookups from a fixed map, so the whole +// authentication chain runs without touching the real internet. +type fakeZone struct { + txt map[string][]string + a map[string][]string +} + +func (z *fakeZone) LookupTXT(_ context.Context, name string) ([]string, error) { + return z.lookupTXT(name) +} + +func (z *fakeZone) lookupTXT(name string) ([]string, error) { + name = strings.TrimSuffix(name, ".") + if records, ok := z.txt[name]; ok { + return records, nil + } + return nil, &net.DNSError{Err: "no such host", Name: name, IsNotFound: true} +} + +func (z *fakeZone) LookupIPAddr(_ context.Context, name string) ([]net.IPAddr, error) { + name = strings.TrimSuffix(name, ".") + var out []net.IPAddr + for _, ip := range z.a[name] { + out = append(out, net.IPAddr{IP: net.ParseIP(ip)}) + } + if len(out) == 0 { + return nil, &net.DNSError{Err: "no such host", Name: name, IsNotFound: true} + } + return out, nil +} + +func (z *fakeZone) LookupMX(_ context.Context, name string) ([]*net.MX, error) { + return nil, &net.DNSError{Err: "no such host", Name: name, IsNotFound: true} +} + +func (z *fakeZone) LookupAddr(_ context.Context, addr string) ([]string, error) { + return nil, &net.DNSError{Err: "no such host", Name: addr, IsNotFound: true} +} + +func newTestAuthenticator(zone *fakeZone) *Authenticator { + return NewAuthenticator("mail.lambdamail.test").WithLookupTXT(zone.lookupTXT, zone) +} + +const plainMessage = "From: sender@sender.test\r\n" + + "To: rcpt@lambdamail.test\r\n" + + "Subject: hello\r\n" + + "Date: Mon, 20 Jul 2026 10:00:00 +0000\r\n" + + "Message-ID: \r\n" + + "\r\n" + + "body\r\n" + +// A sender whose SPF authorises the connecting IP, with a DMARC record in +// place, must come out as pass/pass on the aligned identifier. +func TestAuthenticate_SpfPassAlignsDmarc(t *testing.T) { + zone := &fakeZone{ + txt: map[string][]string{ + "sender.test": {"v=spf1 ip4:192.0.2.10 -all"}, + "_dmarc.sender.test": {"v=DMARC1; p=reject"}, + }, + } + + got := newTestAuthenticator(zone).Authenticate(context.Background(), port.InboundAuthInput{ + ClientIP: net.ParseIP("192.0.2.10"), + HeloDomain: "mx.sender.test", + EnvelopeFrom: "sender@sender.test", + Message: []byte(plainMessage), + }) + + if got.SPF != port.AuthResultPass { + t.Errorf("SPF = %q, want pass", got.SPF) + } + if got.DMARC != port.AuthResultPass { + t.Errorf("DMARC = %q, want pass (SPF aligned with the From: domain)", got.DMARC) + } + if got.DmarcPolicy != "reject" { + t.Errorf("DmarcPolicy = %q, want reject", got.DmarcPolicy) + } + if !strings.Contains(got.AuthenticationResults, "spf=pass") || + !strings.Contains(got.AuthenticationResults, "dmarc=pass") { + t.Errorf("Authentication-Results does not report the verdicts: %s", got.AuthenticationResults) + } + if !strings.HasPrefix(got.AuthenticationResults, "mail.lambdamail.test") { + t.Errorf("Authentication-Results must name this server first: %s", got.AuthenticationResults) + } +} + +// An unauthorised IP must fail SPF, and with no DKIM signature to fall back +// on, DMARC fails too - which is what lets the reject policy be honoured. +func TestAuthenticate_SpfFailWithoutDkimFailsDmarc(t *testing.T) { + zone := &fakeZone{ + txt: map[string][]string{ + "sender.test": {"v=spf1 ip4:192.0.2.10 -all"}, + "_dmarc.sender.test": {"v=DMARC1; p=quarantine"}, + }, + } + + got := newTestAuthenticator(zone).Authenticate(context.Background(), port.InboundAuthInput{ + ClientIP: net.ParseIP("198.51.100.66"), + HeloDomain: "evil.test", + EnvelopeFrom: "sender@sender.test", + Message: []byte(plainMessage), + }) + + if got.SPF != port.AuthResultFail { + t.Errorf("SPF = %q, want fail", got.SPF) + } + if got.DMARC != port.AuthResultFail { + t.Errorf("DMARC = %q, want fail", got.DMARC) + } + if got.DmarcPolicy != "quarantine" { + t.Errorf("DmarcPolicy = %q, want quarantine", got.DmarcPolicy) + } +} + +// A valid DKIM signature must rescue DMARC even when SPF fails, which is +// exactly the forwarding case DKIM exists for. +func TestAuthenticate_DkimPassAlignsDmarcDespiteSpfFail(t *testing.T) { + generated, err := dkim.Generate(dkim.AlgorithmRSA2048) + if err != nil { + t.Fatalf("Generate: %v", err) + } + signer, err := dkim.ParsePrivateKey(generated.PrivateKeyPEM) + if err != nil { + t.Fatalf("ParsePrivateKey: %v", err) + } + + var signed bytes.Buffer + err = dkimlib.Sign(&signed, strings.NewReader(plainMessage), &dkimlib.SignOptions{ + Domain: "sender.test", + Selector: "default", + Signer: signer, + HeaderCanonicalization: dkimlib.CanonicalizationRelaxed, + BodyCanonicalization: dkimlib.CanonicalizationRelaxed, + }) + if err != nil { + t.Fatalf("Sign: %v", err) + } + + zone := &fakeZone{ + txt: map[string][]string{ + "sender.test": {"v=spf1 ip4:192.0.2.10 -all"}, + "_dmarc.sender.test": {"v=DMARC1; p=reject"}, + "default._domainkey.sender.test": {"v=DKIM1; k=rsa; p=" + generated.PublicKeyBase64}, + }, + } + + got := newTestAuthenticator(zone).Authenticate(context.Background(), port.InboundAuthInput{ + ClientIP: net.ParseIP("203.0.113.99"), // forwarder, not in SPF + HeloDomain: "forwarder.test", + EnvelopeFrom: "sender@sender.test", + Message: signed.Bytes(), + }) + + if got.DKIM != port.AuthResultPass { + t.Fatalf("DKIM = %q, want pass", got.DKIM) + } + if got.SPF == port.AuthResultPass { + t.Fatalf("SPF unexpectedly passed; the test cannot prove DKIM rescued DMARC") + } + if got.DMARC != port.AuthResultPass { + t.Errorf("DMARC = %q, want pass via the aligned DKIM signature", got.DMARC) + } +} + +// A tampered body must invalidate the signature. +func TestAuthenticate_TamperedBodyFailsDkim(t *testing.T) { + generated, _ := dkim.Generate(dkim.AlgorithmRSA2048) + signer, _ := dkim.ParsePrivateKey(generated.PrivateKeyPEM) + + var signed bytes.Buffer + dkimlib.Sign(&signed, strings.NewReader(plainMessage), &dkimlib.SignOptions{ + Domain: "sender.test", Selector: "default", Signer: signer, + HeaderCanonicalization: dkimlib.CanonicalizationRelaxed, + BodyCanonicalization: dkimlib.CanonicalizationRelaxed, + }) + + tampered := bytes.Replace(signed.Bytes(), []byte("body"), []byte("evil"), 1) + + zone := &fakeZone{txt: map[string][]string{ + "default._domainkey.sender.test": {"v=DKIM1; k=rsa; p=" + generated.PublicKeyBase64}, + }} + + got := newTestAuthenticator(zone).Authenticate(context.Background(), port.InboundAuthInput{ + ClientIP: net.ParseIP("203.0.113.99"), + Message: tampered, + }) + + if got.DKIM != port.AuthResultFail { + t.Errorf("DKIM = %q, want fail for a tampered body", got.DKIM) + } +} + +// A domain publishing no DMARC record yields "none", never a hard failure. +func TestAuthenticate_NoDmarcRecordYieldsNone(t *testing.T) { + zone := &fakeZone{txt: map[string][]string{"sender.test": {"v=spf1 -all"}}} + + got := newTestAuthenticator(zone).Authenticate(context.Background(), port.InboundAuthInput{ + ClientIP: net.ParseIP("198.51.100.1"), + EnvelopeFrom: "sender@sender.test", + Message: []byte(plainMessage), + }) + + if got.DMARC != port.AuthResultNone { + t.Errorf("DMARC = %q, want none", got.DMARC) + } + if got.DmarcPolicy != "" { + t.Errorf("DmarcPolicy = %q, want empty", got.DmarcPolicy) + } +} + +// A bounce arrives with an empty envelope sender; RFC 7208 section 2.4 says +// the HELO identity is checked instead, and it must not crash. +func TestAuthenticate_EmptyEnvelopeFromChecksHelo(t *testing.T) { + zone := &fakeZone{txt: map[string][]string{ + "mx.sender.test": {"v=spf1 ip4:192.0.2.10 -all"}, + }} + + got := newTestAuthenticator(zone).Authenticate(context.Background(), port.InboundAuthInput{ + ClientIP: net.ParseIP("192.0.2.10"), + HeloDomain: "mx.sender.test", + EnvelopeFrom: "", + Message: []byte(plainMessage), + }) + + if got.SPF != port.AuthResultPass { + t.Errorf("SPF = %q, want pass against the HELO identity", got.SPF) + } +} diff --git a/services/protocols/internal/infrastructure/netdns/public_verifier.go b/services/protocols/internal/infrastructure/netdns/public_verifier.go new file mode 100644 index 0000000..a40f9cf --- /dev/null +++ b/services/protocols/internal/infrastructure/netdns/public_verifier.go @@ -0,0 +1,106 @@ +package netdns + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/miekg/dns" + + "lambdamail/protocols/internal/domain/entity" +) + +// DefaultPublicResolvers are the two independent resolvers PLAN.md section 7.5 +// names. Two operators are used deliberately: agreement between them is much +// stronger evidence of real propagation than one authoritative answer. +var DefaultPublicResolvers = []string{"1.1.1.1:53", "8.8.8.8:53"} + +// PublicVerifier checks published records through public resolvers. +type PublicVerifier struct { + resolvers []string + client *dns.Client +} + +func NewPublicVerifier(resolvers ...string) *PublicVerifier { + if len(resolvers) == 0 { + resolvers = DefaultPublicResolvers + } + normalized := make([]string, 0, len(resolvers)) + for _, r := range resolvers { + if !strings.Contains(r, ":") { + r += ":53" + } + normalized = append(normalized, r) + } + return &PublicVerifier{ + resolvers: normalized, + client: &dns.Client{Timeout: 5 * time.Second}, + } +} + +func (v *PublicVerifier) VerifyRecord(ctx context.Context, record entity.DnsRecord) (bool, string) { + qtype, ok := dns.StringToType[strings.ToUpper(record.Type)] + if !ok { + return false, fmt.Sprintf("unsupported record type %q", record.Type) + } + + // A proxied record resolves to the provider's anycast addresses rather + // than to the value we asked for, so comparing values would always + // disagree. Presence is all that can be checked. + checkValue := !record.Proxied + + var lastDetail string + for _, resolver := range v.resolvers { + msg := new(dns.Msg) + msg.SetQuestion(dns.Fqdn(record.Name), qtype) + + reply, _, err := v.client.ExchangeContext(ctx, msg, resolver) + if err != nil { + lastDetail = fmt.Sprintf("%s: %v", resolver, err) + continue + } + if reply.Rcode != dns.RcodeSuccess || len(reply.Answer) == 0 { + lastDetail = fmt.Sprintf("%s: no answer for %s %s", resolver, record.Type, record.Name) + continue + } + if !checkValue { + return true, "" + } + if answerMatches(reply.Answer, record.Value) { + return true, "" + } + lastDetail = fmt.Sprintf("%s: published value does not match the desired one", resolver) + } + + return false, lastDetail +} + +// answerMatches compares the resolver's answer against the desired value. +// Comparison is on the record's own presentation form with the owner name and +// TTL stripped, which is what makes it work uniformly across record types. +func answerMatches(answers []dns.RR, want string) bool { + want = normalizeRdata(want) + + for _, rr := range answers { + // String() renders "\t\t\t\t"; the + // rdata is everything after the fourth tab. + parts := strings.SplitN(rr.String(), "\t", 5) + if len(parts) < 5 { + continue + } + if normalizeRdata(parts[4]) == want { + return true + } + } + return false +} + +// normalizeRdata removes the differences that do not change meaning: quoting +// of TXT strings, trailing dots on names, case, and repeated whitespace. +func normalizeRdata(value string) string { + value = strings.ReplaceAll(value, `"`, "") + value = strings.Join(strings.Fields(value), " ") + value = strings.TrimSuffix(value, ".") + return strings.ToLower(value) +} diff --git a/services/protocols/internal/infrastructure/netdns/tlsa_resolver.go b/services/protocols/internal/infrastructure/netdns/tlsa_resolver.go new file mode 100644 index 0000000..c74f777 --- /dev/null +++ b/services/protocols/internal/infrastructure/netdns/tlsa_resolver.go @@ -0,0 +1,77 @@ +package netdns + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/miekg/dns" + + "lambdamail/protocols/internal/domain/valueobject" +) + +// TlsaResolver looks up DANE associations through a validating resolver. It +// reports whether the answer carried the AD bit, because a TLSA record that +// was not DNSSEC-validated is worthless: an attacker able to spoof DNS could +// otherwise choose the policy we apply (PLAN.md section 5.1). +type TlsaResolver struct { + // serverAddr is the validating resolver to ask, as host:port. + serverAddr string + client *dns.Client +} + +func NewTlsaResolver(serverAddr string) *TlsaResolver { + if serverAddr == "" { + serverAddr = "1.1.1.1:53" + } + if !strings.Contains(serverAddr, ":") { + serverAddr += ":53" + } + return &TlsaResolver{ + serverAddr: serverAddr, + client: &dns.Client{Timeout: 10 * time.Second}, + } +} + +func (r *TlsaResolver) LookupTLSA(ctx context.Context, host string, port int) ([]valueobject.TLSARecord, bool, error) { + name := dns.Fqdn(fmt.Sprintf("_%d._tcp.%s", port, strings.TrimSuffix(host, "."))) + + msg := new(dns.Msg) + msg.SetQuestion(name, dns.TypeTLSA) + // Ask the resolver to validate and to report the result in the AD bit. + msg.SetEdns0(4096, true) + msg.AuthenticatedData = true + + reply, _, err := r.client.ExchangeContext(ctx, msg, r.serverAddr) + if err != nil { + return nil, false, fmt.Errorf("query TLSA for %s: %w", name, err) + } + + switch reply.Rcode { + case dns.RcodeSuccess: + // Fall through to parsing. + case dns.RcodeNameError: + // No TLSA record: the destination is simply not DANE-protected. + return nil, reply.AuthenticatedData, nil + default: + return nil, false, fmt.Errorf("TLSA query for %s returned %s", name, dns.RcodeToString[reply.Rcode]) + } + + var records []valueobject.TLSARecord + for _, answer := range reply.Answer { + tlsa, ok := answer.(*dns.TLSA) + if !ok { + continue + } + // Usage 2 is rejected by explicit decision (PLAN.md section 5.1), so + // unsupported associations are skipped rather than failing the lookup. + record, err := valueobject.NewTLSARecord(tlsa.Usage, tlsa.Selector, tlsa.MatchingType, strings.ToLower(tlsa.Certificate)) + if err != nil { + continue + } + records = append(records, record) + } + + return records, reply.AuthenticatedData, nil +} diff --git a/services/protocols/internal/infrastructure/tlspolicy/resolver.go b/services/protocols/internal/infrastructure/tlspolicy/resolver.go new file mode 100644 index 0000000..64d876d --- /dev/null +++ b/services/protocols/internal/infrastructure/tlspolicy/resolver.go @@ -0,0 +1,233 @@ +// Package tlspolicy resolves the outbound transport security policy of +// PLAN.md section 6.2: DANE when the destination publishes DNSSEC-signed TLSA +// records, MTA-STS when it publishes a policy over HTTPS, opportunistic TLS +// otherwise. +package tlspolicy + +import ( + "context" + "fmt" + "io" + "net/http" + "strconv" + "strings" + "sync" + "time" + + "lambdamail/protocols/internal/application/port" + "lambdamail/protocols/internal/domain/entity" + "lambdamail/protocols/internal/domain/valueobject" +) + +// policyFetchLimit bounds the policy body. RFC 8461 policies are a few +// hundred bytes; anything larger is a resource-exhaustion attempt. +const policyFetchLimit = 64 * 1024 + +// Resolver implements port.TLSPolicyResolver. +type Resolver struct { + tlsa TlsaLookup + httpClient *http.Client + + daneEnabled bool + mtaStsEnabled bool + + mu sync.RWMutex + cache map[string]cachedPolicy + now func() time.Time +} + +// TlsaLookup returns the DANE associations published for a host, and whether +// the answer was DNSSEC-validated. An unvalidated answer must never be used: +// unsigned TLSA records are trivially forged (PLAN.md section 5.1). +type TlsaLookup interface { + LookupTLSA(ctx context.Context, host string, port int) (records []valueobject.TLSARecord, authenticated bool, err error) +} + +type cachedPolicy struct { + policy entity.TLSPolicy + expiresAt time.Time + // absent records that the destination publishes no MTA-STS policy, so + // repeated sends do not re-probe it on every message. + absent bool +} + +type Option func(*Resolver) + +// WithDane turns DANE resolution on. PLAN.md section 5.1 keeps it off unless +// the deployment manages its own certificates, because a TLSA record that +// stops matching after an ACME renewal blackholes mail permanently. +func WithDane(enabled bool, lookup TlsaLookup) Option { + return func(r *Resolver) { + r.daneEnabled = enabled + r.tlsa = lookup + } +} + +func WithMtaSts(enabled bool) Option { + return func(r *Resolver) { r.mtaStsEnabled = enabled } +} + +func WithHTTPClient(client *http.Client) Option { + return func(r *Resolver) { r.httpClient = client } +} + +func WithClock(now func() time.Time) Option { + return func(r *Resolver) { r.now = now } +} + +func NewResolver(options ...Option) *Resolver { + r := &Resolver{ + httpClient: &http.Client{Timeout: 15 * time.Second}, + mtaStsEnabled: true, + cache: map[string]cachedPolicy{}, + now: time.Now, + } + for _, option := range options { + option(r) + } + return r +} + +func (r *Resolver) Resolve(ctx context.Context, destinationDomain string, mxHost string) (entity.TLSPolicy, error) { + // DANE first: it is authenticated by DNSSEC and outranks MTA-STS + // (RFC 8461 section 2). + if r.daneEnabled && r.tlsa != nil { + records, authenticated, err := r.tlsa.LookupTLSA(ctx, mxHost, 25) + if err != nil { + return entity.TLSPolicy{}, fmt.Errorf("lookup TLSA for %s: %w", mxHost, err) + } + if len(records) > 0 { + if !authenticated { + // Unsigned TLSA records carry no security value and acting on + // them would let an attacker choose our policy. + return entity.TLSPolicy{}, fmt.Errorf("TLSA records for %s are not DNSSEC-validated", mxHost) + } + return entity.NewDanePolicy(records), nil + } + } + + if !r.mtaStsEnabled { + return entity.NewTLSPolicy(false, false), nil + } + + policy, found, err := r.mtaStsPolicy(ctx, destinationDomain) + if err != nil { + return entity.TLSPolicy{}, err + } + if !found { + return entity.NewTLSPolicy(false, false), nil + } + return policy, nil +} + +func (r *Resolver) mtaStsPolicy(ctx context.Context, domain string) (entity.TLSPolicy, bool, error) { + domain = strings.ToLower(strings.TrimSuffix(domain, ".")) + + if cached, ok := r.cached(domain); ok { + return cached.policy, !cached.absent, nil + } + + policy, maxAge, found, err := r.fetchPolicy(ctx, domain) + if err != nil { + return entity.TLSPolicy{}, false, err + } + + // Cache the negative answer too, but only briefly: a domain that starts + // publishing a policy should be picked up without a long delay. + ttl := time.Duration(maxAge) * time.Second + if !found { + ttl = time.Hour + } + r.store(domain, cachedPolicy{policy: policy, expiresAt: r.now().Add(ttl), absent: !found}) + + return policy, found, nil +} + +// fetchPolicy retrieves and parses https://mta-sts./.well-known/mta-sts.txt +// (RFC 8461 section 3.3). +func (r *Resolver) fetchPolicy(ctx context.Context, domain string) (policy entity.TLSPolicy, maxAge int, found bool, err error) { + url := fmt.Sprintf("https://mta-sts.%s/.well-known/mta-sts.txt", domain) + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return entity.TLSPolicy{}, 0, false, err + } + + resp, err := r.httpClient.Do(req) + if err != nil { + // A destination that publishes no policy host is simply not an + // MTA-STS destination; that is not an error for us. + return entity.TLSPolicy{}, 0, false, nil + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return entity.TLSPolicy{}, 0, false, nil + } + + body, err := io.ReadAll(io.LimitReader(resp.Body, policyFetchLimit)) + if err != nil { + return entity.TLSPolicy{}, 0, false, nil + } + + mode, patterns, maxAge := parsePolicy(string(body)) + if mode == "" || len(patterns) == 0 { + return entity.TLSPolicy{}, 0, false, nil + } + if mode == string(valueobject.MtaStsModeNone) { + // An explicit "none" is the destination withdrawing its policy. + return entity.TLSPolicy{}, 0, false, nil + } + + enforce := mode == string(valueobject.MtaStsModeEnforce) + return entity.NewMtaStsPolicy(patterns, enforce), maxAge, true, nil +} + +// parsePolicy reads the key/value policy format of RFC 8461 section 3.2. +func parsePolicy(body string) (mode string, mxPatterns []string, maxAge int) { + for _, line := range strings.Split(body, "\n") { + key, value, found := strings.Cut(strings.TrimSpace(line), ":") + if !found { + continue + } + key = strings.ToLower(strings.TrimSpace(key)) + value = strings.TrimSpace(value) + + switch key { + case "mode": + mode = strings.ToLower(value) + case "mx": + mxPatterns = append(mxPatterns, value) + case "max_age": + maxAge, _ = strconv.Atoi(value) + } + } + + // RFC 8461 section 3.2 caps max_age at 31557600 seconds; a missing or + // absurd value falls back to a day so a stale policy cannot be pinned + // forever by a malformed file. + if maxAge <= 0 || maxAge > 31557600 { + maxAge = 86400 + } + return mode, mxPatterns, maxAge +} + +func (r *Resolver) cached(domain string) (cachedPolicy, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + + entry, ok := r.cache[domain] + if !ok || r.now().After(entry.expiresAt) { + return cachedPolicy{}, false + } + return entry, true +} + +func (r *Resolver) store(domain string, entry cachedPolicy) { + r.mu.Lock() + defer r.mu.Unlock() + r.cache[domain] = entry +} + +// Ensure port interface compliance +var _ port.TLSPolicyResolver = (*Resolver)(nil) diff --git a/services/protocols/internal/infrastructure/tlspolicy/resolver_test.go b/services/protocols/internal/infrastructure/tlspolicy/resolver_test.go new file mode 100644 index 0000000..881aa3c --- /dev/null +++ b/services/protocols/internal/infrastructure/tlspolicy/resolver_test.go @@ -0,0 +1,222 @@ +package tlspolicy + +import ( + "context" + "crypto/tls" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "lambdamail/protocols/internal/domain/entity" + "lambdamail/protocols/internal/domain/valueobject" +) + +// policyServer stands in for https://mta-sts./.well-known/mta-sts.txt. +// The resolver builds that URL from the domain, so the test rewrites the +// request through a custom transport instead of resolving real DNS. +func policyServer(t *testing.T, body string, status int) (*http.Client, *int) { + t.Helper() + + hits := 0 + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits++ + if r.URL.Path != "/.well-known/mta-sts.txt" { + w.WriteHeader(http.StatusNotFound) + return + } + w.WriteHeader(status) + w.Write([]byte(body)) + })) + t.Cleanup(server.Close) + + client := server.Client() + base := client.Transport.(*http.Transport).Clone() + base.TLSClientConfig = &tls.Config{InsecureSkipVerify: true} + client.Transport = rewriteTransport{target: strings.TrimPrefix(server.URL, "https://"), inner: base} + + return client, &hits +} + +type rewriteTransport struct { + target string + inner http.RoundTripper +} + +func (rt rewriteTransport) RoundTrip(req *http.Request) (*http.Response, error) { + clone := req.Clone(req.Context()) + clone.URL.Host = rt.target + return rt.inner.RoundTrip(clone) +} + +type stubTlsaLookup struct { + records []valueobject.TLSARecord + authenticated bool + err error +} + +func (s stubTlsaLookup) LookupTLSA(_ context.Context, _ string, _ int) ([]valueobject.TLSARecord, bool, error) { + return s.records, s.authenticated, s.err +} + +func TestResolver_MtaStsEnforcePolicyIsParsed(t *testing.T) { + client, _ := policyServer(t, "version: STSv1\r\nmode: enforce\r\nmx: mx1.example.test\r\nmx: *.backup.test\r\nmax_age: 604800\r\n", http.StatusOK) + + policy, err := NewResolver(WithHTTPClient(client)). + Resolve(context.Background(), "example.test", "mx1.example.test") + if err != nil { + t.Fatalf("Resolve: %v", err) + } + + if policy.Effective() != entity.TLSModeMtaSts { + t.Errorf("Effective = %q, want mta-sts", policy.Effective()) + } + if !policy.RequiresValidation() { + t.Error("an enforce policy must require validation") + } + if !policy.CoversHost("mx1.example.test") || !policy.CoversHost("a.backup.test") { + t.Error("the mx patterns were not carried into the policy") + } + if policy.CoversHost("mx.evil.test") { + t.Error("an unlisted host must not be covered") + } +} + +func TestResolver_MtaStsTestingModeDoesNotEnforce(t *testing.T) { + client, _ := policyServer(t, "version: STSv1\nmode: testing\nmx: mx1.example.test\nmax_age: 86400\n", http.StatusOK) + + policy, err := NewResolver(WithHTTPClient(client)). + Resolve(context.Background(), "example.test", "mx1.example.test") + if err != nil { + t.Fatalf("Resolve: %v", err) + } + if policy.RequiresValidation() { + t.Error("mode: testing must not enforce") + } +} + +// A destination with no policy endpoint is opportunistic, which is not an +// error: most of the internet is in this state. +func TestResolver_MissingPolicyIsOpportunistic(t *testing.T) { + client, _ := policyServer(t, "", http.StatusNotFound) + + policy, err := NewResolver(WithHTTPClient(client)). + Resolve(context.Background(), "example.test", "mx1.example.test") + if err != nil { + t.Fatalf("Resolve: %v", err) + } + if policy.Effective() != entity.TLSModeOpportunistic { + t.Errorf("Effective = %q, want opportunistic", policy.Effective()) + } +} + +// RFC 8461 section 3.2 caches the policy for max_age; re-fetching on every +// message would hammer the destination and slow every send. +func TestResolver_CachesPolicyForMaxAge(t *testing.T) { + client, hits := policyServer(t, "version: STSv1\nmode: enforce\nmx: mx1.example.test\nmax_age: 604800\n", http.StatusOK) + resolver := NewResolver(WithHTTPClient(client)) + ctx := context.Background() + + for i := 0; i < 3; i++ { + if _, err := resolver.Resolve(ctx, "example.test", "mx1.example.test"); err != nil { + t.Fatalf("Resolve %d: %v", i, err) + } + } + + if *hits != 1 { + t.Errorf("fetched the policy %d times, want 1", *hits) + } +} + +func TestResolver_RefetchesAfterCacheExpiry(t *testing.T) { + client, hits := policyServer(t, "version: STSv1\nmode: enforce\nmx: mx1.example.test\nmax_age: 60\n", http.StatusOK) + + now := time.Now() + resolver := NewResolver(WithHTTPClient(client), WithClock(func() time.Time { return now })) + ctx := context.Background() + + resolver.Resolve(ctx, "example.test", "mx1.example.test") + now = now.Add(2 * time.Minute) + resolver.Resolve(ctx, "example.test", "mx1.example.test") + + if *hits != 2 { + t.Errorf("fetched the policy %d times, want 2 (cache expired between calls)", *hits) + } +} + +// An explicit "mode: none" is the destination withdrawing its policy. +func TestResolver_ModeNoneIsOpportunistic(t *testing.T) { + client, _ := policyServer(t, "version: STSv1\nmode: none\nmx: mx1.example.test\nmax_age: 86400\n", http.StatusOK) + + policy, _ := NewResolver(WithHTTPClient(client)). + Resolve(context.Background(), "example.test", "mx1.example.test") + if policy.Effective() != entity.TLSModeOpportunistic { + t.Errorf("Effective = %q, want opportunistic", policy.Effective()) + } +} + +// DANE outranks MTA-STS, so a destination with TLSA records never reaches the +// HTTPS fetch (RFC 8461 section 2). +func TestResolver_DaneWinsAndSkipsMtaStsFetch(t *testing.T) { + client, hits := policyServer(t, "version: STSv1\nmode: enforce\nmx: mx1.example.test\nmax_age: 86400\n", http.StatusOK) + record, _ := valueobject.NewTLSARecord(3, 1, 1, strings.Repeat("ab", 32)) + + policy, err := NewResolver( + WithHTTPClient(client), + WithDane(true, stubTlsaLookup{records: []valueobject.TLSARecord{record}, authenticated: true}), + ).Resolve(context.Background(), "example.test", "mx1.example.test") + if err != nil { + t.Fatalf("Resolve: %v", err) + } + + if policy.Effective() != entity.TLSModeDane { + t.Errorf("Effective = %q, want dane", policy.Effective()) + } + if *hits != 0 { + t.Error("the MTA-STS policy was fetched even though DANE applies") + } +} + +// Acting on unsigned TLSA records would let anyone who can spoof DNS choose +// our transport policy (PLAN.md section 5.1). +func TestResolver_RefusesUnauthenticatedTlsaRecords(t *testing.T) { + record, _ := valueobject.NewTLSARecord(3, 1, 1, strings.Repeat("ab", 32)) + + _, err := NewResolver( + WithDane(true, stubTlsaLookup{records: []valueobject.TLSARecord{record}, authenticated: false}), + ).Resolve(context.Background(), "example.test", "mx1.example.test") + + if err == nil { + t.Fatal("expected unsigned TLSA records to be refused") + } + if !strings.Contains(err.Error(), "DNSSEC") { + t.Errorf("error should explain the DNSSEC requirement: %v", err) + } +} + +// A destination with DANE disabled must fall through to MTA-STS. +func TestResolver_DaneDisabledFallsBackToMtaSts(t *testing.T) { + client, _ := policyServer(t, "version: STSv1\nmode: enforce\nmx: mx1.example.test\nmax_age: 86400\n", http.StatusOK) + + policy, err := NewResolver(WithHTTPClient(client), WithDane(false, stubTlsaLookup{})). + Resolve(context.Background(), "example.test", "mx1.example.test") + if err != nil { + t.Fatalf("Resolve: %v", err) + } + if policy.Effective() != entity.TLSModeMtaSts { + t.Errorf("Effective = %q, want mta-sts", policy.Effective()) + } +} + +func TestParsePolicy_ClampsAbsurdMaxAge(t *testing.T) { + _, _, maxAge := parsePolicy("mode: enforce\nmx: a.test\nmax_age: 999999999999\n") + if maxAge != 86400 { + t.Errorf("maxAge = %d, want the 86400 fallback", maxAge) + } + + _, _, missing := parsePolicy("mode: enforce\nmx: a.test\n") + if missing != 86400 { + t.Errorf("maxAge = %d, want the 86400 fallback", missing) + } +}