1package dmarcdb

2

3// Sending TLS reports and DMARC reports is very similar. See ../dmarcdb/eval.go:/similar and ../tlsrptsend/send.go:/similar.

4

5import (

6 "compress/gzip"

7 "context"

8 "encoding/xml"

9 "errors"

10 "fmt"

11 "io"

12 "log/slog"

13 "maps"

14 "mime"

15 "mime/multipart"

16 "net/textproto"

17 "net/url"

18 "os"

19 "runtime/debug"

20 "slices"

21 "strings"

22 "sync"

23 "time"

24

25 "github.com/prometheus/client_golang/prometheus"

26 "github.com/prometheus/client_golang/prometheus/promauto"

27

28 "github.com/mjl-/bstore"

29

30 "github.com/mjl-/mox/dkim"

31 "github.com/mjl-/mox/dmarc"

32 "github.com/mjl-/mox/dmarcrpt"

33 "github.com/mjl-/mox/dns"

34 "github.com/mjl-/mox/message"

35 "github.com/mjl-/mox/metrics"

36 "github.com/mjl-/mox/mlog"

37 "github.com/mjl-/mox/mox-"

38 "github.com/mjl-/mox/moxio"

39 "github.com/mjl-/mox/moxvar"

40 "github.com/mjl-/mox/publicsuffix"

41 "github.com/mjl-/mox/queue"

42 "github.com/mjl-/mox/smtp"

43 "github.com/mjl-/mox/store"

44)

45

46var (

47 metricReport = promauto.NewCounter(

48 prometheus.CounterOpts{

49 Name: "mox_dmarcdb_report_queued_total",

50 Help: "Total messages with DMARC aggregate/error reports queued.",

51 },

52 )

53 metricReportError = promauto.NewCounter(

54 prometheus.CounterOpts{

55 Name: "mox_dmarcdb_report_error_total",

56 Help: "Total errors while composing or queueing DMARC aggregate/error reports.",

57 },

58 )

59)

60

61var (

62 EvalDBTypes = []any{Evaluation{}, SuppressAddress{}} // Types stored in DB.

63 // Exported for backups. For incoming deliveries the SMTP server adds evaluations

64 // to the database. Every hour, a goroutine wakes up that gathers evaluations from

65 // the last hour(s), sends a report, and removes the evaluations from the database.

66 EvalDB *bstore.DB

67)

68

69// Evaluation is the result of an evaluation of a DMARC policy, to be included

70// in a DMARC report.

71type Evaluation struct {

72 ID int64

73

74 // Domain where DMARC policy was found, could be the organizational domain while

75 // evaluation was for a subdomain. Unicode. Same as domain found in

76 // PolicyPublished. A separate field for its index.

77 PolicyDomain string `bstore:"index"`

78

79 // Time of evaluation, determines which report (covering whole hours) this

80 // evaluation will be included in.

81 Evaluated time.Time `bstore:"default now"`

82

83 // If optional, this evaluation is not a reason to send a DMARC report, but it will

84 // be included when a report is sent due to other non-optional evaluations. Set for

85 // evaluations of incoming DMARC reports. We don't want such deliveries causing us to

86 // send a report, or we would keep exchanging reporting messages forever. Also set

87 // for when evaluation is a DMARC reject for domains we haven't positively

88 // interacted with, to prevent being used to flood an unsuspecting domain with

89 // reports.

90 Optional bool

91

92 // Effective aggregate reporting interval in hours. Between 1 and 24, rounded up

93 // from seconds from policy to first number that can divide 24.

94 IntervalHours int

95

96 // "rua" in DMARC record, we only store evaluations for records with aggregate reporting addresses, so always non-empty.

97 Addresses []string

98

99 // Policy used for evaluation. We don't store the "fo" field for failure reporting

100 // options, since we don't send failure reports for individual messages.

101 PolicyPublished dmarcrpt.PolicyPublished

102

103 // For "row" in a report record.

104 SourceIP string

105 Disposition dmarcrpt.Disposition

106 AlignedDKIMPass bool

107 AlignedSPFPass bool

108 OverrideReasons []dmarcrpt.PolicyOverrideReason

109

110 // For "identifiers" in a report record.

111 EnvelopeTo string

112 EnvelopeFrom string

113 HeaderFrom string

114

115 // For "auth_results" in a report record.

116 DKIMResults []dmarcrpt.DKIMAuthResult

117 SPFResults []dmarcrpt.SPFAuthResult

118}

119

120// SuppressAddress is a reporting address for which outgoing DMARC reports

121// will be suppressed for a period.

122type SuppressAddress struct {

123 ID int64

124 Inserted time.Time `bstore:"default now"`

125 ReportingAddress string `bstore:"unique"`

126 Until time.Time `bstore:"nonzero"`

127 Comment string

128}

129

130var dmarcResults = map[bool]dmarcrpt.DMARCResult{

131 false: dmarcrpt.DMARCFail,

132 true: dmarcrpt.DMARCPass,

133}

134

135// ReportRecord turns an evaluation into a record that can be included in a

136// report.

137func (e Evaluation) ReportRecord(count int) dmarcrpt.ReportRecord {

138 return dmarcrpt.ReportRecord{

139 Row: dmarcrpt.Row{

140 SourceIP: e.SourceIP,

141 Count: count,

142 PolicyEvaluated: dmarcrpt.PolicyEvaluated{

143 Disposition: e.Disposition,

144 DKIM: dmarcResults[e.AlignedDKIMPass],

145 SPF: dmarcResults[e.AlignedSPFPass],

146 Reasons: e.OverrideReasons,

147 },

148 },

149 Identifiers: dmarcrpt.Identifiers{

150 EnvelopeTo: e.EnvelopeTo,

151 EnvelopeFrom: e.EnvelopeFrom,

152 HeaderFrom: e.HeaderFrom,

153 },

154 AuthResults: dmarcrpt.AuthResults{

155 DKIM: e.DKIMResults,

156 SPF: e.SPFResults,

157 },

158 }

159}

