bacnet_objects/audit/
notification.rs

1use bacnet_types::primitives::{BACnetTimeStamp, Date, Time};
2
3use super::*;
4
5/// Mutable, object-owned receiver for one authorized Audit notification batch.
6///
7/// Implementations must commit the complete prospective state before exposing
8/// any in-memory change. The server uses this type-erased channel instead of
9/// downcasting an Audit Log object or taking ownership of its persistence.
10pub trait AuditLogNotificationSink: Send + Sync {
11    /// Whether this sink currently accepts Audit notifications.
12    fn notification_logging_enabled(&self) -> bool;
13
14    /// Merge or create every notification in wire order as one durable batch.
15    ///
16    /// `apdu_timeout_ms` is the local Device object's configured APDU_Timeout.
17    /// It supplies the Clause 12.64 complementary timestamp window.
18    fn store_notifications(
19        &mut self,
20        notifications: &[BACnetAuditNotification],
21        apdu_timeout_ms: u32,
22    ) -> Result<(), Error>;
23
24    /// Store a batch and report whether retained record content changed.
25    /// Custom sinks must opt in to change reporting before they can forward.
26    fn store_notifications_with_change(
27        &mut self,
28        notifications: &[BACnetAuditNotification],
29        apdu_timeout_ms: u32,
30    ) -> Result<bool, Error> {
31        self.store_notifications(notifications, apdu_timeout_ms)?;
32        Ok(false)
33    }
34
35    /// Store records and receipt atomically, reporting record changes separately
36    /// from receipt-only commits. A duplicate must always report `false`.
37    fn store_confirmed_notifications_with_change(
38        &mut self,
39        notifications: &[BACnetAuditNotification],
40        apdu_timeout_ms: u32,
41        receipt: CompletedAuditReceipt,
42    ) -> Result<(ConfirmedAuditNotificationOutcome, bool), Error> {
43        self.store_confirmed_notifications(notifications, apdu_timeout_ms, receipt)
44            .map(|outcome| (outcome, false))
45    }
46
47    /// Check the durable completed confirmed receipts without mutating them.
48    fn has_completed_confirmed_receipt(
49        &self,
50        _key: &[u8],
51        _now_unix_millis: u64,
52    ) -> Result<bool, Error> {
53        Ok(false)
54    }
55
56    /// Atomically merge one confirmed batch and retain its completed receipt.
57    fn store_confirmed_notifications(
58        &mut self,
59        notifications: &[BACnetAuditNotification],
60        apdu_timeout_ms: u32,
61        _receipt: CompletedAuditReceipt,
62    ) -> Result<ConfirmedAuditNotificationOutcome, Error> {
63        self.store_notifications(notifications, apdu_timeout_ms)?;
64        Ok(ConfirmedAuditNotificationOutcome::Stored)
65    }
66
67    /// Stage a batch whose commit the caller awaits without the database
68    /// guard (#1270): with `receipt` as
69    /// [`store_confirmed_notifications_with_change`](Self::store_confirmed_notifications_with_change)
70    /// stores it, without as
71    /// [`store_notifications_with_change`](Self::store_notifications_with_change)
72    /// does. Nothing changes in memory until
73    /// [`finish_notification_batch`](Self::finish_notification_batch) takes
74    /// the committed batch.
75    ///
76    /// The default stores the batch at once and returns
77    /// [`AuditBatchStage::Done`], so a sink that does not stage keeps its
78    /// commit where it was.
79    fn stage_notification_batch(
80        &mut self,
81        notifications: &[BACnetAuditNotification],
82        apdu_timeout_ms: u32,
83        receipt: Option<CompletedAuditReceipt>,
84    ) -> Result<AuditBatchStage, Error> {
85        match receipt {
86            Some(receipt) => self
87                .store_confirmed_notifications_with_change(notifications, apdu_timeout_ms, receipt)
88                .map(|(outcome, changed)| AuditBatchStage::Done(outcome, changed)),
89            None => self
90                .store_notifications_with_change(notifications, apdu_timeout_ms)
91                .map(|changed| {
92                    AuditBatchStage::Done(ConfirmedAuditNotificationOutcome::Stored, changed)
93                }),
94        }
95    }
96
97    /// Take a batch [`stage_notification_batch`](Self::stage_notification_batch)
98    /// staged, once its commit has run: the batch's outcome and whether its
99    /// retained records changed, or the commit's error with memory unchanged.
100    ///
101    /// The default fails with DEVICE / OPERATIONAL_PROBLEM, since the default
102    /// stage never stages.
103    fn finish_notification_batch(
104        &mut self,
105        staged: StagedAuditBatch,
106    ) -> Result<(ConfirmedAuditNotificationOutcome, bool), Error> {
107        let _ = staged;
108        Err(Error::Protocol {
109            class: ErrorClass::DEVICE.to_raw() as u32,
110            code: ErrorCode::OPERATIONAL_PROBLEM.to_raw() as u32,
111        })
112    }
113}
114
115impl AuditLogObject {
116    fn store_notification_batch(
117        &mut self,
118        notifications: &[BACnetAuditNotification],
119        apdu_timeout_ms: u32,
120    ) -> Result<bool, Error> {
121        if !self.log_enable {
122            return Err(Error::Protocol {
123                class: ErrorClass::SERVICES.to_raw() as u32,
124                code: ErrorCode::SERVICE_REQUEST_DENIED.to_raw() as u32,
125            });
126        }
127        if notifications.is_empty() {
128            return Err(Error::OutOfRange(
129                "Audit notification batch must not be empty".into(),
130            ));
131        }
132
133        let mut prospective = self.snapshot_for_next_generation()?;
134        let changed =
135            self.apply_notification_batch(&mut prospective, notifications, apdu_timeout_ms)?;
136        if changed {
137            self.commit_and_apply(prospective)?;
138        }
139        Ok(changed && self.buffer_size != 0)
140    }
141
142    fn store_confirmed_notification_batch(
143        &mut self,
144        notifications: &[BACnetAuditNotification],
145        apdu_timeout_ms: u32,
146        receipt: CompletedAuditReceipt,
147    ) -> Result<(ConfirmedAuditNotificationOutcome, bool), Error> {
148        if !self.log_enable {
149            return Err(Error::Protocol {
150                class: ErrorClass::SERVICES.to_raw() as u32,
151                code: ErrorCode::SERVICE_REQUEST_DENIED.to_raw() as u32,
152            });
153        }
154        if notifications.is_empty() {
155            return Err(Error::OutOfRange(
156                "Audit notification batch must not be empty".into(),
157            ));
158        }
159        if receipt::contains(
160            &self.completed_receipts,
161            receipt.key(),
162            receipt.completed_at_unix_millis(),
163        )? {
164            return Ok((ConfirmedAuditNotificationOutcome::Duplicate, false));
165        }
166
167        let mut prospective = self.snapshot_for_next_generation()?;
168        let changed =
169            self.apply_notification_batch(&mut prospective, notifications, apdu_timeout_ms)?;
170        let outcome = receipt::insert(&mut prospective.completed_receipts, receipt)?;
171        if outcome == ConfirmedAuditNotificationOutcome::Duplicate {
172            return Ok((outcome, false));
173        }
174        self.commit_and_apply(prospective)?;
175        Ok((outcome, changed && self.buffer_size != 0))
176    }
177
178    pub(super) fn apply_notification_batch(
179        &self,
180        prospective: &mut AuditLogSnapshot,
181        notifications: &[BACnetAuditNotification],
182        apdu_timeout_ms: u32,
183    ) -> Result<bool, Error> {
184        let mut changed = false;
185        for notification in notifications {
186            if let Some(index) = matching_record_index(
187                &prospective.records,
188                notification,
189                u64::from(apdu_timeout_ms) * 2,
190            ) {
191                let BACnetAuditLogDatum::AuditNotification(stored) =
192                    &mut prospective.records[index].record.datum
193                else {
194                    unreachable!("matching_record_index only returns Audit notifications");
195                };
196                if stored.source_timestamp.is_some() && stored.target_timestamp.is_some() {
197                    continue;
198                }
199                merge_notification(stored, notification);
200                validate_record(&prospective.records[index].record)?;
201                changed = true;
202            } else {
203                let timestamp = self.valid_timestamp()?;
204                let record = BACnetAuditLogRecord {
205                    timestamp,
206                    datum: BACnetAuditLogDatum::AuditNotification(notification.clone()),
207                };
208                validate_record(&record)?;
209                append_record(prospective, record);
210                changed = true;
211            }
212        }
213        Ok(changed)
214    }
215}
216
217impl AuditLogNotificationSink for AuditLogObject {
218    fn notification_logging_enabled(&self) -> bool {
219        self.log_enable
220    }
221
222    fn store_notifications(
223        &mut self,
224        notifications: &[BACnetAuditNotification],
225        apdu_timeout_ms: u32,
226    ) -> Result<(), Error> {
227        self.store_notification_batch(notifications, apdu_timeout_ms)
228            .map(|_| ())
229    }
230
231    fn store_notifications_with_change(
232        &mut self,
233        notifications: &[BACnetAuditNotification],
234        apdu_timeout_ms: u32,
235    ) -> Result<bool, Error> {
236        self.store_notification_batch(notifications, apdu_timeout_ms)
237    }
238
239    fn store_confirmed_notifications_with_change(
240        &mut self,
241        notifications: &[BACnetAuditNotification],
242        apdu_timeout_ms: u32,
243        receipt: CompletedAuditReceipt,
244    ) -> Result<(ConfirmedAuditNotificationOutcome, bool), Error> {
245        self.store_confirmed_notification_batch(notifications, apdu_timeout_ms, receipt)
246    }
247
248    fn has_completed_confirmed_receipt(
249        &self,
250        key: &[u8],
251        now_unix_millis: u64,
252    ) -> Result<bool, Error> {
253        receipt::contains(&self.completed_receipts, key, now_unix_millis)
254    }
255
256    fn store_confirmed_notifications(
257        &mut self,
258        notifications: &[BACnetAuditNotification],
259        apdu_timeout_ms: u32,
260        receipt: CompletedAuditReceipt,
261    ) -> Result<ConfirmedAuditNotificationOutcome, Error> {
262        self.store_confirmed_notification_batch(notifications, apdu_timeout_ms, receipt)
263            .map(|(outcome, _)| outcome)
264    }
265
266    fn stage_notification_batch(
267        &mut self,
268        notifications: &[BACnetAuditNotification],
269        apdu_timeout_ms: u32,
270        receipt: Option<CompletedAuditReceipt>,
271    ) -> Result<AuditBatchStage, Error> {
272        AuditLogObject::stage_notification_batch(self, notifications, apdu_timeout_ms, receipt)
273    }
274
275    fn finish_notification_batch(
276        &mut self,
277        staged: StagedAuditBatch,
278    ) -> Result<(ConfirmedAuditNotificationOutcome, bool), Error> {
279        AuditLogObject::finish_notification_batch(self, staged)
280    }
281}
282
283fn matching_record_index(
284    records: &[BACnetAuditLogRecordResult],
285    incoming: &BACnetAuditNotification,
286    window_ms: u64,
287) -> Option<usize> {
288    records.iter().rposition(|result| {
289        let BACnetAuditLogDatum::AuditNotification(stored) = &result.record.datum else {
290            return false;
291        };
292        notification_identity_matches(stored, incoming)
293            && complementary_timestamps_match(stored, incoming, window_ms)
294    })
295}
296
297fn notification_identity_matches(
298    stored: &BACnetAuditNotification,
299    incoming: &BACnetAuditNotification,
300) -> bool {
301    // Clause 12.64 names the exact equality coordinates. Its
302    // "operation-source" is the Clause 19.6 client actor represented by
303    // Source Device; Source Object is optional extra information. Likewise,
304    // Target Property is the BACnetPropertyReference parameter, distinct from
305    // the separately defined Target Object parameter. Source Object and Target
306    // Object are therefore merged when absent but are not match coordinates.
307    stored.source_device == incoming.source_device
308        && stored.operation == incoming.operation
309        && stored.invoke_id == incoming.invoke_id
310        && stored.target_device == incoming.target_device
311        && stored.target_property == incoming.target_property
312        && optional_agrees(&stored.source_user_id, &incoming.source_user_id)
313        && optional_agrees(&stored.source_user_role, &incoming.source_user_role)
314        && optional_agrees(&stored.target_value, &incoming.target_value)
315}
316
317fn optional_agrees<T: PartialEq>(left: &Option<T>, right: &Option<T>) -> bool {
318    match (left, right) {
319        (Some(left), Some(right)) => left == right,
320        _ => true,
321    }
322}
323
324fn complementary_timestamps_match(
325    stored: &BACnetAuditNotification,
326    incoming: &BACnetAuditNotification,
327    window_ms: u64,
328) -> bool {
329    match (
330        stored.source_timestamp.as_ref(),
331        incoming.target_timestamp.as_ref(),
332    ) {
333        (Some(left), Some(right)) if timestamps_within(left, right, window_ms) => true,
334        _ => matches!(
335            (
336                stored.target_timestamp.as_ref(),
337                incoming.source_timestamp.as_ref()
338            ),
339            (Some(left), Some(right)) if timestamps_within(left, right, window_ms)
340        ),
341    }
342}
343
344/// Compare like timestamp variants only. DateTime uses an absolute civil-time
345/// distance, Time uses the shortest circular time-of-day distance (so reports
346/// can straddle midnight), and SequenceNumber has no duration scale and must
347/// therefore be equal. Mixed variants do not match.
348fn timestamps_within(left: &BACnetTimeStamp, right: &BACnetTimeStamp, window_ms: u64) -> bool {
349    let window_centiseconds = window_ms / 10;
350    match (left, right) {
351        (BACnetTimeStamp::Time(left), BACnetTimeStamp::Time(right)) => {
352            let (Some(left), Some(right)) = (time_centiseconds(*left), time_centiseconds(*right))
353            else {
354                return false;
355            };
356            let distance = left.abs_diff(right);
357            distance.min(8_640_000 - distance) <= window_centiseconds
358        }
359        (
360            BACnetTimeStamp::DateTime {
361                date: left_date,
362                time: left_time,
363            },
364            BACnetTimeStamp::DateTime {
365                date: right_date,
366                time: right_time,
367            },
368        ) => {
369            let (Some(left), Some(right)) = (
370                datetime_centiseconds(*left_date, *left_time),
371                datetime_centiseconds(*right_date, *right_time),
372            ) else {
373                return false;
374            };
375            left.abs_diff(right) <= window_centiseconds
376        }
377        (BACnetTimeStamp::SequenceNumber(left), BACnetTimeStamp::SequenceNumber(right)) => {
378            left == right
379        }
380        _ => false,
381    }
382}
383
384fn time_centiseconds(time: Time) -> Option<u64> {
385    (time.hour <= 23 && time.minute <= 59 && time.second <= 59 && time.hundredths <= 99).then(
386        || {
387            (((u64::from(time.hour) * 60 + u64::from(time.minute)) * 60 + u64::from(time.second))
388                * 100)
389                + u64::from(time.hundredths)
390        },
391    )
392}
393
394fn datetime_centiseconds(date: Date, time: Time) -> Option<u64> {
395    let year = date.actual_year()?;
396    if !(1..=12).contains(&date.month) || !(1..=7).contains(&date.day_of_week) {
397        return None;
398    }
399    let max_day = days_in_month(year, date.month);
400    if date.day == 0 || date.day > max_day {
401        return None;
402    }
403    let days = days_from_civil(i64::from(year), i64::from(date.month), i64::from(date.day));
404    let expected_day_of_week = (days + 3).rem_euclid(7) as u8 + 1;
405    if date.day_of_week != expected_day_of_week {
406        return None;
407    }
408    let time = time_centiseconds(time)?;
409    let shifted_days = days.checked_add(719_468)?;
410    u64::try_from(shifted_days)
411        .ok()?
412        .checked_mul(8_640_000)?
413        .checked_add(time)
414}
415
416fn days_in_month(year: u16, month: u8) -> u8 {
417    match month {
418        1 | 3 | 5 | 7 | 8 | 10 | 12 => 31,
419        4 | 6 | 9 | 11 => 30,
420        2 if year.is_multiple_of(4) && (!year.is_multiple_of(100) || year.is_multiple_of(400)) => {
421            29
422        }
423        2 => 28,
424        _ => 0,
425    }
426}
427
428fn days_from_civil(year: i64, month: i64, day: i64) -> i64 {
429    let year = year - i64::from(month <= 2);
430    let era = year.div_euclid(400);
431    let year_of_era = year - era * 400;
432    let month_prime = month + if month > 2 { -3 } else { 9 };
433    let day_of_year = (153 * month_prime + 2) / 5 + day - 1;
434    let day_of_era = year_of_era * 365 + year_of_era / 4 - year_of_era / 100 + day_of_year;
435    era * 146_097 + day_of_era - 719_468
436}
437
438fn merge_notification(stored: &mut BACnetAuditNotification, incoming: &BACnetAuditNotification) {
439    let incoming_is_target_report = incoming.target_timestamp.is_some();
440    fill_missing(&mut stored.source_timestamp, &incoming.source_timestamp);
441    fill_missing(&mut stored.target_timestamp, &incoming.target_timestamp);
442    fill_missing(&mut stored.source_object, &incoming.source_object);
443    fill_missing(&mut stored.source_comment, &incoming.source_comment);
444    fill_missing(&mut stored.target_comment, &incoming.target_comment);
445    fill_missing(&mut stored.source_user_id, &incoming.source_user_id);
446    fill_missing(&mut stored.source_user_role, &incoming.source_user_role);
447    fill_missing(&mut stored.target_object, &incoming.target_object);
448    fill_missing(&mut stored.target_property, &incoming.target_property);
449    fill_missing(&mut stored.target_priority, &incoming.target_priority);
450    fill_missing(&mut stored.target_value, &incoming.target_value);
451    fill_missing(&mut stored.result, &incoming.result);
452    if incoming_is_target_report && incoming.current_value.is_some() {
453        stored.current_value.clone_from(&incoming.current_value);
454    } else {
455        fill_missing(&mut stored.current_value, &incoming.current_value);
456    }
457}
458
459fn fill_missing<T: Clone>(stored: &mut Option<T>, incoming: &Option<T>) {
460    if stored.is_none() {
461        stored.clone_from(incoming);
462    }
463}
bacnet_objects/audit/notification.rs

bacnet_objects/audit/
notification.rs
