diff --git a/crates/relay/src/deliver/mod.rs b/crates/relay/src/deliver/mod.rs index 5cc19a47..aa7bf3d7 100644 --- a/crates/relay/src/deliver/mod.rs +++ b/crates/relay/src/deliver/mod.rs @@ -368,6 +368,16 @@ impl Target<'_, R> { if let Some(failure) = failure { data.insert("reason".into(), json!(failure.reason)); } + // The policy itself, for the downstream's TLS-RPT reports + match (mode, sts) { + (TlsMode::Dane(tlsa), _) => { + data.insert("tlsa".into(), json!(tlsa.as_ref())); + } + (TlsMode::Verified, Some(policy)) => { + data.insert("sts".into(), json!(policy)); + } + _ => {} + } self.app.record("outbound.tls", data); } } diff --git a/crates/relay/src/deliver/tests.rs b/crates/relay/src/deliver/tests.rs index b40ee0e7..55d6e994 100644 --- a/crates/relay/src/deliver/tests.rs +++ b/crates/relay/src/deliver/tests.rs @@ -395,6 +395,9 @@ async fn dane_is_enforced() { let tls = events(&app, "outbound.tls"); assert_eq!(tls[0]["policyType"], "tlsa"); assert_eq!(tls[0]["result"], "success"); + // The records themselves, as the downstream's reports take them + let sent: Tlsa = serde_json::from_value(tls[0]["tlsa"].clone()).unwrap(); + assert_eq!(sent.entries.len(), 1); assert!( tls[0]["policyStrings"][0] .as_str() @@ -466,6 +469,12 @@ async fn mta_sts_enforce_is_enforced() { let tls = events(&app, "outbound.tls"); assert_eq!(tls[0]["policyType"], "sts", "{name}"); assert_ne!(tls[0]["result"], "success", "{name}"); + let sent: Policy = serde_json::from_value(tls[0]["sts"].clone()).unwrap(); + assert_eq!( + (sent.id.as_str(), sent.mode), + ("1", Mode::Enforce), + "{name}" + ); assert!( tls[0]["policyStrings"] .as_array() diff --git a/crates/services/src/inbuxa_relay_link.rs b/crates/services/src/inbuxa_relay_link.rs index 68fc4c9f..c62f7f5a 100644 --- a/crates/services/src/inbuxa_relay_link.rs +++ b/crates/services/src/inbuxa_relay_link.rs @@ -23,7 +23,8 @@ //! relay is told, so one offered again isn't delivered twice (RL-16). //! //! The relay's own events become this server's `relay.*` trace events, so -//! they reach the log, webhooks and OpenTelemetry. On a cluster only the +//! they reach the log, webhooks and OpenTelemetry, and its TLS results go +//! into this server's TLS-RPT reports. On a cluster only the //! linked node with the lowest id takes them. A node with no link to a //! joined relay for an hour says so with `relay.link-down` (RL-T23). @@ -73,7 +74,7 @@ use serde_json::{Value, json}; use smtp::{ core::Relayed, inbound::relayed::{RelayedMessage, RelayedOutcome, ingest_relayed}, - outbound::relay::{Applied, RelayedStatus, apply_relayed}, + outbound::relay::{Applied, RelayedStatus, apply_relayed, report_relayed_tls}, }; use std::{ collections::{BTreeMap, BTreeSet, HashMap, HashSet}, @@ -992,7 +993,7 @@ async fn session( answers: answers_tx, }; taker.take(&mut ws).await?; - take_events(&mut ws, data, &relay, status.current.node_id).await?; + take_events(&mut ws, server, &relay, status.current.node_id).await?; // Outbound mail can go through this link now (RL-24), until it drops let _sending = Sending::up(server, &relay, dialer.clone()); @@ -1019,7 +1020,7 @@ async fn session( taker.take(&mut ws).await?; } if ws.events_changed { - take_events(&mut ws, data, &relay, status.current.node_id).await?; + take_events(&mut ws, server, &relay, status.current.node_id).await?; } if ws.outbound_changed { follower.follow(&mut ws, server, &relay, status.current.node_id).await?; @@ -1058,7 +1059,7 @@ async fn session( refresh_status(&mut ws, data, &relay, &sent, status).await?; taker.relay_name = relay.name.clone(); taker.take(&mut ws).await?; - take_events(&mut ws, data, &relay, status.current.node_id).await?; + take_events(&mut ws, server, &relay, status.current.node_id).await?; follower .follow(&mut ws, server, &relay, status.current.node_id) .await?; @@ -1071,12 +1072,12 @@ async fn session( /// the relay can forget them. One node takes them: the lowest linked one. async fn take_events( ws: &mut Ws, - data: &store::Store, + server: &Server, relay: &Relay, node_id: u64, ) -> Result<(), LinkError> { ws.events_changed = false; - if !first_linked(data, relay.id, node_id).await? { + if !first_linked(server.store(), relay.id, node_id).await? { return Ok(()); } let answer = ws.ok(method::RELAY_EVENT_GET, json!({})).await?; @@ -1084,6 +1085,10 @@ async fn take_events( for event in answer["list"].as_array().into_iter().flatten() { if let Ok(event) = serde_json::from_value::(event.clone()) { trace_event(relay, &event); + // The relay's sessions count in this server's TLS-RPT reports + if event.kind == "outbound.tls" { + report_relayed_tls(server, &event.data).await; + } ids.push(event.id); } } diff --git a/crates/smtp/src/outbound/relay.rs b/crates/smtp/src/outbound/relay.rs index 9d242b63..ef41db7b 100644 --- a/crates/smtp/src/outbound/relay.rs +++ b/crates/smtp/src/outbound/relay.rs @@ -19,14 +19,25 @@ use crate::queue::{ Error, ErrorDetails, HostResponse, MessageWrapper, RCPT_RELAYED, Status, UnexpectedResponse, spool::SmtpSpool, }; +use crate::{queue::RecipientDomain, reporting::send::MtaReportSend}; use common::{Server, config::smtp::queue::QueueName, ipc::QueueEvent}; +use common::{ + config::smtp::report::AggregateFrequency, + ipc::{PolicyType, TlsEvent}, +}; use inbuxa_features::relay::outbound::{Handed, Refused, record_handed}; +use inbuxa_outbound_tls::{Policy, Tlsa}; use inbuxa_relay_proto::types::{Address, Envelope}; +use mail_auth::{ + mta_sts::TlsRpt, + report::tlsrpt::{FailureDetails, ResultType}, +}; +use serde_json::Value; use smtp_proto::{ MAIL_BODY_8BITMIME, MAIL_REQUIRETLS, MAIL_RET_FULL, MAIL_RET_HDRS, RCPT_NOTIFY_DELAY, RCPT_NOTIFY_FAILURE, RCPT_NOTIFY_NEVER, RCPT_NOTIFY_SUCCESS, Response, }; -use std::collections::BTreeMap; +use std::{collections::BTreeMap, sync::Arc}; use store::write::now; use trc::RelayEvent; use types::id::Id; @@ -355,10 +366,172 @@ pub async fn apply_relayed( Applied::Done } +/// One of a relay's `outbound.tls` events in this server's TLS-RPT reports +/// (RFC 8460), as if this server had made the session: the same policy, +/// result and frequency, for a domain that publishes a TLS-RPT record. +pub async fn report_relayed_tls(server: &Server, data: &serde_json::Map) { + let Some(tls) = relayed_tls(data) else { + return; + }; + let span_id = server.inner.data.span_id_gen.generate(); + let interval = match server + .eval_if( + &server.core.smtp.report.tls.send, + &RecipientDomain::new(&tls.domain), + span_id, + ) + .await + .unwrap_or(AggregateFrequency::Never) + { + interval @ (AggregateFrequency::Hourly + | AggregateFrequency::Daily + | AggregateFrequency::Weekly) => interval, + AggregateFrequency::Never => return, + }; + let record = match server + .core + .smtp + .resolvers + .dns + .txt_lookup::( + format!("_smtp._tls.{}.", tls.domain), + Some(&server.inner.cache.dns_txt), + ) + .await + { + Ok(record) => record, + // No record: nobody to report to + Err(_) => return, + }; + server + .schedule_report(TlsEvent { + domain: tls.domain, + policy: tls.policy, + failure: tls.failure, + tls_record: record, + interval, + span_id, + }) + .await; +} + +/// What an `outbound.tls` event says, as this server's report takes it. +#[derive(Debug, PartialEq)] +struct RelayedTls { + domain: String, + policy: PolicyType, + failure: Option, +} + +fn relayed_tls(data: &serde_json::Map) -> Option { + let text = |key: &str| data.get(key).and_then(Value::as_str); + let domain = text("domain")?.trim_end_matches('.').to_lowercase(); + if domain.is_empty() { + return None; + } + let policy = match text("policyType")? { + "tlsa" => PolicyType::Tlsa( + data.get("tlsa") + .and_then(|t| serde_json::from_value::(t.clone()).ok()) + .map(Arc::new), + ), + "sts" => PolicyType::Sts( + data.get("sts") + .and_then(|p| serde_json::from_value::(p.clone()).ok()) + .map(Arc::new), + ), + _ => PolicyType::None, + }; + let failure = match text("result")? { + "success" => None, + result => { + let result_type = + serde_json::from_value::(Value::String(result.to_string())) + .unwrap_or(ResultType::Other); + let mut failure = FailureDetails::new(result_type); + if let Some(mx) = text("mx") { + failure = failure.with_receiving_mx_hostname(mx); + } + if let Some(ip) = text("ip").and_then(|ip| ip.parse().ok()) { + failure = failure.with_receiving_ip(ip); + } + if let Some(reason) = text("reason") { + failure = failure.with_failure_reason_code(reason); + } + Some(failure) + } + }; + Some(RelayedTls { + domain, + policy, + failure, + }) +} + #[cfg(test)] mod tests { use super::*; + #[test] + fn relayed_tls_events() { + let event = |data: Value| relayed_tls(data.as_object().unwrap()); + let ok = event(serde_json::json!({ + "domain": "Example.ORG", "mx": "mx.example.org", "ip": "192.0.2.1", + "policyType": "no-policy-found", "policyStrings": [], "result": "success" + })) + .unwrap(); + assert_eq!(ok.domain, "example.org"); + assert_eq!((ok.policy, ok.failure), (PolicyType::None, None)); + + let tlsa = Tlsa { + entries: vec![inbuxa_outbound_tls::TlsaEntry { + is_end_entity: true, + is_spki: true, + matching: inbuxa_outbound_tls::TlsaMatching::Sha256, + data: vec![1, 2, 3], + }], + has_end_entities: true, + has_intermediates: false, + }; + let dane = event(serde_json::json!({ + "domain": "example.org", "mx": "mx.example.org", "ip": "192.0.2.1", + "policyType": "tlsa", "result": "validation-failure", + "reason": "no certificate matched", "tlsa": tlsa + })) + .unwrap(); + assert_eq!(dane.policy, PolicyType::Tlsa(Some(Arc::new(tlsa)))); + let failure = dane.failure.unwrap(); + assert_eq!(failure.result_type, ResultType::ValidationFailure); + assert_eq!( + failure.receiving_mx_hostname.as_deref(), + Some("mx.example.org") + ); + assert_eq!(failure.receiving_ip, Some("192.0.2.1".parse().unwrap())); + assert_eq!( + failure.failure_reason_code.as_deref(), + Some("no certificate matched") + ); + + let policy = Policy { + id: "20260101".into(), + mode: inbuxa_outbound_tls::Mode::Enforce, + mx: vec![inbuxa_outbound_tls::MxPattern::Equals( + "mx.example.org".into(), + )] + .into_boxed_slice(), + max_age: 86_400, + }; + let sts = event(serde_json::json!({ + "domain": "example.org", "mx": "other.example.org", + "policyType": "sts", "result": "something-new", "sts": policy + })) + .unwrap(); + assert_eq!(sts.policy, PolicyType::Sts(Some(Arc::new(policy)))); + assert_eq!(sts.failure.unwrap().result_type, ResultType::Other); + + assert!(event(serde_json::json!({"policyType": "tlsa", "result": "success"})).is_none()); + } + #[test] fn replies() { let r = parse_reply("550 5.1.1 No such user here"); diff --git a/docs/spec/features/relay.md b/docs/spec/features/relay.md index add4a4ae..dbe5962a 100644 --- a/docs/spec/features/relay.md +++ b/docs/spec/features/relay.md @@ -449,7 +449,10 @@ When no link is up, or after RL-13: downstream writes any DSN itself, and its history shows the remote server's real reply. The relay never writes a message of its own. - TLS results for each destination are sent as `RelayEvent`s. The downstream - builds TLS-RPT reports from them, as it does from its own deliveries. + builds TLS-RPT reports from them, as it does from its own deliveries: the + same report frequency, sent only for a domain that publishes a TLS-RPT + record. Each event carries the policy itself (`tlsa`, the TLSA records, + or `sts`, the MTA-STS policy) beside its `policyStrings`. - How a domain is tried: its MX hosts best first (no MX means the domain itself; a null MX, RFC 7505, refuses every recipient with `556 5.1.10`), at most five hosts and two addresses each. With the MX answer and the @@ -834,8 +837,11 @@ In progress. Done so far: - REQUIRETLS through relays: handed over with the parameter, delivered by the relay only over TLS with a verified certificate (tested against a scripted MX with a self-signed one). -- Still to come in step 4: the named `inbuxaRelay` route, and TLS-RPT - reports from the relay's `outbound.tls` events. +- TLS-RPT from relays: the node that takes a relay's events turns each + `outbound.tls` into a session in this server's TLS-RPT reports, with the + policy, result, MX host, address and reason (tested with a relay's + STARTTLS failure for REQUIRETLS mail). +- Still to come in step 4: the named `inbuxaRelay` route. - `crates/outbound-tls` (RL-30): the TLSA and MTA-STS types, the DANE certificate check, MTA-STS policy parsing and matching, and the DNSSEC lookups (with the bogus-answer recheck), moved out of `crates/smtp` and diff --git a/tests/src/system/relay.rs b/tests/src/system/relay.rs index 72dfc262..2336719e 100644 --- a/tests/src/system/relay.rs +++ b/tests/src/system/relay.rs @@ -21,12 +21,17 @@ use crate::utils::{ account::Account, + dns::DnsCache, server::{TestServer, TestServerBuilder}, }; use inbuxa_relay_proto::{time, token::JoinToken}; +use mail_auth::{common::parse::TxtRecordParser, mta_sts::TlsRpt}; use registry::schema::{ prelude::{ObjectType, Property}, - structs::{CertificateManagement, DkimManagement, DnsManagement, Domain, Tenant, UserRoles}, + structs::{ + CertificateManagement, DkimManagement, DnsManagement, Domain, Tenant, TlsInternalReport, + UserRoles, + }, }; use serde_json::{Value, json}; use std::str::FromStr; @@ -963,7 +968,13 @@ pub async fn link(test: &mut TestServer) { tokio::time::sleep(std::time::Duration::from_millis(500)).await; } // REQUIRETLS (RFC 8689): never sent in the clear, and this MX has no - // STARTTLS, so it waits, saying why + // STARTTLS, so it waits, saying why. The domain publishes TLS-RPT, so + // the relay's failed session lands in this server's report (RFC 8460) + test.server.txt_add( + "_smtp._tls.remote.test", + TlsRpt::parse(b"v=TLSRPTv1; rua=mailto:tls@remote.test").unwrap(), + std::time::Instant::now() + std::time::Duration::from_secs(3600), + ); submit_with( &jane, &identity, @@ -997,6 +1008,19 @@ pub async fn link(test: &mut TestServer) { .any(|m| m.contains("Subject: Strict one")), "sent in the clear" ); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30); + loop { + let reports = admin.registry_get_all::().await; + let text = format!("{reports:?}"); + if text.contains("remote.test") && text.contains("StartTlsNotSupported") { + break; + } + assert!( + std::time::Instant::now() < deadline, + "the relay's TLS failure is in a TLS-RPT report: {text}" + ); + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + } tokio::time::sleep(std::time::Duration::from_secs(2)).await; assert_eq!( subjects(&jane)