160

161var intervalOpts = []int{24, 12, 8, 6, 4, 3, 2}

162

163func intervalHours(seconds int) int {

164 hours := (seconds + 3600 - 1) / 3600

165 for _, opt := range intervalOpts {

166 if hours >= opt {

167 return opt

168 }

169 }

170 return 1

171}

172

173// AddEvaluation adds the result of a DMARC evaluation for an incoming message

174// to the database.

175//

176// AddEvaluation sets Evaluation.IntervalHours based on

177// aggregateReportingIntervalSeconds.

178func AddEvaluation(ctx context.Context, aggregateReportingIntervalSeconds int, e *Evaluation) error {

179 e.IntervalHours = intervalHours(aggregateReportingIntervalSeconds)

180

181 e.ID = 0

182 return EvalDB.Insert(ctx, e)

183}

184

185// Evaluations returns all evaluations in the database.

186func Evaluations(ctx context.Context) ([]Evaluation, error) {

187 q := bstore.QueryDB[Evaluation](ctx, EvalDB)

188 q.SortAsc("Evaluated")

189 return q.List()

190}

191

192// EvaluationStat summarizes stored evaluations, for inclusion in an upcoming

193// aggregate report, for a domain.

194type EvaluationStat struct {

195 Domain dns.Domain

196 Dispositions []string

197 Count int

198 SendReport bool

199}

200

201// EvaluationStats returns evaluation counts and report-sending status per domain.

202func EvaluationStats(ctx context.Context) (map[string]EvaluationStat, error) {

203 r := map[string]EvaluationStat{}

204

205 err := bstore.QueryDB[Evaluation](ctx, EvalDB).ForEach(func(e Evaluation) error {

206 if stat, ok := r[e.PolicyDomain]; ok {

207 if !slices.Contains(stat.Dispositions, string(e.Disposition)) {

208 stat.Dispositions = append(stat.Dispositions, string(e.Disposition))

209 }

210 stat.Count++

211 stat.SendReport = stat.SendReport || !e.Optional

212 r[e.PolicyDomain] = stat

213 } else {

214 dom, err := dns.ParseDomain(e.PolicyDomain)

215 if err != nil {

216 return fmt.Errorf("parsing domain %q: %v", e.PolicyDomain, err)

217 }

218 r[e.PolicyDomain] = EvaluationStat{

219 Domain: dom,

220 Dispositions: []string{string(e.Disposition)},

221 Count: 1,

222 SendReport: !e.Optional,

223 }

224 }

225 return nil

226 })

227 return r, err

228}

229

230// EvaluationsDomain returns all evaluations for a domain.

231func EvaluationsDomain(ctx context.Context, domain dns.Domain) ([]Evaluation, error) {

232 q := bstore.QueryDB[Evaluation](ctx, EvalDB)

233 q.FilterNonzero(Evaluation{PolicyDomain: domain.Name()})

234 q.SortAsc("Evaluated")

235 return q.List()

236}

237

238// RemoveEvaluationsDomain removes evaluations for domain so they won't be sent in

239// an aggregate report.

240func RemoveEvaluationsDomain(ctx context.Context, domain dns.Domain) error {

241 q := bstore.QueryDB[Evaluation](ctx, EvalDB)

242 q.FilterNonzero(Evaluation{PolicyDomain: domain.Name()})

243 _, err := q.Delete()

244 return err

245}

246

247var jitterRand = mox.NewPseudoRand()

248

249// time to sleep until next whole hour t, replaced by tests.

250// Jitter so we don't cause load at exactly whole hours, other processes may

251// already be doing that.

252var jitteredTimeUntil = func(t time.Time) time.Duration {

253 return time.Until(t.Add(time.Duration(30+jitterRand.IntN(60)) * time.Second))

254}

255

256// Start launches a goroutine that wakes up at each whole hour (plus jitter) and

257// sends DMARC reports to domains that requested them.

258func Start(resolver dns.Resolver) {

259 go func() {

260 log := mlog.New("dmarcdb", nil)

261

262 defer func() {

263 // In case of panic don't take the whole program down.

264 x := recover()

265 if x != nil {

266 log.Error("recover from panic", slog.Any("panic", x))

267 debug.PrintStack()

268 metrics.PanicInc(metrics.Dmarcdb)

269 }

270 }()

271

272 timer := time.NewTimer(time.Hour)

273 defer timer.Stop()

274

275 ctx := mox.Shutdown

276

277 for {

278 now := time.Now()

279 nextEnd := nextWholeHour(now)

280 timer.Reset(jitteredTimeUntil(nextEnd))

281

282 select {

283 case <-ctx.Done():

284 log.Info("dmarc aggregate report sender shutting down")

285 return

286 case <-timer.C:

287 }

288

289 // Gather report intervals we want to process now. Multiples of hours that can

290 // divide 24, starting from UTC.

291 // ../rfc/7489:1750

292 utchour := nextEnd.UTC().Hour()

293 if utchour == 0 {

294 utchour = 24

295 }

296 intervals := []int{}

297 for _, ival := range intervalOpts {

298 if ival*(utchour/ival) == utchour {

299 intervals = append(intervals, ival)

300 }

301 }

302 intervals = append(intervals, 1)

303

304 // Remove evaluations older than 48 hours (2 reports with the default and maximum

305 // 24 hour interval). They should have been processed by now. We may have kept them

306 // during temporary errors, but persistent temporary errors shouldn't fill up our

307 // database. This also cleans up evaluations that were all optional for a domain.

308 _, err := bstore.QueryDB[Evaluation](ctx, EvalDB).FilterLess("Evaluated", nextEnd.Add(-48*time.Hour)).Delete()

309 log.Check(err, "removing stale dmarc evaluations from database")

310

311 clog := log.WithCid(mox.Cid())

312 clog.Info("sending dmarc aggregate reports", slog.Time("end", nextEnd.UTC()), slog.Any("intervals", intervals))

313 if err := sendReports(ctx, clog, resolver, EvalDB, nextEnd, intervals); err != nil {

314 clog.Errorx("sending dmarc aggregate reports", err)

315 metricReportError.Inc()

316 } else {

317 clog.Info("finished sending dmarc aggregate reports")

318 }

319 }

320 }()

321}

