1package dmarcdb

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

5import (

6 "compress/gzip"

7 "context"

8 "encoding/xml"

9 "errors"

10 "fmt"

11 "io"

12 "log/slog"

13 "mime"

14 "mime/multipart"

15 "net/textproto"

16 "net/url"

17 "os"

18 "runtime/debug"

19 "slices"

20 "sort"

21 "strings"

22 "sync"

23 "time"

25 "golang.org/x/exp/maps"

27 "github.com/prometheus/client_golang/prometheus"

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

30 "github.com/mjl-/bstore"

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

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

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

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

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

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

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

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

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

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

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

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

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

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

46)

48var (

49 metricReport = promauto.NewCounter(

50 prometheus.CounterOpts{

51 Name: "mox_dmarcdb_report_queued_total",

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

53 },

54 )

55 metricReportError = promauto.NewCounter(

56 prometheus.CounterOpts{

57 Name: "mox_dmarcdb_report_error_total",

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

59 },

60 )

61)

63var (

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

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

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

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

68 EvalDB *bstore.DB

69)

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

72// in a DMARC report.

73type Evaluation struct {

74 ID int64

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

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

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

79 PolicyDomain string `bstore:"index"`

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

82 // evaluation will be included in.

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

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

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

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

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

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

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

91 // reports.

92 Optional bool

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

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

96 IntervalHours int

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

99 Addresses []string

100

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

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

103 PolicyPublished dmarcrpt.PolicyPublished

104

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

106 SourceIP string

107 Disposition dmarcrpt.Disposition

108 AlignedDKIMPass bool

109 AlignedSPFPass bool

110 OverrideReasons []dmarcrpt.PolicyOverrideReason

111

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

113 EnvelopeTo string

114 EnvelopeFrom string

115 HeaderFrom string

116

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

118 DKIMResults []dmarcrpt.DKIMAuthResult

119 SPFResults []dmarcrpt.SPFAuthResult

120}

121

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

123// will be suppressed for a period.

124type SuppressAddress struct {

125 ID int64

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

127 ReportingAddress string `bstore:"unique"`

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

129 Comment string

130}

131

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

133 false: dmarcrpt.DMARCFail,

134 true: dmarcrpt.DMARCPass,

135}

136

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

138// report.

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

140 return dmarcrpt.ReportRecord{

141 Row: dmarcrpt.Row{

142 SourceIP: e.SourceIP,

143 Count: count,

144 PolicyEvaluated: dmarcrpt.PolicyEvaluated{

145 Disposition: e.Disposition,

146 DKIM: dmarcResults[e.AlignedDKIMPass],

147 SPF: dmarcResults[e.AlignedSPFPass],

148 Reasons: e.OverrideReasons,

149 },

150 },

151 Identifiers: dmarcrpt.Identifiers{

152 EnvelopeTo: e.EnvelopeTo,

153 EnvelopeFrom: e.EnvelopeFrom,

154 HeaderFrom: e.HeaderFrom,

155 },

156 AuthResults: dmarcrpt.AuthResults{

157 DKIM: e.DKIMResults,

158 SPF: e.SPFResults,

159 },

160 }

161}

162

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

164

165func intervalHours(seconds int) int {

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

167 for _, opt := range intervalOpts {

168 if hours >= opt {

169 return opt

170 }

171 }

172 return 1

173}

174

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

176// to the database.

177//

178// AddEvaluation sets Evaluation.IntervalHours based on

179// aggregateReportingIntervalSeconds.

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

181 e.IntervalHours = intervalHours(aggregateReportingIntervalSeconds)

182

183 e.ID = 0

184 return EvalDB.Insert(ctx, e)

185}

186

187// Evaluations returns all evaluations in the database.

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

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

190 q.SortAsc("Evaluated")

191 return q.List()

192}

193

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

195// aggregate report, for a domain.

196type EvaluationStat struct {

197 Domain dns.Domain

198 Dispositions []string

199 Count int

200 SendReport bool

201}

202

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

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

