kumo_log_types/
lib.rs

1use crate::rfc5965::ARFReport;
2use bounce_classify::BounceClass;
3use chrono::{DateTime, Utc};
4use kumo_address::host_or_socket::HostOrSocketAddress;
5use kumo_address::socket::SocketAddress;
6use rfc5321::Response;
7use serde::{Deserialize, Serialize};
8use serde_json::Value;
9use serde_with::formats::PreferOne;
10use serde_with::{serde_as, OneOrMany};
11use std::borrow::Cow;
12use std::collections::HashMap;
13use std::net::SocketAddr;
14use uuid::Uuid;
15
16pub mod rfc3464;
17pub mod rfc5965;
18
19#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
20pub struct ResolvedAddress {
21    pub name: String,
22    pub addr: HostOrSocketAddress,
23    /// Whether the address was resolved via a DNSSEC-validated (secure) chain.
24    /// Used to decide DANE eligibility for the connection. Defaults to false
25    /// for addresses that did not originate from a secure DNS lookup (literals,
26    /// inbound peers, etc).
27    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
28    pub is_secure: bool,
29}
30
31impl std::fmt::Display for ResolvedAddress {
32    fn fmt(&self, fmt: &mut std::fmt::Formatter) -> std::fmt::Result {
33        let addr = format!("{}", self.addr);
34        if addr == self.name {
35            // likely: unix domain socket path
36            write!(fmt, "{addr}")
37        } else {
38            write!(fmt, "{}/{addr}", self.name)
39        }
40    }
41}
42
43#[derive(Serialize, Deserialize, Debug, Copy, Clone, Eq, PartialEq, Hash, Ord, PartialOrd)]
44pub enum RecordType {
45    /// Recorded by a receiving listener
46    Reception,
47    /// Recorded by the delivery side, most likely as a
48    /// result of attempting a delivery to a remote host
49    Delivery,
50    Bounce,
51    TransientFailure,
52    /// Recorded when a message is expiring from the queue
53    Expiration,
54    /// Administratively failed
55    AdminBounce,
56    /// Contains information about an OOB bounce
57    OOB,
58    /// Contains a feedback report
59    Feedback,
60
61    /// SMTP Listener responded with a 4xx or 5xx
62    Rejection,
63
64    /// Administratively rebound from one queue to another
65    AdminRebind,
66
67    /// Moved from the special deferred injection queue
68    /// and into some other queue
69    DeferredInjectionRebind,
70
71    /// Explains why a message was put into the scheduled queue
72    Delayed,
73
74    /// Message was transferred to another kumomta node,
75    /// which is now responsible for delivery.
76    XferOut,
77
78    /// Message was received from another kumomta node,
79    /// and we are now responsible for delivery.
80    XferIn,
81
82    /// Special for matching anything in the logging config
83    Any,
84}
85
86impl RecordType {
87    /// Returns true if it makes sense to run the corresponding record
88    /// through the bounce classifier module.
89    /// The rule of thumb for that is if the response came from the
90    /// destination when attempting delivery, but we also include
91    /// administrative bounces and message expirations.
92    pub const fn is_bounce_classifiable(&self) -> bool {
93        match self {
94            Self::Any
95            | Self::Reception
96            | Self::Delivery
97            | Self::DeferredInjectionRebind
98            | Self::AdminRebind
99            | Self::XferOut
100            | Self::XferIn
101            | Self::Delayed => false,
102            Self::Bounce
103            | Self::TransientFailure
104            | Self::Expiration
105            | Self::AdminBounce
106            | Self::OOB
107            | Self::Feedback
108            | Self::Rejection => true,
109        }
110    }
111}
112
113/// Unfortunately, when we defined the `timestamp` and `created` fields
114/// in the log structure, we made the decision to log as the integer
115/// unix timestamp format, which causes us to discard the sub-second
116/// information that we otherwise have available.
117///
118/// We'd like to now include the full info in the serialized log
119/// record, without bloating the in-memory representation, or otherwise
120/// explicitly duplicating data to arrange for serde to emit it for us.
121///
122/// That's where this macro comes in; it allows us to serialize those
123/// fields via a proxy type that effectively causes serde to emit two
124/// different serializations of the same value.
125///
126/// Usage is: `ts_serializer(MODULE_NAME, STRUCT_NAME, SECONDS_FIELD, FULL_FIELD)`
127///
128/// The MODULE_NAME and STRUCT_NAME are not especially important and
129/// are really present just for namespacing.
130///
131/// The SECONDS_FIELD defines the name of the field to be emitted
132/// as the unix timestamp (in seconds).
133///
134/// The FULL_FIELD defines the name of the field to be emitted
135/// as the full RFC 3339 datetime.
136///
137/// The macro defines a module and struct that can be used as a proxy
138/// for serialization.
139///
140/// To actually use it, you need to annotate the field in the struct;
141///
142/// ```norun
143/// #[serde(flatten, with = "ts_serializer")]
144/// pub timestamp: DateTime<Utc>,
145/// ```
146///
147/// It is important that `flatten` is used to avoid serde emitting
148/// a nested/child struct, and the `with` attribute is what points
149/// the serialization to the defined proxy module; it must
150/// reference the MODULE_NAME you defined.
151macro_rules! ts_serializer {
152    ($module:ident, $name:ident, $seconds:ident, $full:ident) => {
153        mod $module {
154            use super::*;
155            use serde::{Deserializer, Serializer};
156
157            #[derive(Serialize, Deserialize, Copy, Clone, Eq, Hash, Default, Debug, PartialEq)]
158            struct $name {
159                #[serde(with = "chrono::serde::ts_seconds")]
160                pub $seconds: DateTime<Utc>,
161
162                /// Optional for backwards compatibility: we don't
163                /// expect $full to be present, but we'll take it
164                /// if it is!
165                #[serde(default)]
166                pub $full: Option<DateTime<Utc>>,
167            }
168
169            impl std::ops::Deref for $name {
170                type Target = DateTime<Utc>;
171                fn deref(&self) -> &DateTime<Utc> {
172                    self.$full.as_ref().unwrap_or(&self.$seconds)
173                }
174            }
175
176            impl<T: chrono::TimeZone> From<DateTime<T>> for $name
177            where
178                DateTime<Utc>: From<DateTime<T>>,
179            {
180                fn from(value: DateTime<T>) -> $name {
181                    let timestamp: DateTime<Utc> = value.into();
182                    $name {
183                        $seconds: timestamp,
184                        $full: Some(timestamp),
185                    }
186                }
187            }
188
189            impl From<$name> for DateTime<Utc> {
190                fn from(value: $name) -> DateTime<Utc> {
191                    *value
192                }
193            }
194
195            pub fn serialize<S>(d: &DateTime<Utc>, s: S) -> Result<S::Ok, S::Error>
196            where
197                S: Serializer,
198            {
199                let proxy: $name = (*d).into();
200                proxy.serialize(s)
201            }
202
203            pub fn deserialize<'a, D>(d: D) -> Result<DateTime<Utc>, D::Error>
204            where
205                D: Deserializer<'a>,
206            {
207                $name::deserialize(d).map(|p| p.into())
208            }
209        }
210    };
211}
212
213ts_serializer!(ts_serializer, TimestampSerializer, timestamp, event_time);
214ts_serializer!(ct_serializer, CreationTimeSerializer, created, created_time);
215
216#[serde_as]
217#[derive(Serialize, Deserialize, Debug, Clone)]
218pub struct JsonLogRecord {
219    /// What kind of record this is
220    #[serde(rename = "type")]
221    pub kind: RecordType,
222    /// The message id
223    pub id: String,
224    /// The envelope sender
225    pub sender: String,
226    /// The envelope recipient
227    #[serde_as(as = "OneOrMany<_, PreferOne>")]
228    pub recipient: Vec<String>,
229    /// Which named queue the message was associated with
230    pub queue: String,
231    /// Which MX site the message was being delivered to
232    pub site: String,
233    /// The size of the message, in bytes
234    pub size: u64,
235    /// The response from/to the peer
236    pub response: Response,
237    /// The address of the peer, and our sense of its
238    /// hostname or EHLO domain
239    pub peer_address: Option<ResolvedAddress>,
240    /// The time at which we are logging this event
241    #[serde(flatten, with = "ts_serializer")]
242    pub timestamp: DateTime<Utc>,
243    /// The time at which the message was initially received and created
244    #[serde(flatten, with = "ct_serializer")]
245    pub created: DateTime<Utc>,
246    /// The number of delivery attempts that have been made.
247    /// Note that this may be approximate after a restart; use the
248    /// number of logged events to determine the true number
249    pub num_attempts: u16,
250
251    pub bounce_classification: BounceClass,
252
253    pub egress_pool: Option<String>,
254    pub egress_source: Option<String>,
255    pub source_address: Option<MaybeProxiedSourceAddress>,
256
257    pub feedback_report: Option<Box<ARFReport>>,
258
259    pub meta: HashMap<String, Value>,
260    pub headers: HashMap<String, Value>,
261
262    /// The protocol used to deliver, or attempt to deliver, this message
263    pub delivery_protocol: Option<String>,
264
265    /// The protocol used to receive this message
266    pub reception_protocol: Option<String>,
267
268    /// The id of the node on which the event occurred
269    pub nodeid: Uuid,
270
271    /// The TLS Cipher used, if applicable
272    #[serde(skip_serializing_if = "Option::is_none")]
273    pub tls_cipher: Option<String>,
274
275    /// The TLS protocol version used, if applicable
276    #[serde(skip_serializing_if = "Option::is_none")]
277    pub tls_protocol_version: Option<String>,
278
279    /// The Subject Name from the peer TLS certificate, if applicable
280    #[serde(skip_serializing_if = "Option::is_none")]
281    pub tls_peer_subject_name: Option<Vec<String>>,
282
283    /// The provider name, if any.
284    /// This is a way of grouping destination sites operated
285    /// by the same provider.
286    #[serde(skip_serializing_if = "Option::is_none")]
287    pub provider_name: Option<String>,
288
289    /// Uuid identifying a connection/session for either inbound
290    /// or outbound (depending on the type of the record).
291    /// This is useful when correlating a series of messages to
292    /// the same connection for either ingress or egress
293    #[serde(skip_serializing_if = "Option::is_none")]
294    pub session_id: Option<Uuid>,
295}
296
297#[derive(Debug, Clone, Serialize, Deserialize)]
298pub struct MaybeProxiedSourceAddress {
299    pub address: SocketAddress,
300    #[serde(skip_serializing_if = "Option::is_none")]
301    pub server: Option<SocketAddr>,
302    #[serde(skip_serializing_if = "Option::is_none")]
303    pub protocol: Option<Cow<'static, str>>,
304}
305
306#[cfg(all(test, target_pointer_width = "64"))]
307#[test]
308fn sizes() {
309    assert_eq!(std::mem::size_of::<JsonLogRecord>(), 720);
310}
kumo_log_types/lib.rs

kumo_log_types/
lib.rs