322

323func nextWholeHour(now time.Time) time.Time {

324 t := now

325 t = t.Add(time.Hour)

326 return time.Date(t.Year(), t.Month(), t.Day(), t.Hour(), 0, 0, 0, t.Location())

327}

328

329// We don't send reports at full speed. In the future, we could try to stretch out

330// reports a bit smarter. E.g. over 5 minutes with some minimum interval, and

331// perhaps faster and in parallel when there are lots of reports. Perhaps also

332// depending on reporting interval (faster for 1h, slower for 24h).

333// Replaced by tests.

334var sleepBetween = func(ctx context.Context, between time.Duration) (ok bool) {

335 t := time.NewTimer(between)

336 select {

337 case <-ctx.Done():

338 t.Stop()

339 return false

340 case <-t.C:

341 return true

342 }

343}

344

345// sendReports gathers all policy domains that have evaluations that should

346// receive a DMARC report and sends a report to each.

347func sendReports(ctx context.Context, log mlog.Log, resolver dns.Resolver, db *bstore.DB, endTime time.Time, intervals []int) error {

348 ivals := make([]any, len(intervals))

349 for i, v := range intervals {

350 ivals[i] = v

351 }

352

353 destDomains := map[string]bool{}

354

355 // Gather all domains that we plan to send to.

356 nsend := 0

357 q := bstore.QueryDB[Evaluation](ctx, db)

358 q.FilterLess("Evaluated", endTime)

359 q.FilterEqual("IntervalHours", ivals...)

360 err := q.ForEach(func(e Evaluation) error {

361 if !e.Optional && !destDomains[e.PolicyPublished.Domain] {

362 nsend++

363 }

364 destDomains[e.PolicyPublished.Domain] = destDomains[e.PolicyPublished.Domain] || !e.Optional

365 return nil

366 })

367 if err != nil {

368 return fmt.Errorf("looking for domains to send reports to: %v", err)

369 }

370

371 var wg sync.WaitGroup

372

373 // Sleep in between sending reports. We process hourly, and spread the reports over

374 // the hour, but with max 5 minute interval.

375 between := 45 * time.Minute

376 if nsend > 0 {

377 between /= time.Duration(nsend)

378 }

379 if between > 5*time.Minute {

380 between = 5 * time.Minute

381 }

382

383 // Attempt to send report to each domain.

384 n := 0

385 for d, send := range destDomains {

386 // Cleanup evaluations for domain with only optionals.

387 if !send {

388 removeEvaluations(ctx, log, db, endTime, d)

389 continue

390 }

391

392 if n > 0 {

393 if ok := sleepBetween(ctx, between); !ok {

394 return nil

395 }

396 }

397 n++

398

399 // Send in goroutine, so a slow process doesn't block progress.

400 wg.Add(1)

401 go func(domain string) {

402 defer func() {

403 // In case of panic don't take the whole program down.

404 x := recover()

405 if x != nil {

406 log.Error("unhandled panic in dmarcdb sendReports", slog.Any("panic", x))

407 debug.PrintStack()

408 metrics.PanicInc(metrics.Dmarcdb)

409 }

410 }()

411 defer wg.Done()

412

413 rlog := log.WithCid(mox.Cid()).With(slog.Any("domain", domain))

414 rlog.Info("sending dmarc report")

415 if _, err := sendReportDomain(ctx, rlog, resolver, db, endTime, domain); err != nil {

416 rlog.Errorx("sending dmarc aggregate report to domain", err)

417 metricReportError.Inc()

418 }

419 }(d)

420 }

421

422 wg.Wait()

423

424 return nil

425}

426

427type recipient struct {

428 address smtp.Address

429 maxSize uint64

430}

431

432func parseRecipient(log mlog.Log, uri dmarc.URI) (r recipient, ok bool) {

433 log = log.With(slog.Any("uri", uri.Address))

434

435 u, err := url.Parse(uri.Address)

436 if err != nil {

437 log.Debugx("parsing uri in dmarc record rua value", err)

438 return r, false

439 }

440 if !strings.EqualFold(u.Scheme, "mailto") {

441 log.Debug("skipping unrecognized scheme in dmarc record rua value")

442 return r, false

443 }

444 addr, err := smtp.ParseAddress(u.Opaque)

445 if err != nil {

446 log.Debugx("parsing mailto uri in dmarc record rua value", err)

447 return r, false

448 }

449

450 r = recipient{addr, uri.MaxSize}

451 // ../rfc/7489:1197

452 switch uri.Unit {

453 case "k", "K":

454 r.maxSize *= 1024

455 case "m", "M":

456 r.maxSize *= 1024 * 1024

457 case "g", "G":

458 r.maxSize *= 1024 * 1024 * 1024

459 case "t", "T":

460 // Oh yeah, terabyte-sized reports!

461 r.maxSize *= 1024 * 1024 * 1024 * 1024

462 case "":

463 default:

464 log.Debug("unrecognized max size unit in dmarc record rua value", slog.String("unit", uri.Unit))

465 return r, false

466 }

467

468 return r, true

469}