205 r := map[string]EvaluationStat{}

206

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

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

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

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

211 }

212 stat.Count++

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

214 r[e.PolicyDomain] = stat

215 } else {

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

217 if err != nil {

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

219 }

220 r[e.PolicyDomain] = EvaluationStat{

221 Domain: dom,

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

223 Count: 1,

224 SendReport: !e.Optional,

225 }

226 }

227 return nil

228 })

229 return r, err

230}

231

232// EvaluationsDomain returns all evaluations for a domain.

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

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

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

236 q.SortAsc("Evaluated")

237 return q.List()

238}

239

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

241// an aggregate report.

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

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

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

245 _, err := q.Delete()

246 return err

247}

248

249var jitterRand = mox.NewPseudoRand()

250

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

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

253// already be doing that.

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

255 return time.Until(t.Add(time.Duration(30+jitterRand.Intn(60)) * time.Second))

256}

257

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

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

260func Start(resolver dns.Resolver) {

261 go func() {

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

263

264 defer func() {

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

266 x := recover()

267 if x != nil {

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

269 debug.PrintStack()

270 metrics.PanicInc(metrics.Dmarcdb)

271 }

272 }()

273

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

275 defer timer.Stop()

276

277 ctx := mox.Shutdown

278

279 for {

280 now := time.Now()

281 nextEnd := nextWholeHour(now)

282 timer.Reset(jitteredTimeUntil(nextEnd))

283

284 select {

285 case <-ctx.Done():

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

287 return

288 case <-timer.C:

289 }

290

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

292 // divide 24, starting from UTC.

293 // ../rfc/7489:1750

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

295 if utchour == 0 {

296 utchour = 24

297 }

298 intervals := []int{}

299 for _, ival := range intervalOpts {

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

301 intervals = append(intervals, ival)

302 }

303 }

304 intervals = append(intervals, 1)

305

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

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

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

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

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

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

312

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

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

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

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

317 metricReportError.Inc()

318 } else {

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

320 }

321 }

322 }()

323}

324

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

326 t := now

327 t = t.Add(time.Hour)

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

329}

330

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

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

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

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

335// Replaced by tests.

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

337 t := time.NewTimer(between)

338 select {

339 case <-ctx.Done():

340 t.Stop()

341 return false

342 case <-t.C:

343 return true

344 }

345}

346

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

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

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

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

351 for i, v := range intervals {

352 ivals[i] = v

353 }

354

355 destDomains := map[string]bool{}

356

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

358 nsend := 0

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

360 q.FilterLess("Evaluated", endTime)

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

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

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

364 nsend++

365 }

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

367 return nil

368 })

369 if err != nil {

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

371 }

372

373 var wg sync.WaitGroup

374

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

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

377 between := 45 * time.Minute

378 if nsend > 0 {

379 between /= time.Duration(nsend)

380 }

381 if between > 5*time.Minute {

382 between = 5 * time.Minute

383 }

384

385 // Attempt to send report to each domain.

386 n := 0

387 for d, send := range destDomains {

388 // Cleanup evaluations for domain with only optionals.

389 if !send {

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

391 continue

392 }

393

394 if n > 0 {

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

396 return nil

397 }

398 }

399 n++

400

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

402 wg.Add(1)

403 go func(domain string) {

404 defer func() {

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

406 x := recover()

407 if x != nil {

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

409 debug.PrintStack()

410 metrics.PanicInc(metrics.Dmarcdb)

411 }

412 }()

413 defer wg.Done()

414

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

416 rlog.Info("sending dmarc report")

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

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

419 metricReportError.Inc()

420 }

421 }(d)

422 }

423

424 wg.Wait()

425

426 return nil

427}

428

429type recipient struct {

430 address smtp.Address

431 maxSize uint64

432}

433

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

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

436

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

438 if err != nil {

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

440 return r, false

441 }

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

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

444 return r, false

445 }

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

447 if err != nil {

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

449 return r, false

450 }

451

452 r = recipient{addr, uri.MaxSize}

453 // ../rfc/7489:1197

