bacnet_server/server/
time_sync_policy.rs

1//! Local opt-in inbound clock policy, independent of DCC and mutation policy.
2
3use super::dcc_policy::{address_length_fits, routed_entry_names_direct_source};
4use super::{BipServerBuilder, ServerBuilder, TransportPort, runtime_clock};
5use bacnet_encoding::npdu::NpduAddress;
6use bacnet_network::layer::ReceivedApdu;
7use bacnet_objects::clock::ClockFrame;
8use bacnet_types::constructed::BACnetAddress;
9use bacnet_types::error::Error;
10use std::collections::HashMap;
11use std::sync::{Arc, Mutex};
12use std::time::{Duration, Instant};
13
14/// Exact claimed time-sync source, not an authenticated principal.
15///
16/// Unconfirmed dispatch exposes transport MAC and NPDU source only. On SC the
17/// MAC is the source VMAC, not a certificate identity or proof of trust. Routed
18/// identity takes precedence; a trusted router MAC does not trust its clients.
19#[derive(Clone, Debug, Eq, PartialEq, Hash)]
20pub enum TimeSyncSource {
21    /// Complete transport-native MAC (1 to [`BACnetAddress::MAX_MAC_LEN`]
22    /// octets), only when no routed source is present.
23    Direct(Vec<u8>),
24    /// Complete claimed routed source, independent of the immediate router.
25    /// Once the server knows its own network's number, an entry on that
26    /// network also matches the station's direct requests from `address`
27    /// with no routed source (#1458).
28    Routed {
29        /// Source network (1..=65534).
30        network: u16,
31        /// Source address (1 to [`BACnetAddress::MAX_MAC_LEN`] octets).
32        address: Vec<u8>,
33    },
34}
35
36impl TimeSyncSource {
37    fn validate(&self) -> Result<(), Error> {
38        let address = match self {
39            Self::Direct(address) => address,
40            Self::Routed { network, address } => {
41                if !(1..=65534).contains(network) {
42                    return Err(denied("source network must be 1..=65534"));
43                }
44                address
45            }
46        };
47        if !address_length_fits(address.len()) {
48            return Err(denied(&format!(
49                "source address must contain 1..={} octets",
50                BACnetAddress::MAX_MAC_LEN
51            )));
52        }
53        Ok(())
54    }
55
56    /// The rate and coalescing key of a received request: one node, one
57    /// budget. A request relayed with this network's own number as SNET
58    /// comes from the station its SADR names, the direct source a request
59    /// from that MAC with no SNET has, as [`CanonicalPeer::from_source`]
60    /// reads it (#1458). A routed source with no SADR stays routed, and so
61    /// fails validation.
62    ///
63    /// [`CanonicalPeer::from_source`]: bacnet_endpoint_core::coordinator::CanonicalPeer::from_source
64    fn from_received(received: &ReceivedApdu, local_network: Option<u16>) -> Result<Self, Error> {
65        use bacnet_endpoint_core::coordinator::CanonicalPeer;
66        let source = match &received.source_network {
67            Some(source) if !source.mac_address.is_empty() => {
68                match CanonicalPeer::from_source(&received.source_mac, Some(source), local_network)
69                {
70                    CanonicalPeer::Direct(mac) => Self::Direct(mac.to_vec()),
71                    CanonicalPeer::Routed { network, address } => Self::Routed {
72                        network,
73                        address: address.to_vec(),
74                    },
75                }
76            }
77            Some(source) => Self::Routed {
78                network: source.network,
79                address: Vec::new(),
80            },
81            None => Self::Direct(received.source_mac.to_vec()),
82        };
83        source.validate()?;
84        Ok(source)
85    }
86}
87
88/// Validated exact-source allowlist. A configured empty list denies everyone.
89/// `None` in [`TimeSyncPolicy`] leaves sources unrestricted.
90#[derive(Clone, Eq, PartialEq)]
91pub struct TimeSyncSourceRestriction(Vec<TimeSyncSource>);
92
93impl std::fmt::Debug for TimeSyncSourceRestriction {
94    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
95        f.debug_struct("TimeSyncSourceRestriction")
96            .field("entries", &self.0.len())
97            .finish_non_exhaustive()
98    }
99}
100
101impl TimeSyncSourceRestriction {
102    /// Accept at most 256 entries, each with 1 to [`BACnetAddress::MAX_MAC_LEN`]
103    /// (18) address octets, the longest source the network layer delivers
104    /// (#1198), so a longer entry could never match (#1266). Routed networks
105    /// must be 1..=65534. These are local configuration limits, not
106    /// authentication guarantees.
107    pub fn new(sources: Vec<TimeSyncSource>) -> Result<Self, Error> {
108        if sources.len() > 256 {
109            return Err(denied("source restriction allows at most 256 entries"));
110        }
111        for source in &sources {
112            source.validate()?;
113        }
114        Ok(Self(sources))
115    }
116
117    /// Whether a request from link MAC `mac`, with `routed` as its SNET and
118    /// SADR when a router relayed it, comes from a listed source. Once
119    /// `local_network`, this network's own number, is known, a routed entry
120    /// on it also lists the station's direct requests, as the DCC
121    /// restriction's does (#1458).
122    pub(super) fn allows(
123        &self,
124        mac: &[u8],
125        routed: Option<&NpduAddress>,
126        local_network: Option<u16>,
127    ) -> bool {
128        // Do not use admission's malformed-routed fallback for authorization.
129        match routed {
130            Some(source) => self.0.iter().any(|entry| matches!(entry,
131                TimeSyncSource::Routed { network, address }
132                if *network == source.network && address.as_slice() == source.mac_address.as_slice())),
133            None => self.0.iter().any(|entry| match entry {
134                TimeSyncSource::Direct(address) => address.as_slice() == mac,
135                TimeSyncSource::Routed { network, address } => {
136                    routed_entry_names_direct_source(*network, address, mac, local_network)
137                }
138            }),
139        }
140    }
141}
142
143/// Opt-in token bucket for accepted local and UTC time-sync requests combined.
144#[derive(Clone, Copy, Debug, PartialEq)]
145pub struct TimeSyncRateLimit {
146    /// Positive finite refill rate in requests per second of monotonic time.
147    pub max_per_second: f64,
148    /// Positive initial and maximum token capacity.
149    pub burst_capacity: u32,
150}
151
152impl TimeSyncRateLimit {
153    fn validate(self) -> Result<(), Error> {
154        if !self.max_per_second.is_finite()
155            || self.max_per_second <= 0.0
156            || self.burst_capacity == 0
157        {
158            return Err(denied(
159                "rate must be positive and finite with a positive burst",
160            ));
161        }
162        Ok(())
163    }
164}
165
166/// Default-allow inbound TimeSynchronization / UTCTimeSynchronization policy.
167///
168/// Every restriction is opt-in and independent. No default step cap is imposed:
169/// a controller booting with an incorrect clock may need a large initial sync.
170/// Operators should configure trusted claimed sources (not authentication), and
171/// choose a cap only when their boot/recovery process can tolerate rejection.
172/// This is local policy, not a change to the unconfirmed wire behavior (16.7/16.8).
173/// Denials send no response and do not change the clock or invoke its observer.
174/// Decoding/validation precedes source and rate policy, then step-check and clock
175/// update are serialized. The observer runs afterwards outside the limiter lock;
176/// a concurrent accepted request may already have advanced the clock again.
177#[derive(Clone, Debug, PartialEq)]
178pub struct TimeSyncPolicy {
179    /// Accept inbound time sync by default. False disables clock touch and callback.
180    pub enabled: bool,
181    /// Optional exact-source allowlist; Some(empty) denies all, None allows all.
182    pub source_restriction: Option<TimeSyncSourceRestriction>,
183    /// Optional absolute step cap, inclusive at the boundary, for forward/backward
184    /// corrections. Zero allows only an exact match. No readable clock means a
185    /// configured cap fails closed; without a cap, boot synchronization still works.
186    pub max_step: Option<Duration>,
187    /// Optional per-source token bucket, shared across local and UTC requests.
188    pub per_source_rate: Option<TimeSyncRateLimit>,
189    /// Optional server-wide token bucket, including all sources and both services.
190    pub global_rate: Option<TimeSyncRateLimit>,
191    /// Minimum spacing between accepted requests from one source, regardless of
192    /// value or local/UTC form. Zero disables coalescing. Drops do not extend it.
193    pub coalesce_window: Duration,
194    /// Minimum spacing between accepted requests globally, across all sources
195    /// and both services. Zero disables global coalescing. Drops do not extend it.
196    pub global_coalesce_window: Duration,
197    /// Bound on rate/coalescing source state (1..=65536, default 256). A full
198    /// table rejects new sources until an entry is fully refilled and outside its
199    /// coalescing window; active entries are never evicted to reset their budget.
200    pub max_sources: usize,
201}
202
203impl Default for TimeSyncPolicy {
204    fn default() -> Self {
205        Self {
206            enabled: true,
207            source_restriction: None,
208            max_step: None,
209            per_source_rate: None,
210            global_rate: None,
211            coalesce_window: Duration::ZERO,
212            global_coalesce_window: Duration::ZERO,
213            max_sources: 256,
214        }
215    }
216}
217
218impl TimeSyncPolicy {
219    /// Validate before transport startup/dialing, even when acceptance is disabled.
220    pub fn validate(&self) -> Result<(), Error> {
221        for limit in [self.per_source_rate, self.global_rate]
222            .into_iter()
223            .flatten()
224        {
225            limit.validate()?;
226        }
227        if !(1..=65536).contains(&self.max_sources) {
228            return Err(denied("max_sources must be 1..=65536"));
229        }
230        Ok(())
231    }
232}
233
234impl<T: TransportPort + 'static> ServerBuilder<T> {
235    /// Configure opt-in inbound clock restrictions; default is unrestricted allow.
236    pub fn time_sync_policy(mut self, policy: TimeSyncPolicy) -> Self {
237        self.config.time_sync_policy = policy;
238        self
239    }
240}
241
242impl BipServerBuilder {
243    /// Configure opt-in inbound clock restrictions; see [`TimeSyncPolicy`].
244    pub fn time_sync_policy(mut self, policy: TimeSyncPolicy) -> Self {
245        self.config.time_sync_policy = policy;
246        self
247    }
248}
249
250#[cfg(feature = "sc-tls")]
251impl super::ScServerBuilder {
252    /// Configure inbound clock restrictions on claimed SC VMAC / routed sources.
253    /// Dispatch does not expose certificate principals; see [`TimeSyncSource`].
254    pub fn time_sync_policy(mut self, policy: TimeSyncPolicy) -> Self {
255        self.config.time_sync_policy = policy;
256        self
257    }
258}
259
260pub(super) fn denied(reason: &str) -> Error {
261    Error::Encoding(format!("time sync: {reason}"))
262}
263
264/// Compare in the request's time basis; no clock mutation or floating-point time.
265pub(super) fn step_hundredths(supplied: i128, is_utc: bool, frame: ClockFrame) -> Option<u128> {
266    let local = super::clock::date_time_to_hundredths(frame.local_date, frame.local_time).ok()?;
267    let current = if is_utc {
268        local + i128::from(frame.utc_offset) * 6000
269            - if frame.daylight_savings_status {
270                360_000
271            } else {
272                0
273            }
274    } else {
275        local
276    };
277    Some(supplied.abs_diff(current))
278}
279
280pub(super) fn check_step(delta: Option<u128>, cap: Option<Duration>) -> Result<(), Error> {
281    if let Some(cap) = cap {
282        let delta = delta.ok_or_else(|| denied("clock unreadable for step cap"))?;
283        if delta * 10_000_000 > cap.as_nanos() {
284            return Err(denied("step cap exceeded"));
285        }
286    }
287    Ok(())
288}
289
290struct Bucket {
291    tokens: f64,
292    last_refill: Instant,
293}
294
295impl Bucket {
296    fn new(limit: Option<TimeSyncRateLimit>, now: Instant) -> Self {
297        Self {
298            tokens: limit.map_or(0.0, |l| f64::from(l.burst_capacity)),
299            last_refill: now,
300        }
301    }
302
303    fn refill(&mut self, limit: Option<TimeSyncRateLimit>, now: Instant) {
304        if let Some(limit) = limit {
305            let elapsed = now
306                .saturating_duration_since(self.last_refill)
307                .as_secs_f64();
308            self.tokens =
309                (self.tokens + elapsed * limit.max_per_second).min(f64::from(limit.burst_capacity));
310        }
311        self.last_refill = self.last_refill.max(now);
312    }
313
314    fn ready(&self, limit: Option<TimeSyncRateLimit>) -> bool {
315        limit.is_none() || self.tokens >= 1.0
316    }
317
318    fn deduct(&mut self, limit: Option<TimeSyncRateLimit>) {
319        if limit.is_some() {
320            self.tokens -= 1.0;
321        }
322    }
323}
324
325struct SourceState {
326    bucket: Bucket,
327    last_accepted: Instant,
328}
329
330struct State {
331    global: Bucket,
332    last_accepted: Option<Instant>,
333    sources: HashMap<TimeSyncSource, SourceState>,
334}
335
336/// Server-lifetime synchronization admission state, separate from discovery.
337/// The mutex also serializes step-check + clock update across request tasks.
338pub(super) struct TimeSyncLimiter {
339    policy: TimeSyncPolicy,
340    state: Mutex<State>,
341}
342
343// Keep construction plumbing out of the size-capped lifecycle. Policies and
344// mutable state remain independent; both Arcs have the native server lifetime.
345pub(super) fn request_limiters(
346    config: &super::ServerConfig,
347    device_instance: Option<u32>,
348) -> (Arc<super::DiscoveryLimiter>, Arc<TimeSyncLimiter>) {
349    (
350        Arc::new(super::DiscoveryLimiter::new(
351            config.discovery_policy.sanitized(),
352            device_instance,
353        )),
354        Arc::new(TimeSyncLimiter::new(config.time_sync_policy.clone())),
355    )
356}
357
358impl TimeSyncLimiter {
359    pub(super) fn new(policy: TimeSyncPolicy) -> Self {
360        Self {
361            state: Mutex::new(State {
362                global: Bucket::new(policy.global_rate, runtime_clock::now()),
363                last_accepted: None,
364                sources: HashMap::new(),
365            }),
366            policy,
367        }
368    }
369
370    /// Admit and apply one request. `local_network`, this network's own
371    /// number once known, keys a station's direct and relayed forms alike
372    /// ([`TimeSyncSource::from_received`]). `now` is read from
373    /// [`runtime_clock::now`], the clock the limiter was built on (#1550).
374    pub(super) fn apply_at(
375        &self,
376        received: &ReceivedApdu,
377        local_network: Option<u16>,
378        now: Instant,
379        apply: impl FnOnce() -> Result<(), Error>,
380    ) -> Result<(), Error> {
381        // RB-07 compat mode: provenance threaded here for RB-09, no decision.
382        let _ = received.provenance;
383        let policy = &self.policy;
384        let track_source = policy.per_source_rate.is_some() || !policy.coalesce_window.is_zero();
385        let key = track_source
386            .then(|| TimeSyncSource::from_received(received, local_network))
387            .transpose()?;
388        let mut state = self
389            .state
390            .lock()
391            .map_err(|_| denied("limiter lock poisoned"))?;
392        if state
393            .last_accepted
394            .is_some_and(|last| now.saturating_duration_since(last) < policy.global_coalesce_window)
395        {
396            return Err(denied("coalesced globally"));
397        }
398        state.global.refill(policy.global_rate, now);
399        if !state.global.ready(policy.global_rate) {
400            return Err(denied("global rate limit"));
401        }
402        if let Some(key) = &key {
403            if let Some(source) = state.sources.get_mut(key) {
404                if now.saturating_duration_since(source.last_accepted) < policy.coalesce_window {
405                    return Err(denied("coalesced source"));
406                }
407                source.bucket.refill(policy.per_source_rate, now);
408                if !source.bucket.ready(policy.per_source_rate) {
409                    return Err(denied("source rate limit"));
410                }
411            } else if state.sources.len() >= policy.max_sources {
412                state.sources.retain(|_, source| {
413                    source.bucket.refill(policy.per_source_rate, now);
414                    let full = policy.per_source_rate.is_none_or(|limit| {
415                        source.bucket.tokens >= f64::from(limit.burst_capacity)
416                    });
417                    !full
418                        || now.saturating_duration_since(source.last_accepted)
419                            < policy.coalesce_window
420                });
421                if state.sources.len() >= policy.max_sources {
422                    return Err(denied("source table full"));
423                }
424            }
425        }
426        // No await or observer under this lock. Failed step/sync does not spend
427        // tokens or postpone the next legitimate synchronization.
428        apply()?;
429        state.global.deduct(policy.global_rate);
430        state.last_accepted = Some(now);
431        if let Some(key) = key {
432            let source = state.sources.entry(key).or_insert_with(|| SourceState {
433                bucket: Bucket::new(policy.per_source_rate, now),
434                last_accepted: now,
435            });
436            source.bucket.deduct(policy.per_source_rate);
437            source.last_accepted = now;
438        }
439        Ok(())
440    }
441}
442
443#[cfg(test)]
444#[path = "time_sync_policy_tests.rs"]
445mod tests;
bacnet_server/server/time_sync_policy.rs

bacnet_server/server/
time_sync_policy.rs