470

471func removeEvaluations(ctx context.Context, log mlog.Log, db *bstore.DB, endTime time.Time, domain string) {

472 q := bstore.QueryDB[Evaluation](ctx, db)

473 q.FilterLess("Evaluated", endTime)

474 q.FilterNonzero(Evaluation{PolicyDomain: domain})

475 _, err := q.Delete()

476 log.Check(err, "removing evaluations after processing for dmarc aggregate report")

477}

478

479// replaceable for testing.

480var queueAdd = queue.Add

481

482func sendReportDomain(ctx context.Context, log mlog.Log, resolver dns.Resolver, db *bstore.DB, endTime time.Time, domain string) (cleanup bool, rerr error) {

483 dom, err := dns.ParseDomain(domain)

484 if err != nil {

485 return false, fmt.Errorf("parsing domain for sending reports: %v", err)

486 }

487

488 // We'll cleanup records by default.

489 cleanup = true

490 // If we encounter a temporary error we cancel cleanup of evaluations on error.

491 tempError := false

492

493 defer func() {

494 if !cleanup || tempError {

495 log.Debug("not cleaning up evaluations after attempting to send dmarc aggregate report")

496 } else {

497 removeEvaluations(ctx, log, db, endTime, domain)

498 }

499 }()

500

501 // We're going to build up this report.

502 report := dmarcrpt.Feedback{

503 Version: "1.0",

504 ReportMetadata: dmarcrpt.ReportMetadata{

505 OrgName: mox.Conf.Static.HostnameDomain.ASCII,

506 Email: "postmaster@" + mox.Conf.Static.HostnameDomain.ASCII,

507 // ReportID and DateRange are set after we've seen evaluations.

508 // Errors is filled below when we encounter problems.

509 },

510 // We'll fill the records below.

511 Records: []dmarcrpt.ReportRecord{},

512 }

513

514 var errors []string // For report.ReportMetaData.Errors

515

516 // Check if we should be sending a report at all: if there are rua URIs in the

517 // current DMARC record. The interval may have changed too, but we'll flush out our

518 // evaluations regardless. We always use the latest DMARC record when sending, but

519 // we'll lump all policies of the last interval into one report.

520 // ../rfc/7489:1714

521 status, _, record, _, _, err := dmarc.Lookup(ctx, log.Logger, resolver, dom)

522 if err != nil {

523 // todo future: we could perhaps still send this report, assuming the values we know. in case of temporary error, we could also schedule again regardless of next interval hour (we would now only retry a 24h-interval report after 24h passed).

524 // Remove records unless it was a temporary error. We'll try again next round.

525 cleanup = status != dmarc.StatusTemperror

526 return cleanup, fmt.Errorf("looking up current dmarc record for reporting address: %v", err)

527 }

528

529 var recipients []recipient

530

531 // Gather all aggregate reporting addresses to try to send to. We'll start with

532 // those in the initial DMARC record, but will follow external reporting addresses

533 // and possibly update the list.

534 for _, uri := range record.AggregateReportAddresses {

535 r, ok := parseRecipient(log, uri)

536 if !ok {

537 continue

538 }

539

540 // Check if domain of rua recipient has the same organizational domain as for the

541 // evaluations. If not, we need to verify we are allowed to send.

542 ruaOrgDom := publicsuffix.Lookup(ctx, log.Logger, r.address.Domain)

543 evalOrgDom := publicsuffix.Lookup(ctx, log.Logger, dom)

544

545 if ruaOrgDom == evalOrgDom {

546 recipients = append(recipients, r)

547 continue

548 }

549

550 // Verify and follow addresses in other organizational domain through

551 // <policydomain>._report._dmarc.<host> lookup.

552 // ../rfc/7489:1556

553 accepts, status, records, _, _, err := dmarc.LookupExternalReportsAccepted(ctx, log.Logger, resolver, evalOrgDom, r.address.Domain)

554 log.Debugx("checking if rua address with different organization domain has opted into receiving dmarc reports", err,

555 slog.Any("policydomain", evalOrgDom),

556 slog.Any("destinationdomain", r.address.Domain),

557 slog.Bool("accepts", accepts),

558 slog.Any("status", status))

559 if status == dmarc.StatusTemperror {

560 // With a temporary error, we'll try to get the report the delivered anyway,

561 // perhaps there are multiple recipients.

562 // ../rfc/7489:1578

563 tempError = true

564 errors = append(errors, "temporary error checking authorization for report delegation to external address")

565 }

566 if !accepts {

567 errors = append(errors, fmt.Sprintf("rua %s is external domain that does not opt-in to receiving dmarc records through _report dmarc record", r.address))

568 continue

569 }

570

571 // We can follow a _report DMARC DNS record once. In that record, a domain may

572 // specify alternative addresses that we should send reports to instead. Such

573 // alternative address(es) must have the same host. If not, we ignore the new

574 // value. Behaviour for multiple records and/or multiple new addresses is

575 // underspecified. We'll replace an address with one or more new addresses, and

576 // keep the original if there was no candidate (which covers the case of invalid

577 // alternative addresses and no new address specified).

578 // ../rfc/7489:1600

579 foundReplacement := false

580 rlog := log.With(slog.Any("followedaddress", uri.Address))

581 for _, record := range records {

582 for _, exturi := range record.AggregateReportAddresses {

583 extr, ok := parseRecipient(rlog, exturi)

584 if !ok {

585 continue

586 }

587 if extr.address.Domain != r.address.Domain {

588 rlog.Debug("rua address in external _report dmarc record has different host than initial dmarc record, ignoring new name", slog.Any("externaladdress", extr.address))

589 errors = append(errors, fmt.Sprintf("rua %s is external domain with a replacement address %s with different host", r.address, extr.address))

590 } else {

591 rlog.Debug("using replacement rua address from external _report dmarc record", slog.Any("externaladdress", extr.address))

592 foundReplacement = true

593 recipients = append(recipients, extr)

594 }

595 }

596 }

597 if !foundReplacement {

598 recipients = append(recipients, r)

599 }

600 }

601

602 if len(recipients) == 0 {

603 // No reports requested, perfectly fine, no work to do for us.

604 log.Debug("no aggregate reporting addresses configured")

605 return true, nil

606 }

607

608 // We count idential records. Can be common with a domain sending quite some email.

609 // Though less if the sending domain has many IPs. In the future, we may want to

610 // remove some details from records so we can aggregate them into fewer rows.

611 type recordCount struct {

612 dmarcrpt.ReportRecord

613 count int

614 }

615 counts := map[string]recordCount{}

616

617 var first, last Evaluation // For daterange.

618 var sendReport bool

619

620 q := bstore.QueryDB[Evaluation](ctx, db)

621 q.FilterLess("Evaluated", endTime)

622 q.FilterNonzero(Evaluation{PolicyDomain: domain})

623 q.SortAsc("Evaluated")

624 err = q.ForEach(func(e Evaluation) error {

625 if first.ID == 0 {

626 first = e

627 }

628 last = e

629

630 record := e.ReportRecord(0)

631

632 // todo future: if we see many unique records from a single ip (exact ipv4 or ipv6 subnet), we may want to coalesce them into a single record, leaving out the fields that make them: a single ip could cause a report to contain many records with many unique domains, selectors, etc. it may compress relatively well, but the reports could still be huge.

633

634 // Simple but inefficient way to aggregate identical records. We may want to turn

635 // records into smaller representation in the future.

636 recbuf, err := xml.Marshal(record)

637 if err != nil {

638 return fmt.Errorf("xml marshal of report record: %v", err)

639 }

640 recstr := string(recbuf)

641 counts[recstr] = recordCount{record, counts[recstr].count + 1}

642 if !e.Optional {

643 sendReport = true

644 }

645 return nil

646 })

647 if err != nil {

648 return false, fmt.Errorf("gathering evaluations for report: %v", err)

649 }

650

651 if !sendReport {

652 log.Debug("no non-optional evaluations for domain, not sending dmarc aggregate report")

653 return true, nil

654 }

655

656 // Set begin and end date range. We try to set it to whole intervals as requested

657 // by the domain owner. The typical, default and maximum interval is 24 hours. But

658 // we allow any whole number of hours that can divide 24 hours. If we have an

659 // evaluation that is older, we may have had a failure to send earlier. We include

660 // those earlier intervals in this report as well.

661 //

662 // Although "end" could be interpreted as exclusive, to be on the safe side

663 // regarding client behaviour, and (related) to mimic large existing DMARC report

664 // senders, we set it to the last second of the period this report covers.

665 report.ReportMetadata.DateRange.End = endTime.Add(-time.Second).Unix()

666 interval := time.Duration(first.IntervalHours) * time.Hour

667 beginTime := endTime.Add(-interval)

668 for first.Evaluated.Before(beginTime) {

669 beginTime = beginTime.Add(-interval)

670 }

671 report.ReportMetadata.DateRange.Begin = beginTime.Unix()

672

673 // yyyymmddHH, we only send one report per hour, so should be unique per policy

674 // domain. We also add a truly unique id based on first evaluation id used without

675 // revealing the number of evaluations we have. Reuse of ReceivedID is not great,

676 // but shouldn't hurt.

677 report.ReportMetadata.ReportID = endTime.UTC().Format("20060102.15") + "." + mox.ReceivedID(first.ID)

678

679 // We may include errors we encountered when composing the report. We

680 // don't currently include errors about dmarc evaluations, e.g. DNS

681 // lookup errors during incoming deliveries.

682 report.ReportMetadata.Errors = errors

683

684 // We'll fill this with the last-used record, not the one we fetch fresh from DSN.

685 // They will almost always be the same, but if not, the fresh record was never

686 // actually used for evaluations, so no point in reporting it.

687 report.PolicyPublished = last.PolicyPublished

688

689 // Process records in-order for testable results.

690 for _, recstr := range slices.Sorted(maps.Keys(counts)) {

691 rc := counts[recstr]

692 rc.ReportRecord.Row.Count = rc.count

693 report.Records = append(report.Records, rc.ReportRecord)

694 }

695

696 reportFile, err := store.CreateMessageTemp(log, "dmarcreportout")

697 if err != nil {

698 return false, fmt.Errorf("creating temporary file for outgoing dmarc aggregate report: %v", err)

699 }

700 defer store.CloseRemoveTempFile(log, reportFile, "generated dmarc aggregate report")

701

702 gzw := gzip.NewWriter(reportFile)

703 _, err = fmt.Fprint(gzw, xml.Header)

704 enc := xml.NewEncoder(gzw)

705 enc.Indent("", "\t") // Keep up pretention that xml is human-readable.

706 if err == nil {

707 err = enc.Encode(report)

708 }

709 if err == nil {

710 err = enc.Close()

711 }

712 if err == nil {

713 err = gzw.Close()

714 }

715 if err != nil {

716 return true, fmt.Errorf("writing dmarc aggregate report as xml with gzip: %v", err)

717 }

718

719 msgf, err := store.CreateMessageTemp(log, "dmarcreportmsgout")

720 if err != nil {

721 return false, fmt.Errorf("creating temporary message file with outgoing dmarc aggregate report: %v", err)

722 }

723 defer store.CloseRemoveTempFile(log, msgf, "message with generated dmarc aggregate report")

724

725 // We are sending reports from our host's postmaster address. In a

726 // typical setup the host is a subdomain of a configured domain with

727 // DKIM keys, so we can DKIM-sign our reports. SPF should pass anyway.

728 // A single report can contain deliveries from a single policy domain

729 // to multiple of our configured domains.

730 from := smtp.NewAddress("postmaster", mox.Conf.Static.HostnameDomain)

731

732 // Subject follows the form in RFC. ../rfc/7489:1871

733 subject := fmt.Sprintf("Report Domain: %s Submitter: %s Report-ID: <%s>", dom.ASCII, mox.Conf.Static.HostnameDomain.ASCII, report.ReportMetadata.ReportID)

734

735 // Human-readable part for convenience. ../rfc/7489:1803

736 text := fmt.Sprintf(`Attached is an aggregate DMARC report with results of evaluations of the DMARC

737policy of your domain for messages received by us that have your domain in the

738message From header. You are receiving this message because your address is

739specified in the "rua" field of the DMARC record for your domain.

740

741Report domain: %s

742Submitter: %s

743Report-ID: %s

744Period: %s - %s UTC

745`, dom, mox.Conf.Static.HostnameDomain, report.ReportMetadata.ReportID, beginTime.UTC().Format(time.DateTime), endTime.UTC().Format(time.DateTime))

746

747 // The attached file follows the naming convention from the RFC. ../rfc/7489:1812

748 reportFilename := fmt.Sprintf("%s!%s!%d!%d.xml.gz", mox.Conf.Static.HostnameDomain.ASCII, dom.ASCII, beginTime.Unix(), endTime.Add(-time.Second).Unix())

749

750 var addrs []message.NameAddress

751 for _, rcpt := range recipients {

752 addrs = append(addrs, message.NameAddress{Address: rcpt.address})

753 }

754

755 // Compose the message.

756 msgPrefix, has8bit, smtputf8, messageID, err := composeAggregateReport(ctx, log, msgf, from, addrs, subject, text, reportFilename, reportFile)

757 if err != nil {

758 return false, fmt.Errorf("composing message with outgoing dmarc aggregate report: %v", err)

759 }

760

761 // Get size of message after all compression and encodings (base64 makes it big

762 // again), and go through potentials recipients (rua). If they are willing to

763 // accept the report, queue it.

764 msgInfo, err := msgf.Stat()

765 if err != nil {

766 return false, fmt.Errorf("stat message with outgoing dmarc aggregate report: %v", err)

767 }

768 msgSize := int64(len(msgPrefix)) + msgInfo.Size()

769 var queued bool

770 for _, rcpt := range recipients {

771 // If recipient is on suppression list, we won't queue the reporting message.

772 q := bstore.QueryDB[SuppressAddress](ctx, db)

773 q.FilterNonzero(SuppressAddress{ReportingAddress: rcpt.address.Path().String()})

774 q.FilterGreater("Until", time.Now())

775 exists, err := q.Exists()

776 if err != nil {

777 return false, fmt.Errorf("querying suppress list: %v", err)

778 }

779 if exists {

780 log.Info("suppressing outgoing dmarc aggregate report", slog.Any("reportingaddress", rcpt.address))

781 continue

782 }

783

784 // Only send to addresses where we don't exceed their size limit. The RFC mentions

785 // the size of the report, but then continues about the size after compression and

786 // transport encodings (i.e. gzip and the mime base64 attachment, so the intention

787 // is probably to compare against the size of the message that contains the report.

788 // ../rfc/7489:1773

789 if rcpt.maxSize > 0 && msgSize > int64(rcpt.maxSize) {

790 continue

791 }

792

793 qm := queue.MakeMsg(from.Path(), rcpt.address.Path(), has8bit, smtputf8, msgSize, messageID, []byte(msgPrefix), nil, time.Now(), subject)

794 // Don't try as long as regular deliveries, and stop before we would send the

795 // delayed DSN. Though we also won't send that due to IsDMARCReport.

796 qm.MaxAttempts = 5

797 qm.IsDMARCReport = true

798

799 err = queueAdd(ctx, log, mox.Conf.Static.Postmaster.Account, msgf, qm)

800 if err != nil {

801 tempError = true

802 log.Errorx("queueing message with dmarc aggregate report", err)

803 metricReportError.Inc()

804 } else {

805 log.Debug("dmarc aggregate report queued", slog.Any("recipient", rcpt.address))

806 queued = true

807 metricReport.Inc()

808 }

809 }

810

811 if !queued {

812 if err := sendErrorReport(ctx, log, db, from, addrs, dom, report.ReportMetadata.ReportID, msgSize); err != nil {

813 log.Errorx("sending dmarc error reports", err)

814 metricReportError.Inc()

815 }

816 }

817

818 // Regardless of whether we queued a report, we are not going to keep the

819 // evaluations around. Though this can be overridden if tempError is set.

820 // ../rfc/7489:1785

821

822 return true, nil

823}