454 switch uri.Unit {

455 case "k", "K":

456 r.maxSize *= 1024

457 case "m", "M":

458 r.maxSize *= 1024 * 1024

459 case "g", "G":

460 r.maxSize *= 1024 * 1024 * 1024

461 case "t", "T":

462 // Oh yeah, terabyte-sized reports!

463 r.maxSize *= 1024 * 1024 * 1024 * 1024

464 case "":

465 default:

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

467 return r, false

468 }

469

470 return r, true

471}

472

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

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

475 q.FilterLess("Evaluated", endTime)

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

477 _, err := q.Delete()

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

479}

480

481// replaceable for testing.

482var queueAdd = queue.Add

483

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

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

486 if err != nil {

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

488 }

489

490 // We'll cleanup records by default.

491 cleanup = true

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

493 tempError := false

494

495 defer func() {

496 if !cleanup || tempError {

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

498 } else {

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

500 }

501 }()

502

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

504 report := dmarcrpt.Feedback{

505 Version: "1.0",

506 ReportMetadata: dmarcrpt.ReportMetadata{

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

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

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

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

511 },

512 // We'll fill the records below.

513 Records: []dmarcrpt.ReportRecord{},

514 }

515

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

517

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

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

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

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

522 // ../rfc/7489:1714

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

524 if err != nil {

525 // 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).

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

527 cleanup = status != dmarc.StatusTemperror

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

529 }

530

531 var recipients []recipient

532

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

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

535 // and possibly update the list.

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

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

538 if !ok {

539 continue

540 }

541

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

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

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

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

546

547 if ruaOrgDom == evalOrgDom {

548 recipients = append(recipients, r)

549 continue

550 }

551

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

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

554 // ../rfc/7489:1556

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

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

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

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

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

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

561 if status == dmarc.StatusTemperror {

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

563 // perhaps there are multiple recipients.

564 // ../rfc/7489:1578

565 tempError = true

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

567 }

568 if !accepts {

569 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))

570 continue

571 }

572

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

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

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

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

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

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

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

580 // ../rfc/7489:1600

581 foundReplacement := false

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

583 for _, record := range records {

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

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

586 if !ok {

587 continue

588 }

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

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

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

592 } else {

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

594 foundReplacement = true

595 recipients = append(recipients, extr)

596 }

597 }

598 }

599 if !foundReplacement {

600 recipients = append(recipients, r)

601 }

602 }

603

604 if len(recipients) == 0 {

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

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

607 return true, nil

608 }

609

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

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

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

613 type recordCount struct {

614 dmarcrpt.ReportRecord

615 count int

616 }

617 counts := map[string]recordCount{}

618

619 var first, last Evaluation // For daterange.

620 var sendReport bool

621

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

623 q.FilterLess("Evaluated", endTime)

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

625 q.SortAsc("Evaluated")

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

627 if first.ID == 0 {

628 first = e

629 }

630 last = e

631

632 record := e.ReportRecord(0)

633

634 // 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.

635

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

637 // records into smaller representation in the future.

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

639 if err != nil {

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

641 }

642 recstr := string(recbuf)

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

644 if !e.Optional {

645 sendReport = true

646 }

647 return nil

648 })

649 if err != nil {

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

651 }

652

653 if !sendReport {

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

655 return true, nil

656 }

657

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

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

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

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

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

663 //

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

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

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

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

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

669 beginTime := endTime.Add(-interval)

670 for first.Evaluated.Before(beginTime) {

671 beginTime = beginTime.Add(-interval)

672 }

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

674

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

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

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

678 // but shouldn't hurt.

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

680

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

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

683 // lookup errors during incoming deliveries.

684 report.ReportMetadata.Errors = errors

685

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

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

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

689 report.PolicyPublished = last.PolicyPublished

690

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

692 recstrs := maps.Keys(counts)

693 sort.Strings(recstrs)

694 for _, recstr := range recstrs {

695 rc := counts[recstr]

696 rc.ReportRecord.Row.Count = rc.count

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

698 }

