kumo_dmarc/
lib.rs

1#![allow(dead_code)]
2
3use crate::types::date_range::DateRange;
4use crate::types::feedback::Feedback;
5use crate::types::identifier::Identifier;
6use crate::types::mode::Mode;
7use crate::types::policy::Policy;
8use crate::types::policy_published::PolicyPublished;
9use crate::types::record::Record;
10use crate::types::report_failure::ReportFailure;
11use crate::types::report_metadata::ReportMetadata;
12use crate::types::results::{AuthResults, DmarcResult, PolicyEvaluated, Results, Row};
13pub use crate::types::results::{Disposition, DispositionWithContext};
14use bstr::BString;
15use chrono::{DateTime, Utc};
16use dns_resolver::Resolver;
17use mailparsing::AuthenticationResult;
18use serde::{Deserialize, Serialize};
19use std::collections::{BTreeMap, HashMap};
20use std::fs::File;
21use std::io::{BufRead, BufReader, Write};
22use std::net::IpAddr;
23use std::str::FromStr;
24use std::time::SystemTime;
25use uuid::Uuid;
26
27mod types;
28
29#[cfg(test)]
30mod tests;
31
32const DMARC_REPORT_LOG_FILEPATH: &str = "/var/log/kumomta/dmarc.log";
33
34pub struct DmarcPassContext {
35    /// Domain of the sender in the "From:"
36    pub from_domain: String,
37
38    /// Domain that provides the sought-after authorization information.
39    ///
40    /// The "MAIL FROM" email address if available.
41    pub mail_from_domain: Option<String>,
42
43    /// The envelope to
44    pub recipient_domain_list: Vec<String>,
45
46    /// The source IP address
47    pub received_from: String,
48
49    /// The results of the DKIM part of the checks
50    pub dkim_results: Vec<AuthenticationResult>,
51
52    /// The results of the SPF part of the checks
53    pub spf_result: AuthenticationResult,
54
55    /// The additional information needed to perform reporting
56    pub reporting_info: Option<ReportingInfo>,
57}
58
59impl DmarcPassContext {
60    pub async fn check(self, resolver: &dyn Resolver) -> DispositionWithContext {
61        let Self {
62            from_domain,
63            mail_from_domain,
64            recipient_domain_list: recipient_list,
65            received_from,
66            dkim_results,
67            spf_result,
68            reporting_info,
69        } = self;
70
71        let mut dmarc_context = DmarcContext::new(
72            &from_domain,
73            mail_from_domain.as_deref(),
74            &recipient_list[..],
75            received_from.as_str(),
76            &dkim_results[..],
77            &spf_result,
78            reporting_info.as_ref(),
79        );
80
81        dmarc_context.check(resolver).await
82    }
83}
84
85#[derive(Clone, Copy)]
86pub(crate) enum SenderDomainAlignment {
87    /// Sender domain is an exact match to the dmarc record
88    Exact,
89
90    /// Sender domain has no exact matching dmarc record
91    /// but its organizational domain does
92    OrganizationalDomain,
93}
94
95pub(crate) enum DmarcRecordResolution {
96    /// DNS could not be resolved at this time
97    TempError,
98
99    /// DNS was resolved, but no DMARC record was found
100    PermError,
101
102    /// DNS was resolved, and DMARC record was found
103    Records(Vec<Record>),
104}
105
106impl From<DmarcRecordResolution> for Disposition {
107    fn from(value: DmarcRecordResolution) -> Self {
108        match value {
109            DmarcRecordResolution::TempError => Disposition::TempError,
110            DmarcRecordResolution::PermError => Disposition::PermError,
111            DmarcRecordResolution::Records(_) => {
112                panic!("records must be parsed before being used in disposition")
113            }
114        }
115    }
116}
117
118#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
119#[serde(deny_unknown_fields)]
120pub struct ReportingInfo {
121    org_name: String,
122    email: String,
123    extra_contact_info: Option<String>,
124}
125
126/// The individual error records that are then aggregated for output in the report
127#[derive(Serialize, Deserialize, Clone)]
128pub(crate) struct ErrorRecord {
129    pub(crate) version: String,
130    pub(crate) org_name: String,
131    pub(crate) email: String,
132    pub(crate) extra_contact_info: Option<String>,
133    pub(crate) when: DateTime<Utc>,
134    pub(crate) error: String,
135    pub(crate) domain: String,
136    pub(crate) align_dkim: Option<Mode>,
137    pub(crate) align_spf: Option<Mode>,
138    pub(crate) policy: Policy,
139    pub(crate) subdomain_policy: Policy,
140    pub(crate) rate: u8,
141    pub(crate) report_failure: ReportFailure,
142    pub(crate) source_ip: IpAddr,
143    pub(crate) policy_evaluated: PolicyEvaluated,
144    pub(crate) identifier: Identifier,
145    pub(crate) auth_results: AuthResults,
146}
147
148struct DmarcContext<'a> {
149    pub(crate) from_domain: &'a str,
150    pub(crate) mail_from_domain: Option<&'a str>,
151    pub(crate) recipient_list: &'a [String],
152    pub(crate) received_from: &'a str,
153    pub(crate) now: SystemTime,
154    pub(crate) dkim_results: &'a [AuthenticationResult],
155    pub(crate) spf_result: &'a AuthenticationResult,
156    pub(crate) dkim_aligned: DmarcResult,
157    pub(crate) spf_aligned: DmarcResult,
158    pub(crate) reporting_info: Option<&'a ReportingInfo>,
159}
160
161impl<'a> DmarcContext<'a> {
162    /// Create a new evaluation context.
163    ///
164    /// - `from_domain` is the domain of the "From:" header
165    /// - `mail_from_domain` is the domain portion of the "MAIL FROM" identity
166    /// - `client_ip` is the IP address of the SMTP client that is emitting the mail
167    fn new(
168        from_domain: &'a str,
169        mail_from_domain: Option<&'a str>,
170        recipient_list: &'a [String],
171        received_from: &'a str,
172        dkim_results: &'a [AuthenticationResult],
173        spf_result: &'a AuthenticationResult,
174        reporting_info: Option<&'a ReportingInfo>,
175    ) -> DmarcContext<'a> {
176        Self {
177            from_domain,
178            mail_from_domain,
179            recipient_list,
180            received_from,
181            now: SystemTime::now(),
182            dkim_results,
183            spf_result,
184            dkim_aligned: DmarcResult::Pass,
185            spf_aligned: DmarcResult::Pass,
186            reporting_info,
187        }
188    }
189
190    pub async fn report_error(
191        &self,
192        record: &Record,
193        dmarc_domain: &str,
194        sender_domain_alignment: SenderDomainAlignment,
195        error: &str,
196    ) -> std::io::Result<()> {
197        let source_ip = self
198            .received_from
199            .parse()
200            .map_err(|x: std::net::AddrParseError| {
201                std::io::Error::new(std::io::ErrorKind::AddrNotAvailable, x.to_string())
202            })?;
203
204        if let Some(reporting_info) = self.reporting_info {
205            let error_record = ErrorRecord {
206                version: "1.0".to_string(),
207                org_name: reporting_info.org_name.to_string(),
208                email: reporting_info.email.to_string(),
209                extra_contact_info: reporting_info.extra_contact_info.to_owned(),
210                when: Utc::now(),
211                error: error.to_string(),
212                domain: dmarc_domain.to_string(),
213                align_dkim: Some(record.align_dkim),
214                align_spf: Some(record.align_spf),
215                policy: record.policy,
216                subdomain_policy: record.subdomain_policy.unwrap_or(record.policy),
217                rate: record.rate,
218                report_failure: record.report_failure,
219                source_ip,
220                policy_evaluated: PolicyEvaluated {
221                    disposition: record.policy_result(sender_domain_alignment),
222                    dkim: self.dkim_aligned,
223                    spf: self.spf_aligned,
224                    reason: vec![],
225                },
226                identifier: Identifier {
227                    envelope_to: self.recipient_list.into(),
228                    envelope_from: if let Some(mail_from_domain) = self.mail_from_domain {
229                        vec![mail_from_domain.into()]
230                    } else {
231                        vec![]
232                    },
233                    header_from: self.from_domain.into(),
234                },
235                auth_results: AuthResults {
236                    dkim: self.dkim_results.iter().map(|x| x.clone().into()).collect(),
237                    spf: vec![self.spf_result.clone().into()],
238                },
239            };
240
241            let result = serde_json::to_string(&error_record)?;
242
243            let mut f = File::options()
244                .append(true)
245                .open(DMARC_REPORT_LOG_FILEPATH)?;
246
247            writeln!(f, "{result}")?;
248        }
249
250        Ok(())
251    }
252
253    pub async fn aggregate(&self) -> anyhow::Result<()> {
254        let mut input_records = vec![];
255        let file = File::open(DMARC_REPORT_LOG_FILEPATH)?;
256        let lines = BufReader::new(file).lines();
257
258        for line in lines.map_while(Result::ok) {
259            let result: anyhow::Result<ErrorRecord> = serde_json::from_str::<ErrorRecord>(&line)
260                .map_err(|error| {
261                    anyhow::Error::new(error).context(format!(
262                        "Failed to decode a line from the DMARC report file \
263           {DMARC_REPORT_LOG_FILEPATH}. \
264           The line was: {line}. \
265           Is the file corrupt?"
266                    ))
267                });
268
269            input_records.push(result?);
270        }
271
272        let mut errors_grouped_by_email: HashMap<String, BTreeMap<IpAddr, Vec<ErrorRecord>>> =
273            HashMap::new();
274
275        for record in input_records {
276            let entry = errors_grouped_by_email.entry(record.email.clone());
277            let record_source_ip = record.source_ip;
278
279            entry
280                .and_modify(|entry| {
281                    entry
282                        .entry(record.source_ip)
283                        .and_modify(|x| x.push(record.clone()))
284                        .or_insert_with(|| vec![record.clone()]);
285                })
286                .or_insert({
287                    let mut new_group = BTreeMap::new();
288
289                    new_group.insert(record_source_ip, vec![record]);
290
291                    new_group
292                });
293        }
294
295        for (email, errors_grouped_by_ip) in errors_grouped_by_email.iter_mut() {
296            let mut errors = vec![];
297            let mut record = vec![];
298
299            //we know this is safe to do because for this list to be present, we will have found it earlier
300            let (_, first_records) = errors_grouped_by_ip
301                .iter()
302                .next()
303                .expect("guaranteed to not be empty by the logic above");
304
305            let first_record = &first_records[0];
306
307            let version = first_record.version.clone();
308            let org_name = first_record.org_name.clone();
309            let email = email.clone();
310            let extra_contact_info = first_record.extra_contact_info.clone();
311
312            let mut date_range = DateRange::new(first_record.when, first_record.when);
313
314            let report_id = Uuid::new_v4().to_string();
315
316            let domain = first_record.domain.clone();
317            let align_dkim = first_record.align_dkim;
318            let align_spf = first_record.align_spf;
319            let policy = first_record.policy;
320            let subdomain_policy = first_record.subdomain_policy;
321            let rate = first_record.rate;
322            let report_failure = first_record.report_failure;
323
324            for (ip, error_group_for_ip) in errors_grouped_by_ip.iter_mut() {
325                let row = Row {
326                    source_ip: *ip,
327                    count: error_group_for_ip.len() as u64,
328                    policy_evaluated: error_group_for_ip[0].policy_evaluated.clone(),
329                };
330
331                let mut results = Results {
332                    row,
333                    identifiers: Identifier {
334                        envelope_to: vec![],
335                        envelope_from: vec![],
336                        header_from: String::new(),
337                    },
338                    auth_results: AuthResults {
339                        dkim: vec![],
340                        spf: vec![],
341                    },
342                };
343
344                for group_error in error_group_for_ip.iter() {
345                    errors.push(group_error.error.clone());
346
347                    date_range.begin = std::cmp::min(date_range.begin, group_error.when);
348                    date_range.end = std::cmp::max(date_range.end, group_error.when);
349
350                    results
351                        .identifiers
352                        .envelope_from
353                        .extend_from_slice(&group_error.identifier.envelope_from);
354                    results
355                        .identifiers
356                        .envelope_to
357                        .extend_from_slice(&group_error.identifier.envelope_to);
358
359                    results
360                        .auth_results
361                        .dkim
362                        .extend_from_slice(&group_error.auth_results.dkim);
363                    results
364                        .auth_results
365                        .spf
366                        .extend_from_slice(&group_error.auth_results.spf);
367                }
368
369                record.push(results);
370            }
371
372            let _feedback = Feedback {
373                version,
374                metadata: ReportMetadata {
375                    org_name,
376                    email,
377                    extra_contact_info,
378                    report_id,
379                    date_range,
380                    error: errors,
381                },
382                policy: PolicyPublished::new(
383                    domain,
384                    align_dkim,
385                    align_spf,
386                    policy,
387                    subdomain_policy,
388                    rate,
389                    report_failure,
390                ),
391                record,
392            };
393
394            // if let Ok(result) = instant_xml::to_string(&feedback) {
395            //     println!("log: {}", result);
396            // }
397        }
398
399        Ok(())
400    }
401
402    pub async fn check(&mut self, resolver: &dyn Resolver) -> DispositionWithContext {
403        let dmarc_domain = format!("_dmarc.{}", self.from_domain);
404        match fetch_dmarc_records(&dmarc_domain, resolver).await {
405            DmarcRecordResolution::Records(records) => {
406                if let Some(record) = records.into_iter().next() {
407                    let mut result = record
408                        .evaluate(self, &dmarc_domain, SenderDomainAlignment::Exact)
409                        .await;
410                    result.props.extend(policy_tags(&record));
411                    return result;
412                }
413            }
414            x => {
415                let normalized_from = psl_utils::normalize_domain(self.from_domain);
416                if let Some(organizational_domain) = psl_utils::domain_str(&normalized_from) {
417                    if organizational_domain != normalized_from {
418                        let address = format!("_dmarc.{}", organizational_domain);
419                        match fetch_dmarc_records(&address, resolver).await {
420                            DmarcRecordResolution::TempError => {
421                                return DispositionWithContext {
422                                    result: Disposition::TempError,
423                                    context: format!(
424                                        "DNS records could not be resolved for {}",
425                                        address
426                                    ),
427                                    props: BTreeMap::new(),
428                                }
429                            }
430                            DmarcRecordResolution::PermError => {
431                                return DispositionWithContext {
432                                    result: Disposition::PermError,
433                                    context: format!("no DMARC records found for {}", address),
434                                    props: BTreeMap::new(),
435                                }
436                            }
437                            DmarcRecordResolution::Records(records) => {
438                                if let Some(record) = records.into_iter().next() {
439                                    let mut result = record
440                                        .evaluate(
441                                            self,
442                                            &address,
443                                            SenderDomainAlignment::OrganizationalDomain,
444                                        )
445                                        .await;
446                                    result.props.extend(policy_tags(&record));
447                                    return result;
448                                }
449                            }
450                        }
451                    } else {
452                        return DispositionWithContext {
453                            result: x.into(),
454                            context: format!("no DMARC records found for {}", &self.from_domain),
455                            props: BTreeMap::new(),
456                        };
457                    }
458                }
459            }
460        }
461
462        DispositionWithContext {
463            result: Disposition::None,
464            context: format!("no DMARC records found for {}", &self.from_domain),
465            props: BTreeMap::new(),
466        }
467    }
468}
469
470fn policy_tags(record: &Record) -> BTreeMap<String, BString> {
471    let mut props = BTreeMap::new();
472
473    for (tag, value) in record.tags() {
474        props.insert(format!("policy.{tag}"), value.as_str().into());
475    }
476
477    props
478}
479
480// The output is wrapped in a Result to allow matching on errors.
481// Returns an Iterator to the Reader of the lines of the file.
482fn read_lines<P>(filename: P) -> std::io::Result<std::io::Lines<std::io::BufReader<File>>>
483where
484    P: AsRef<std::path::Path>,
485{
486    let file = File::open(filename)?;
487    Ok(std::io::BufReader::new(file).lines())
488}
489
490pub(crate) async fn fetch_dmarc_records(
491    address: &str,
492    resolver: &dyn Resolver,
493) -> DmarcRecordResolution {
494    let initial_txt = match resolver.resolve_txt(address).await {
495        Ok(answer) => {
496            if answer.records.is_empty() || answer.nxdomain {
497                return DmarcRecordResolution::PermError;
498            } else {
499                answer.as_txt()
500            }
501        }
502        Err(_) => {
503            return DmarcRecordResolution::TempError;
504        }
505    };
506
507    let mut records = vec![];
508
509    // TXT records can contain all sorts of stuff, let's walk through
510    // the set that we retrieved and take the first one that parses
511    for txt in initial_txt {
512        if txt.starts_with("v=DMARC1;") {
513            if let Ok(record) = Record::from_str(&txt) {
514                records.push(record);
515            }
516        }
517    }
518
519    if records.is_empty() {
520        return DmarcRecordResolution::PermError;
521    }
522
523    DmarcRecordResolution::Records(records)
524}
kumo_dmarc/lib.rs

kumo_dmarc/
lib.rs