824

825func composeAggregateReport(ctx context.Context, log mlog.Log, mf *os.File, fromAddr smtp.Address, recipients []message.NameAddress, subject, text, filename string, reportXMLGzipFile *os.File) (msgPrefix string, has8bit, smtputf8 bool, messageID string, rerr error) {

826 // We only use smtputf8 if we have to, with a utf-8 localpart. For IDNA, we use ASCII domains.

827 smtputf8 = fromAddr.Localpart.IsInternational()

828 for _, r := range recipients {

829 if smtputf8 {

830 smtputf8 = r.Address.Localpart.IsInternational()

831 break

832 }

833 }

834 xc := message.NewComposer(mf, 100*1024*1024, smtputf8)

835 defer func() {

836 x := recover()

837 if x == nil {

838 return

839 }

840 if err, ok := x.(error); ok && errors.Is(err, message.ErrCompose) {

841 rerr = err

842 return

843 }

844 panic(x)

845 }()

846

847 xc.HeaderAddrs("From", []message.NameAddress{{Address: fromAddr}})

848 xc.HeaderAddrs("To", recipients)

849 xc.Subject(subject)

850 messageID = fmt.Sprintf("<%s>", mox.MessageIDGen(xc.SMTPUTF8))

851 xc.Header("Message-Id", messageID)

852 xc.Header("Date", time.Now().Format(message.RFC5322Z))

853 xc.Header("User-Agent", "mox/"+moxvar.Version)

854 xc.Header("MIME-Version", "1.0")

855

856 // Multipart message, with a text/plain and the report attached.

857 mp := multipart.NewWriter(xc)

858 xc.Header("Content-Type", fmt.Sprintf(`multipart/mixed; boundary="%s"`, mp.Boundary()))

859 xc.Line()

860

861 // Textual part, just mentioning this is a DMARC report.

862 textBody, ct, cte := xc.TextPart("plain", text)

863 textHdr := textproto.MIMEHeader{}

864 textHdr.Set("Content-Type", ct)

865 textHdr.Set("Content-Transfer-Encoding", cte)

866 textp, err := mp.CreatePart(textHdr)

867 xc.Checkf(err, "adding text part to message")

868 _, err = textp.Write(textBody)

869 xc.Checkf(err, "writing text part")

870

871 // DMARC report as attachment.

872 ahdr := textproto.MIMEHeader{}

873 ahdr.Set("Content-Type", "application/gzip")

874 ahdr.Set("Content-Transfer-Encoding", "base64")

875 cd := mime.FormatMediaType("attachment", map[string]string{"filename": filename})

876 ahdr.Set("Content-Disposition", cd)

877 ap, err := mp.CreatePart(ahdr)

878 xc.Checkf(err, "adding dmarc aggregate report to message")

879 wc := moxio.Base64Writer(ap)

880 _, err = io.Copy(wc, &moxio.AtReader{R: reportXMLGzipFile})

881 xc.Checkf(err, "adding attachment")

882 err = wc.Close()

883 xc.Checkf(err, "flushing attachment")

884

885 err = mp.Close()

886 xc.Checkf(err, "closing multipart")

887

888 xc.Flush()

889

890 msgPrefix = dkimSign(ctx, log, fromAddr, xc.SMTPUTF8, mf)

891

892 return msgPrefix, xc.Has8bit, xc.SMTPUTF8, messageID, nil

893}