699

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

701 if err != nil {

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

703 }

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

705

706 gzw := gzip.NewWriter(reportFile)

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

708 enc := xml.NewEncoder(gzw)

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

710 if err == nil {

711 err = enc.Encode(report)

712 }

713 if err == nil {

714 err = enc.Close()

715 }

716 if err == nil {

717 err = gzw.Close()

718 }

719 if err != nil {

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

721 }

722

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

724 if err != nil {

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

726 }

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

728

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

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

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

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

733 // to multiple of our configured domains.

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

735

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

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

738

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

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

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

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

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

744

745Report domain: %s

746Submitter: %s

747Report-ID: %s

748Period: %s - %s UTC

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

750

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

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

753

754 var addrs []message.NameAddress

755 for _, rcpt := range recipients {

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

757 }

758

759 // Compose the message.

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

761 if err != nil {

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

763 }

764

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

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

767 // accept the report, queue it.

768 msgInfo, err := msgf.Stat()

769 if err != nil {

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

771 }

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

773 var queued bool

774 for _, rcpt := range recipients {

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

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

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

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

779 exists, err := q.Exists()

780 if err != nil {

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

782 }

783 if exists {

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

785 continue

786 }

787

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

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

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

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

792 // ../rfc/7489:1773

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

794 continue

795 }

796

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

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

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

800 qm.MaxAttempts = 5

801 qm.IsDMARCReport = true

802

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

804 if err != nil {

805 tempError = true

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

807 metricReportError.Inc()

808 } else {

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

810 queued = true

811 metricReport.Inc()

812 }

813 }

814

815 if !queued {

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

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

818 metricReportError.Inc()

819 }

820 }

821

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

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

824 // ../rfc/7489:1785

825

826 return true, nil

827}

828

829func 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) {

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

831 smtputf8 = fromAddr.Localpart.IsInternational()

832 for _, r := range recipients {

833 if smtputf8 {

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

835 break

836 }

837 }

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

839 defer func() {

840 x := recover()

841 if x == nil {

842 return

843 }

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

845 rerr = err

846 return

847 }

848 panic(x)

849 }()

850

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

852 xc.HeaderAddrs("To", recipients)

853 xc.Subject(subject)

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

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

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

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

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

859

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

861 mp := multipart.NewWriter(xc)

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

863 xc.Line()

864

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

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

867 textHdr := textproto.MIMEHeader{}

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

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

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

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

872 _, err = textp.Write(textBody)

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

874

875 // DMARC report as attachment.

876 ahdr := textproto.MIMEHeader{}

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

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

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

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

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

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

883 wc := moxio.Base64Writer(ap)

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

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

886 err = wc.Close()

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

888

889 err = mp.Close()

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

891

892 xc.Flush()

893

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

895

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

897}

898

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

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

901// ../rfc/7489:1918

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

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

904

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

906 if err != nil {

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

908 }

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

910

911 var recipientStrs []string

912 for _, rcpt := range recipients {

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

914 }

915

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

917 // ../rfc/7489:1926

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

919Report-Domain: %s

920Report-ID: %s

921Report-Size: %d

922Submitter: %s

923Submitting-URI: %s

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

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

926

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

928 if err != nil {

929 return err

930 }

931

932 msgInfo, err := msgf.Stat()

933 if err != nil {

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

935 }

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

937

938 for _, rcpt := range recipients {

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

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

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

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

943 exists, err := q.Exists()

944 if err != nil {

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

946 }

947 if exists {

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

949 continue

950 }

951

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

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

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

955 qm.MaxAttempts = 5

956 qm.IsDMARCReport = true

957

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

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

960 metricReportError.Inc()

961 } else {

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

963 metricReport.Inc()

964 }

965 }

966 return nil

967}

968

969func 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) {

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

971 smtputf8 = fromAddr.Localpart.IsInternational()

972 for _, r := range recipients {

973 if smtputf8 {

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

975 break

976 }

977 }

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

979 defer func() {

980 x := recover()

981 if x == nil {