bacnet_ops/bbmd/
mod.rs

1//! B/IP broadcast management (Annex J) as plain data: BBMD table rows, what
2//! reading a table gave, the breadth-first survey that follows each BDT to
3//! the BBMDs it lists, and the consistency matrix over the tables read
4//! ([`matrix`]). The TUI's Network screen and the MCP server's network map
5//! share it; [`Engine`](crate::engine::Engine) reads the tables.
6//!
7//! Only the read side of BVLL is used: Read-Broadcast-Distribution-Table
8//! and Read-Foreign-Device-Table (Annex J.2.3 and J.2.7). Nothing here
9//! writes a table or registers.
10
11pub mod matrix;
12
13use std::collections::{BTreeMap, HashSet};
14use std::future::Future;
15use std::net::{Ipv4Addr, SocketAddrV4};
16
17use bacnet_transport::bbmd::{BdtEntry, FdtEntryWire};
18use bacnet_transport::bip::BvlcExchangeError;
19use bacnet_transport::bvll::BvlcResultCode;
20use futures_util::stream::{FuturesUnordered, StreamExt};
21
22pub use self::matrix::{Cell, Finding, Matrix, Probe};
23
24/// The most BBMDs a survey reads. A larger set comes back with the rest
25/// unread ([`Outcome::Skipped`]).
26pub const MAX_SURVEY: usize = 16;
27
28/// The most BBMDs a survey reads at once. Each read takes one of the B/IP
29/// transport's eight exchange slots (one per peer and request), so this
30/// leaves room for the foreign-device registration and other reads, and
31/// silent peers wait out their timeouts together.
32pub const SURVEY_AT_ONCE: usize = 4;
33
34/// An FDT entry with less than this many seconds left is near expiry. The
35/// BBMD keeps an entry 30 s past its time-to-live (Annex J.5.2.3), so under
36/// this the device is already past its own time-to-live without having
37/// registered again.
38pub const FDT_EXPIRING_S: u16 = 30;
39
40/// One BDT entry: a BBMD and the mask its peers forward to it with.
41#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
42pub struct BdtRow {
43    /// The BBMD's B/IP address.
44    pub address: SocketAddrV4,
45    /// Its broadcast distribution mask.
46    pub mask: Ipv4Addr,
47}
48
49impl BdtRow {
50    /// True for an all-ones mask: peers send forwarded broadcasts to the
51    /// BBMD itself, which broadcasts them on its own subnet (Annex J.4.3.1,
52    /// the two-hop method).
53    pub fn is_unicast(&self) -> bool {
54        self.mask == Ipv4Addr::BROADCAST
55    }
56
57    /// Where a peer sends a forwarded broadcast for this entry: the address
58    /// with every bit the mask clears set (Annex J.4.5). With an all-ones
59    /// mask that is the BBMD itself.
60    pub fn forward_to(&self) -> Ipv4Addr {
61        Ipv4Addr::from(self.address.ip().to_bits() | !self.mask.to_bits())
62    }
63
64    /// One line on what the mask means for forwarding.
65    pub fn explain(&self) -> String {
66        if self.is_unicast() {
67            "peers unicast to it, and it rebroadcasts on its subnet (two-hop)".into()
68        } else {
69            format!(
70                "peers send a directed broadcast to {} (one-hop; routers must pass it)",
71                self.forward_to()
72            )
73        }
74    }
75}
76
77impl From<&BdtEntry> for BdtRow {
78    fn from(entry: &BdtEntry) -> Self {
79        Self {
80            address: SocketAddrV4::new(Ipv4Addr::from(entry.ip), entry.port),
81            mask: Ipv4Addr::from(entry.broadcast_mask),
82        }
83    }
84}
85
86/// One FDT entry.
87#[derive(Debug, Clone, Copy, PartialEq, Eq)]
88pub struct FdtRow {
89    /// The address the BBMD recorded for the foreign device, which is the
90    /// source address its registration arrived from.
91    pub address: SocketAddrV4,
92    /// The time-to-live the device asked for, in seconds.
93    pub ttl: u16,
94    /// Seconds left before the BBMD drops the entry, as of the read. The
95    /// BBMD adds a grace period to the time-to-live (Annex J.5.2.3).
96    pub remaining: u16,
97}
98
99impl FdtRow {
100    /// Whether the entry is near expiry: under [`FDT_EXPIRING_S`] left.
101    pub fn expiring(&self) -> bool {
102        self.remaining < FDT_EXPIRING_S
103    }
104}
105
106impl From<&FdtEntryWire> for FdtRow {
107    fn from(entry: &FdtEntryWire) -> Self {
108        Self {
109            address: SocketAddrV4::new(Ipv4Addr::from(entry.ip), entry.port),
110            ttl: entry.ttl,
111            remaining: entry.seconds_remaining,
112        }
113    }
114}
115
116/// The 6-octet B/IP MAC of `address`.
117pub fn mac(address: SocketAddrV4) -> [u8; 6] {
118    let [a, b, c, d] = address.ip().octets();
119    let [high, low] = address.port().to_be_bytes();
120    [a, b, c, d, high, low]
121}
122
123/// The address in a 6-octet B/IP MAC.
124pub fn address_of(mac: &[u8]) -> Option<SocketAddrV4> {
125    let [a, b, c, d, high, low] = <[u8; 6]>::try_from(mac).ok()?;
126    Some(SocketAddrV4::new(
127        Ipv4Addr::new(a, b, c, d),
128        u16::from_be_bytes([high, low]),
129    ))
130}
131
132/// What reading one of a BBMD's tables gave.
133#[derive(Debug, Clone, PartialEq, Eq)]
134pub enum Outcome<R> {
135    /// The table.
136    Table(R),
137    /// It answered with this NAK result: it doesn't carry out the read, as
138    /// a node that is no BBMD answers.
139    Nak(BvlcResultCode),
140    /// It did not answer, or the request could not be sent.
141    Unreachable(String),
142    /// Not read: the survey reached its limit, or the caller's budget ran
143    /// out first.
144    Skipped,
145}
146
147impl<R> Outcome<R> {
148    /// The table, when one was read.
149    pub fn table(&self) -> Option<&R> {
150        match self {
151            Self::Table(rows) => Some(rows),
152            _ => None,
153        }
154    }
155
156    /// Sort a failed table read into a refusal or no answer (#1822): a NAK
157    /// is a refusal, while anything else (no answer, a reply the read does
158    /// not take, a send failure, a malformed reply) means the BBMD could not
159    /// be read.
160    pub fn from_error(error: &BvlcExchangeError) -> Self {
161        match error {
162            BvlcExchangeError::Nak(code) => Self::Nak(*code),
163            other => Self::Unreachable(unreachable_reason(other)),
164        }
165    }
166}
167
168/// [`Outcome::from_error`] for a BDT read.
169pub fn probe_error(error: &BvlcExchangeError) -> Probe {
170    Probe::from_error(error)
171}
172
173/// [`unreachable_reason`] for a BBMD that didn't answer in time.
174pub const NO_ANSWER: &str = "no answer";
175
176/// Why a BBMD gave no table, as text: no answer in time, or what failed.
177pub fn unreachable_reason(error: &BvlcExchangeError) -> String {
178    match error {
179        BvlcExchangeError::NoReply(_) => NO_ANSWER.into(),
180        other => other.to_string(),
181    }
182}
183
184/// One BBMD a survey read: its BDT (or why there is none) and what the
185/// caller's read gave besides.
186#[derive(Debug, Clone, PartialEq, Eq)]
187pub struct Surveyed<X> {
188    /// The BBMD.
189    pub bbmd: SocketAddrV4,
190    /// Its BDT, or why there is none.
191    pub probe: Probe,
192    /// What else the read gave (the TUI's is nothing, the MCP server's the
193    /// FDT).
194    pub extra: X,
195}
196
197/// Why a BDT entry was not followed: its address is no BBMD a read may go
198/// to.
199#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
200pub enum NotFollowed {
201    /// All zeros: a BBMD reports an entry configured by a host name it
202    /// couldn't resolve this way (Annex J.4.4.1).
203    Unresolved,
204    /// An unspecified address or port 0.
205    Unspecified,
206    /// The limited broadcast address.
207    Broadcast,
208    /// A subnet's directed broadcast address: a BDT entry always holds a
209    /// BBMD's own unicast address, so one that is the local broadcast
210    /// address, or has every host bit of its own mask set, is none.
211    DirectedBroadcast,
212    /// A multicast address.
213    Multicast,
214    /// A loopback address, while this node isn't on loopback itself.
215    Loopback,
216    /// This node's own address.
217    Own,
218}
219
220impl NotFollowed {
221    /// The reason as a kebab-case word.
222    pub fn as_str(self) -> &'static str {
223        match self {
224            Self::Unresolved => "unresolved",
225            Self::Unspecified => "unspecified",
226            Self::Broadcast => "broadcast",
227            Self::DirectedBroadcast => "directed-broadcast",
228            Self::Multicast => "multicast",
229            Self::Loopback => "loopback",
230            Self::Own => "own-address",
231        }
232    }
233
234    /// Why `row`, a BDT entry, is not read by a node at `here`; `None` when
235    /// it may be. BDT entries are device data, so a hostile or broken table
236    /// must not turn the survey into a broadcast, a multicast or a send to
237    /// this host's other services.
238    pub fn check(row: &BdtRow, here: &Here) -> Option<Self> {
239        let directed = !row.is_unicast() && row.forward_to() == *row.address.ip();
240        Self::check_address(row.address, here).or(directed.then_some(Self::DirectedBroadcast))
241    }
242
243    /// [`check`](Self::check) for an address without a mask, such as a
244    /// survey's seed.
245    pub fn check_address(address: SocketAddrV4, here: &Here) -> Option<Self> {
246        let own = here.own;
247        let ip = address.ip();
248        if ip.is_unspecified() && address.port() == 0 {
249            Some(Self::Unresolved)
250        } else if ip.is_unspecified() || address.port() == 0 {
251            Some(Self::Unspecified)
252        } else if ip.is_broadcast() {
253            Some(Self::Broadcast)
254        } else if ip.is_multicast() {
255            Some(Self::Multicast)
256        } else if *ip == here.broadcast && here.broadcast != *own.ip() {
257            // A "broadcast" address that is our own (a loopback lab's
258            // setting) broadcasts nothing.
259            Some(Self::DirectedBroadcast)
260        } else if address == own {
261            Some(Self::Own)
262        } else if ip.is_loopback() && !own.ip().is_loopback() {
263            Some(Self::Loopback)
264        } else {
265            None
266        }
267    }
268}
269
270/// Where the surveying node is: its own address, and the broadcast address
271/// its transport sends local broadcasts to.
272#[derive(Debug, Clone, Copy, PartialEq, Eq)]
273pub struct Here {
274    /// This node's B/IP address.
275    pub own: SocketAddrV4,
276    /// Its transport's broadcast address.
277    pub broadcast: Ipv4Addr,
278}
279
280/// A BDT entry a survey did not follow, and the first BBMD that listed it.
281#[derive(Debug, Clone, Copy, PartialEq, Eq)]
282pub struct Unfollowed {
283    /// The address listed.
284    pub address: SocketAddrV4,
285    /// The BBMD whose table listed it first.
286    pub listed_by: SocketAddrV4,
287    /// Why it was not read.
288    pub why: NotFollowed,
289}
290
291/// What a survey found.
292#[derive(Debug, Clone, PartialEq, Eq)]
293pub struct Survey<X> {
294    /// The BBMDs read, at most [`MAX_SURVEY`], in breadth-first order.
295    pub read: Vec<Surveyed<X>>,
296    /// BBMDs the tables list past [`MAX_SURVEY`], in the order found, not
297    /// read.
298    pub unread: Vec<SocketAddrV4>,
299    /// Entries not followed, each address once.
300    pub not_followed: Vec<Unfollowed>,
301}
302
303/// Read the BDT of each seed, then of every BBMD a table read lists,
304/// breadth first, up to [`MAX_SURVEY`] reads; the rest are listed unread.
305/// `read` gives each BBMD's BDT and anything else the caller reads of it;
306/// `here` is where this node is, for [`NotFollowed::check`], which every
307/// listed entry passes before it is read (the seeds are the caller's).
308///
309/// Up to [`SURVEY_AT_ONCE`] reads run at once, so silent peers wait out
310/// their timeouts together. The result is the one reading them one at a
311/// time gives: BBMDs are numbered in the order they are found, a read
312/// starts only for one of the first [`MAX_SURVEY`], and a table's entries
313/// join the list only once every read before it has been taken in, so a
314/// fast answer from a later BBMD can't jump the breadth-first order.
315pub async fn survey<X, F, R>(seeds: &[SocketAddrV4], here: Here, read: F) -> Survey<X>
316where
317    F: Fn(SocketAddrV4) -> R,
318    R: Future<Output = (Probe, X)>,
319{
320    let read = &read;
321    let mut seen = HashSet::new();
322    // Every BBMD found, in breadth-first order.
323    let mut found: Vec<SocketAddrV4> = seeds.iter().copied().filter(|s| seen.insert(*s)).collect();
324    let mut done: Vec<Surveyed<X>> = Vec::new();
325    let mut not_followed = Vec::new();
326    // Reads finished ahead of an earlier one, by their place in `found`.
327    let mut early = BTreeMap::new();
328    let mut started = 0;
329    let mut reads = FuturesUnordered::new();
330    loop {
331        while reads.len() < SURVEY_AT_ONCE && started < found.len().min(MAX_SURVEY) {
332            let (index, bbmd) = (started, found[started]);
333            reads.push(async move { (index, read(bbmd).await) });
334            started += 1;
335        }
336        let Some((index, outcome)) = reads.next().await else {
337            break;
338        };
339        early.insert(index, outcome);
340        while let Some((probe, extra)) = early.remove(&done.len()) {
341            let bbmd = found[done.len()];
342            if let Probe::Table(rows) = &probe {
343                for row in rows {
344                    if !seen.insert(row.address) {
345                        continue;
346                    }
347                    match NotFollowed::check(row, &here) {
348                        Some(why) => not_followed.push(Unfollowed {
349                            address: row.address,
350                            listed_by: bbmd,
351                            why,
352                        }),
353                        None => found.push(row.address),
354                    }
355                }
356            }
357            done.push(Surveyed { bbmd, probe, extra });
358        }
359    }
360    Survey {
361        unread: found.split_off(done.len()),
362        read: done,
363        not_followed,
364    }
365}
366
367#[cfg(test)]
368mod tests {
369    use super::*;
370    use std::time::Duration;
371
372    fn at(last: u8) -> SocketAddrV4 {
373        SocketAddrV4::new(Ipv4Addr::new(10, 0, last, 1), 47808)
374    }
375
376    /// A node at none of the tests' addresses.
377    fn nowhere() -> Here {
378        Here {
379            own: at(250),
380            broadcast: Ipv4Addr::BROADCAST,
381        }
382    }
383
384    fn row(address: &str, mask: [u8; 4]) -> BdtRow {
385        BdtRow {
386            address: address.parse().unwrap(),
387            mask: Ipv4Addr::from(mask),
388        }
389    }
390
391    #[test]
392    fn a_mask_says_how_peers_forward() {
393        let unicast = row("10.0.1.1:47808", [255; 4]);
394        assert!(unicast.is_unicast());
395        assert_eq!(unicast.forward_to(), Ipv4Addr::new(10, 0, 1, 1));
396        assert!(unicast.explain().contains("two-hop"));
397        let directed = row("10.0.2.1:47808", [255, 255, 255, 0]);
398        assert!(!directed.is_unicast());
399        assert_eq!(directed.forward_to(), Ipv4Addr::new(10, 0, 2, 255));
400        assert!(directed.explain().contains("10.0.2.255"));
401    }
402
403    #[test]
404    fn an_entry_under_thirty_seconds_is_expiring() {
405        let entry = |remaining| FdtRow {
406            address: at(9),
407            ttl: 60,
408            remaining,
409        };
410        assert!(entry(29).expiring());
411        assert!(!entry(30).expiring());
412    }
413
414    #[test]
415    fn a_nak_and_a_timeout_sort_apart() {
416        use bacnet_transport::bvll::BvlcFunction;
417        let code = BvlcResultCode::READ_BROADCAST_DISTRIBUTION_TABLE_NAK;
418        assert_eq!(probe_error(&BvlcExchangeError::Nak(code)), Probe::Nak(code));
419        let silent = BvlcExchangeError::NoReply(Duration::from_secs(3));
420        assert_eq!(probe_error(&silent), Probe::Unreachable("no answer".into()));
421        // A success answers no read: not a refusal.
422        let success = BvlcExchangeError::UnexpectedResponse {
423            expected: BvlcFunction::READ_BROADCAST_DISTRIBUTION_TABLE_ACK,
424            got: BvlcFunction::BVLC_RESULT,
425        };
426        assert!(matches!(probe_error(&success), Probe::Unreachable(_)));
427        let local = BvlcExchangeError::Transport(bacnet_types::error::Error::Encoding(
428            "B/IP management client is stopped".into(),
429        ));
430        assert_eq!(
431            probe_error(&local),
432            Probe::Unreachable("encoding error: B/IP management client is stopped".into())
433        );
434        let mac = mac("10.0.0.1:47809".parse().unwrap());
435        assert_eq!(address_of(&mac), Some("10.0.0.1:47809".parse().unwrap()));
436        assert_eq!(address_of(&mac[..5]), None);
437    }
438
439    /// Two seeds are read first, in order, then the BBMDs their tables
440    /// list; a seed listed by the other is read once.
441    #[tokio::test]
442    async fn every_seed_is_read_then_what_the_tables_list() {
443        let tables = |bbmd: SocketAddrV4| match bbmd.ip().octets()[2] {
444            1 => vec![at(1), at(2), at(3)],
445            2 => vec![at(2), at(1)],
446            _ => vec![],
447        };
448        let surveyed = survey(&[at(1), at(2), at(1)], nowhere(), |bbmd| async move {
449            let rows = tables(bbmd)
450                .into_iter()
451                .map(|address| BdtRow {
452                    address,
453                    mask: Ipv4Addr::BROADCAST,
454                })
455                .collect();
456            (Probe::Table(rows), bbmd.port())
457        })
458        .await;
459        let order: Vec<SocketAddrV4> = surveyed.read.iter().map(|s| s.bbmd).collect();
460        assert_eq!(order, [at(1), at(2), at(3)]);
461        assert!(surveyed.read.iter().all(|s| s.extra == 47808));
462        assert!(surveyed.unread.is_empty() && surveyed.not_followed.is_empty());
463    }
464
465    /// Each class of address a BDT may list but no read may go to is
466    /// reported, once, with the BBMD that listed it, and never read; a
467    /// loopback peer is read only by a node on loopback.
468    #[tokio::test]
469    async fn addresses_no_read_may_go_to_are_not_followed() {
470        let here = Here {
471            own: at(1),
472            broadcast: Ipv4Addr::new(10, 0, 9, 255),
473        };
474        const ONES: [u8; 4] = [255; 4];
475        const SUBNET: [u8; 4] = [255, 255, 255, 0];
476        let bad = [
477            ("0.0.0.0:0", ONES, NotFollowed::Unresolved),
478            ("0.0.0.0:47808", ONES, NotFollowed::Unspecified),
479            ("10.0.5.1:0", ONES, NotFollowed::Unspecified),
480            ("255.255.255.255:47808", ONES, NotFollowed::Broadcast),
481            ("224.0.0.1:47808", ONES, NotFollowed::Multicast),
482            ("239.255.255.250:1900", ONES, NotFollowed::Multicast),
483            ("127.0.0.1:53", ONES, NotFollowed::Loopback),
484            // The local broadcast address, whatever its mask.
485            ("10.0.9.255:47808", ONES, NotFollowed::DirectedBroadcast),
486            // Every host bit of its own mask set.
487            ("10.0.7.255:47808", SUBNET, NotFollowed::DirectedBroadcast),
488        ];
489        let mut rows: Vec<BdtRow> = bad
490            .iter()
491            .map(|(address, mask, _)| BdtRow {
492                address: address.parse().unwrap(),
493                mask: Ipv4Addr::from(*mask),
494            })
495            .collect();
496        // Our own address, and one bad entry listed twice.
497        rows.push(BdtRow {
498            address: here.own,
499            mask: Ipv4Addr::BROADCAST,
500        });
501        rows.push(rows[0]);
502        let reads = std::sync::Mutex::new(Vec::new());
503        let surveyed = survey(&[at(2)], here, |bbmd| {
504            reads.lock().unwrap().push(bbmd);
505            let rows = rows.clone();
506            async move { (Probe::Table(rows), ()) }
507        })
508        .await;
509        assert_eq!(*reads.lock().unwrap(), [at(2)], "only the seed is read");
510        let mut expected: Vec<Unfollowed> = bad
511            .iter()
512            .map(|(address, _, why)| Unfollowed {
513                address: address.parse().unwrap(),
514                listed_by: at(2),
515                why: *why,
516            })
517            .collect();
518        expected.push(Unfollowed {
519            address: here.own,
520            listed_by: at(2),
521            why: NotFollowed::Own,
522        });
523        assert_eq!(surveyed.not_followed, expected);
524        // A unicast entry ending in .255, and a subnet mask on a BBMD's own
525        // address, are peers.
526        let entry = |address: &str, mask: [u8; 4]| BdtRow {
527            address: address.parse().unwrap(),
528            mask: Ipv4Addr::from(mask),
529        };
530        assert_eq!(
531            NotFollowed::check(&entry("10.0.7.255:47808", ONES), &here),
532            None
533        );
534        assert_eq!(
535            NotFollowed::check(&entry("10.0.7.1:47808", SUBNET), &here),
536            None
537        );
538        // On loopback, a loopback peer is a peer.
539        let local = Here {
540            own: "127.0.0.1:47808".parse().unwrap(),
541            broadcast: Ipv4Addr::new(127, 255, 255, 255),
542        };
543        let peer = entry("127.0.0.2:47809", ONES);
544        assert_eq!(NotFollowed::check(&peer, &local), None);
545        assert_eq!(
546            NotFollowed::check_address(local.own, &local),
547            Some(NotFollowed::Own)
548        );
549        assert_eq!(NotFollowed::check_address(at(3), &here), None);
550    }
551
552    /// A survey whose tables name far more BBMDs than it reads stops at
553    /// [`MAX_SURVEY`] reads and lists the rest unread.
554    #[tokio::test]
555    async fn a_large_fan_out_reads_at_most_the_cap() {
556        // Every table names 128 peers no other table names.
557        let next = std::sync::atomic::AtomicU32::new(0);
558        let peers = |_: SocketAddrV4| -> Vec<BdtRow> {
559            (0..128)
560                .map(|_| {
561                    let n = next.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
562                    BdtRow {
563                        address: SocketAddrV4::new(Ipv4Addr::from(0x0B00_0000 + n), 47808),
564                        mask: Ipv4Addr::BROADCAST,
565                    }
566                })
567                .collect()
568        };
569        let reads = std::sync::atomic::AtomicUsize::new(0);
570        let surveyed = survey(&[at(1)], nowhere(), |bbmd| {
571            reads.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
572            let rows = peers(bbmd);
573            async move { (Probe::Table(rows), ()) }
574        })
575        .await;
576        assert_eq!(reads.into_inner(), MAX_SURVEY);
577        assert_eq!(surveyed.read.len(), MAX_SURVEY);
578        assert!(surveyed.unread.len() > 1_000, "{}", surveyed.unread.len());
579        // The matrix over what was read stays bounded.
580        let probes: Vec<(SocketAddrV4, Probe)> = surveyed
581            .read
582            .iter()
583            .map(|s| (s.bbmd, s.probe.clone()))
584            .collect();
585        let matrix = Matrix::build(&probes);
586        assert!(matrix.bbmds.len() <= matrix::GRID_CAP);
587        assert!(matrix.omitted > 1_000, "{}", matrix.omitted);
588    }
589
590    /// No seed reads nothing.
591    #[tokio::test]
592    async fn no_seed_reads_nothing() {
593        let surveyed = survey(&[], nowhere(), |_| async { (Probe::Skipped, ()) }).await;
594        assert!(surveyed.read.is_empty() && surveyed.unread.is_empty());
595    }
596}
bacnet_ops/bbmd/mod.rs

bacnet_ops/bbmd/
mod.rs