894

895// Though this functionality is quite underspecified, we'll do our best to send our

896// an error report in case our report is too large for all recipients.

897// ../rfc/7489:1918

898func sendErrorReport(ctx context.Context, log mlog.Log, db *bstore.DB, fromAddr smtp.Address, recipients []message.NameAddress, reportDomain dns.Domain, reportID string, reportMsgSize int64) error {

899 log.Debug("no reporting addresses willing to accept report given size, queuing short error message")

900

901 msgf, err := store.CreateMessageTemp(log, "dmarcreportmsg-out")

902 if err != nil {

903 return fmt.Errorf("creating temporary message file for outgoing dmarc error report: %v", err)

904 }

905 defer store.CloseRemoveTempFile(log, msgf, "outgoing dmarc error report message")

906

907 var recipientStrs []string

908 for _, rcpt := range recipients {

909 recipientStrs = append(recipientStrs, rcpt.Address.String())

910 }

911

912 subject := fmt.Sprintf("DMARC aggregate reporting error report for %s", reportDomain.ASCII)

913 // ../rfc/7489:1926

914 text := fmt.Sprintf(`Report-Date: %s

915Report-Domain: %s

916Report-ID: %s

917Report-Size: %d

918Submitter: %s

919Submitting-URI: %s

920`, time.Now().Format(message.RFC5322Z), reportDomain.ASCII, reportID, reportMsgSize, mox.Conf.Static.HostnameDomain.ASCII, strings.Join(recipientStrs, ","))

921 text = strings.ReplaceAll(text, "\n", "\r\n")

922

923 msgPrefix, has8bit, smtputf8, messageID, err := composeErrorReport(ctx, log, msgf, fromAddr, recipients, subject, text)

924 if err != nil {

925 return err

926 }

927

928 msgInfo, err := msgf.Stat()

929 if err != nil {

930 return fmt.Errorf("stat message with outgoing dmarc error report: %v", err)

931 }

932 msgSize := int64(len(msgPrefix)) + msgInfo.Size()

933

934 for _, rcpt := range recipients {

935 // If recipient is on suppression list, we won't queue the reporting message.

936 q := bstore.QueryDB[SuppressAddress](ctx, db)

937 q.FilterNonzero(SuppressAddress{ReportingAddress: rcpt.Address.Path().String()})

938 q.FilterGreater("Until", time.Now())

939 exists, err := q.Exists()

940 if err != nil {

941 return fmt.Errorf("querying suppress list: %v", err)

942 }

943 if exists {

944 log.Info("suppressing outgoing dmarc error report", slog.Any("reportingaddress", rcpt.Address))

945 continue

946 }

947

948 qm := queue.MakeMsg(fromAddr.Path(), rcpt.Address.Path(), has8bit, smtputf8, msgSize, messageID, []byte(msgPrefix), nil, time.Now(), subject)

949 // Don't try as long as regular deliveries, and stop before we would send the

950 // delayed DSN. Though we also won't send that due to IsDMARCReport.

951 qm.MaxAttempts = 5

952 qm.IsDMARCReport = true

953

954 if err := queueAdd(ctx, log, mox.Conf.Static.Postmaster.Account, msgf, qm); err != nil {

955 log.Errorx("queueing message with dmarc error report", err)

956 metricReportError.Inc()

957 } else {

958 log.Debug("dmarc error report queued", slog.Any("recipient", rcpt))

959 metricReport.Inc()

960 }

961 }

962 return nil

963}

964

965func composeErrorReport(ctx context.Context, log mlog.Log, mf *os.File, fromAddr smtp.Address, recipients []message.NameAddress, subject, text string) (msgPrefix string, has8bit, smtputf8 bool, messageID string, rerr error) {

966 // We only use smtputf8 if we have to, with a utf-8 localpart. For IDNA, we use ASCII domains.

967 smtputf8 = fromAddr.Localpart.IsInternational()

968 for _, r := range recipients {

969 if smtputf8 {

970 smtputf8 = r.Address.Localpart.IsInternational()

971 break

972 }

973 }

974 xc := message.NewComposer(mf, 100*1024*1024, smtputf8)

975 defer func() {

976 x := recover()

977 if x == nil {

978 return

979 }

980 if err, ok := x.(error); ok && errors.Is(err, message.ErrCompose) {

981 rerr = err

982 return

983 }

984 panic(x)

985 }()

986

987 xc.HeaderAddrs("From", []message.NameAddress{{Address: fromAddr}})

988 xc.HeaderAddrs("To", recipients)

989 xc.Header("Subject", subject)

990 messageID = fmt.Sprintf("<%s>", mox.MessageIDGen(xc.SMTPUTF8))

991 xc.Header("Message-Id", messageID)

992 xc.Header("Date", time.Now().Format(message.RFC5322Z))

993 xc.Header("User-Agent", "mox/"+moxvar.Version)

994 xc.Header("MIME-Version", "1.0")

995

996 textBody, ct, cte := xc.TextPart("plain", text)

997 xc.Header("Content-Type", ct)

998 xc.Header("Content-Transfer-Encoding", cte)

999 xc.Line()

1000 _, err := xc.Write(textBody)

1001 xc.Checkf(err, "writing text")

1002

1003 xc.Flush()

1004

1005 msgPrefix = dkimSign(ctx, log, fromAddr, smtputf8, mf)

1006

1007 return msgPrefix, xc.Has8bit, xc.SMTPUTF8, messageID, nil

1008}

1009

1010func dkimSign(ctx context.Context, log mlog.Log, fromAddr smtp.Address, smtputf8 bool, mf *os.File) string {

1011 // Add DKIM-Signature headers if we have a key for (a higher) domain than the from

1012 // address, which is a host name. A signature will only be useful with higher-level

1013 // domains if they have a relaxed dkim check (which is the default). If the dkim

1014 // check is strict, there is no harm, there will simply not be a dkim pass.

1015 fd := fromAddr.Domain

1016 var zerodom dns.Domain

1017 for fd != zerodom {

1018 confDom, ok := mox.Conf.Domain(fd)

1019 selectors := mox.DKIMSelectors(confDom.DKIM)

1020 if len(selectors) > 0 && !confDom.Disabled {

1021 dkimHeaders, err := dkim.Sign(ctx, log.Logger, fromAddr.Localpart, fd, selectors, smtputf8, mf)

1022 if err != nil {

1023 log.Errorx("dkim-signing dmarc report, continuing without signature", err)

1024 metricReportError.Inc()

1025 return ""

1026 }

1027 return dkimHeaders

1028 } else if ok {

1029 return ""

1030 }

1031

1032 var nfd dns.Domain

1033 _, nfd.ASCII, _ = strings.Cut(fd.ASCII, ".")

1034 _, nfd.Unicode, _ = strings.Cut(fd.Unicode, ".")

1035 fd = nfd

1036 }

1037 return ""

1038}

1039

1040// SuppressAdd adds an address to the suppress list.

1041func SuppressAdd(ctx context.Context, ba *SuppressAddress) error {

1042 return EvalDB.Insert(ctx, ba)

1043}

1044

1045// SuppressList returns all reporting addresses on the suppress list.

1046func SuppressList(ctx context.Context) ([]SuppressAddress, error) {

1047 return bstore.QueryDB[SuppressAddress](ctx, EvalDB).SortDesc("ID").List()

1048}

1049

1050// SuppressRemove removes a reporting address record from the suppress list.

1051func SuppressRemove(ctx context.Context, id int64) error {

1052 return EvalDB.Delete(ctx, &SuppressAddress{ID: id})

1053}

1054

1055// SuppressUpdate updates the until field of a reporting address record.

1056func SuppressUpdate(ctx context.Context, id int64, until time.Time) error {

1057 ba := SuppressAddress{ID: id}

1058 err := EvalDB.Get(ctx, &ba)

1059 if err != nil {

1060 return err

1061 }

1062 ba.Until = until

1063 return EvalDB.Update(ctx, &ba)

1064}

1065