1use bacnet_encoding::npdu::{decode_npdu, encode_npdu, Npdu, NpduAddress};
9use bacnet_transport::port::{DataAttribute, TransportPort};
10use bacnet_types::enums::NetworkPriority;
11use bacnet_types::error::Error;
12use bacnet_types::MacAddr;
13use bytes::{Bytes, BytesMut};
14use tokio::sync::{mpsc, oneshot};
15use tokio::task::JoinHandle;
16use tracing::{debug, warn};
17
18pub struct ReceivedApdu {
20 pub apdu: Bytes,
22 pub source_mac: MacAddr,
24 pub source_network: Option<NpduAddress>,
26 pub data_attributes: Vec<DataAttribute>,
28 pub reply_tx: Option<oneshot::Sender<Bytes>>,
31}
32
33impl Clone for ReceivedApdu {
34 fn clone(&self) -> Self {
35 Self {
36 apdu: self.apdu.clone(),
37 source_mac: self.source_mac.clone(),
38 source_network: self.source_network.clone(),
39 data_attributes: self.data_attributes.clone(),
40 reply_tx: None,
41 }
42 }
43}
44
45impl std::fmt::Debug for ReceivedApdu {
46 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
47 f.debug_struct("ReceivedApdu")
48 .field("apdu", &self.apdu)
49 .field("source_mac", &self.source_mac)
50 .field("source_network", &self.source_network)
51 .field("data_attributes", &self.data_attributes)
52 .field("reply_tx", &self.reply_tx.as_ref().map(|_| "Some(...)"))
53 .finish()
54 }
55}
56
57pub struct NetworkLayer<T: TransportPort> {
64 transport: T,
65 dispatch_task: Option<JoinHandle<()>>,
66}
67
68impl<T: TransportPort + 'static> NetworkLayer<T> {
69 pub fn new(transport: T) -> Self {
71 Self {
72 transport,
73 dispatch_task: None,
74 }
75 }
76
77 pub async fn start(&mut self) -> Result<mpsc::Receiver<ReceivedApdu>, Error> {
82 let mut npdu_rx = self.transport.start().await?;
83
84 let (apdu_tx, apdu_rx) = mpsc::channel(256);
85
86 let dispatch_task = tokio::spawn(async move {
87 while let Some(received) = npdu_rx.recv().await {
88 match decode_npdu(received.npdu.clone()) {
89 Ok(npdu) => {
90 if npdu.is_network_message {
91 debug!(
92 message_type = npdu.message_type,
93 "Ignoring network layer message (non-router mode)"
94 );
95 continue;
96 }
97
98 if let Some(ref dest) = npdu.destination {
100 if dest.network != 0xFFFF {
101 debug!(
102 dnet = dest.network,
103 "Discarding routed message (non-router)"
104 );
105 continue;
106 }
107 }
108
109 let source_network = npdu.source.clone();
110
111 let apdu = ReceivedApdu {
112 apdu: npdu.payload,
113 source_mac: received.source_mac,
114 source_network,
115 data_attributes: received.data_attributes,
116 reply_tx: received.reply_tx,
117 };
118
119 if apdu_tx.send(apdu).await.is_err() {
120 break;
121 }
122 }
123 Err(e) => {
124 warn!(error = %e, "Failed to decode NPDU");
125 }
126 }
127 }
128 });
129
130 self.dispatch_task = Some(dispatch_task);
131
132 Ok(apdu_rx)
133 }
134
135 pub async fn send_apdu(
137 &self,
138 apdu: &[u8],
139 destination_mac: &[u8],
140 expecting_reply: bool,
141 priority: NetworkPriority,
142 ) -> Result<(), Error> {
143 self.send_apdu_with_data_attributes(apdu, destination_mac, expecting_reply, priority, &[])
144 .await
145 }
146
147 pub async fn send_apdu_with_data_attributes(
149 &self,
150 apdu: &[u8],
151 destination_mac: &[u8],
152 expecting_reply: bool,
153 priority: NetworkPriority,
154 data_attributes: &[DataAttribute],
155 ) -> Result<(), Error> {
156 let npdu = Npdu {
157 is_network_message: false,
158 expecting_reply,
159 priority,
160 destination: None,
161 source: None,
162 payload: Bytes::copy_from_slice(apdu),
163 ..Npdu::default()
164 };
165
166 let mut buf = BytesMut::with_capacity(2 + apdu.len());
167 encode_npdu(&mut buf, &npdu)?;
168
169 self.transport
170 .send_unicast_with_data_attributes(&buf, destination_mac, data_attributes)
171 .await
172 }
173
174 pub async fn broadcast_apdu(
176 &self,
177 apdu: &[u8],
178 expecting_reply: bool,
179 priority: NetworkPriority,
180 ) -> Result<(), Error> {
181 self.broadcast_apdu_with_data_attributes(apdu, expecting_reply, priority, &[])
182 .await
183 }
184
185 pub async fn broadcast_apdu_with_data_attributes(
187 &self,
188 apdu: &[u8],
189 expecting_reply: bool,
190 priority: NetworkPriority,
191 data_attributes: &[DataAttribute],
192 ) -> Result<(), Error> {
193 let npdu = Npdu {
194 is_network_message: false,
195 expecting_reply,
196 priority,
197 destination: None,
198 source: None,
199 payload: Bytes::copy_from_slice(apdu),
200 ..Npdu::default()
201 };
202
203 let mut buf = BytesMut::with_capacity(2 + apdu.len());
204 encode_npdu(&mut buf, &npdu)?;
205
206 self.transport
207 .send_broadcast_with_data_attributes(&buf, data_attributes)
208 .await
209 }
210
211 pub async fn broadcast_global_apdu(
216 &self,
217 apdu: &[u8],
218 expecting_reply: bool,
219 priority: NetworkPriority,
220 ) -> Result<(), Error> {
221 self.broadcast_global_apdu_with_data_attributes(apdu, expecting_reply, priority, &[])
222 .await
223 }
224
225 pub async fn broadcast_global_apdu_with_data_attributes(
227 &self,
228 apdu: &[u8],
229 expecting_reply: bool,
230 priority: NetworkPriority,
231 data_attributes: &[DataAttribute],
232 ) -> Result<(), Error> {
233 let npdu = Npdu {
234 is_network_message: false,
235 expecting_reply,
236 priority,
237 destination: Some(NpduAddress {
238 network: 0xFFFF,
239 mac_address: MacAddr::new(),
240 }),
241 source: None,
242 hop_count: 255,
243 payload: Bytes::copy_from_slice(apdu),
244 ..Npdu::default()
245 };
246
247 let mut buf = BytesMut::with_capacity(8 + apdu.len());
248 encode_npdu(&mut buf, &npdu)?;
249 self.transport
250 .send_broadcast_with_data_attributes(&buf, data_attributes)
251 .await
252 }
253
254 pub async fn broadcast_to_network(
259 &self,
260 apdu: &[u8],
261 dest_network: u16,
262 expecting_reply: bool,
263 priority: NetworkPriority,
264 ) -> Result<(), Error> {
265 self.broadcast_to_network_with_data_attributes(
266 apdu,
267 dest_network,
268 expecting_reply,
269 priority,
270 &[],
271 )
272 .await
273 }
274
275 pub async fn broadcast_to_network_with_data_attributes(
277 &self,
278 apdu: &[u8],
279 dest_network: u16,
280 expecting_reply: bool,
281 priority: NetworkPriority,
282 data_attributes: &[DataAttribute],
283 ) -> Result<(), Error> {
284 if dest_network == 0xFFFF {
285 return Err(Error::Encoding(
286 "dest_network 0xFFFF is reserved for global broadcasts; use broadcast_global_apdu instead".into(),
287 ));
288 }
289 let npdu = Npdu {
290 is_network_message: false,
291 expecting_reply,
292 priority,
293 destination: Some(NpduAddress {
294 network: dest_network,
295 mac_address: MacAddr::new(),
296 }),
297 source: None,
298 hop_count: 255,
299 payload: Bytes::copy_from_slice(apdu),
300 ..Npdu::default()
301 };
302
303 let mut buf = BytesMut::with_capacity(8 + apdu.len());
304 encode_npdu(&mut buf, &npdu)?;
305 self.transport
306 .send_broadcast_with_data_attributes(&buf, data_attributes)
307 .await
308 }
309
310 pub async fn send_apdu_routed(
316 &self,
317 apdu: &[u8],
318 dest_network: u16,
319 dest_mac: &[u8],
320 router_mac: &[u8],
321 expecting_reply: bool,
322 priority: NetworkPriority,
323 ) -> Result<(), Error> {
324 self.send_apdu_routed_with_data_attributes(
325 apdu,
326 dest_network,
327 dest_mac,
328 router_mac,
329 expecting_reply,
330 priority,
331 &[],
332 )
333 .await
334 }
335
336 pub async fn send_apdu_routed_with_data_attributes(
338 &self,
339 apdu: &[u8],
340 dest_network: u16,
341 dest_mac: &[u8],
342 router_mac: &[u8],
343 expecting_reply: bool,
344 priority: NetworkPriority,
345 data_attributes: &[DataAttribute],
346 ) -> Result<(), Error> {
347 let npdu = Npdu {
348 is_network_message: false,
349 expecting_reply,
350 priority,
351 destination: Some(NpduAddress {
352 network: dest_network,
353 mac_address: MacAddr::from_slice(dest_mac),
354 }),
355 source: None,
356 hop_count: 255,
357 payload: Bytes::copy_from_slice(apdu),
358 ..Npdu::default()
359 };
360
361 let mut buf = BytesMut::with_capacity(8 + dest_mac.len() + apdu.len());
362 encode_npdu(&mut buf, &npdu)?;
363
364 self.transport
365 .send_unicast_with_data_attributes(&buf, router_mac, data_attributes)
366 .await
367 }
368
369 pub fn transport(&self) -> &T {
374 &self.transport
375 }
376
377 pub fn local_mac(&self) -> &[u8] {
379 self.transport.local_mac()
380 }
381
382 pub async fn stop(&mut self) -> Result<(), Error> {
384 if let Some(task) = self.dispatch_task.take() {
385 task.abort();
386 let _ = task.await;
387 }
388 self.transport.stop().await
389 }
390}
391
392#[cfg(test)]
393mod tests {
394 use super::*;
395 use bacnet_transport::bip::BipTransport;
396 use bacnet_transport::sc::{LoopbackWebSocket, ScTransport, WebSocketPort};
397 use bacnet_transport::sc_frame::{
398 decode_sc_message, encode_sc_message, ScFunction, ScMessage, ScOption, Vmac,
399 };
400 use std::net::Ipv4Addr;
401 use tokio::time::{timeout, Duration};
402
403 async fn sc_hub_accept(ws_hub: &LoopbackWebSocket, hub_vmac: Vmac) {
404 let data = ws_hub.recv().await.unwrap();
405 let req = decode_sc_message(&data).unwrap();
406 assert_eq!(req.function, ScFunction::ConnectRequest);
407
408 let mut accept_payload = Vec::with_capacity(26);
409 accept_payload.extend_from_slice(&hub_vmac);
410 accept_payload.extend_from_slice(&[0u8; 16]);
411 accept_payload.extend_from_slice(&1476u16.to_be_bytes());
412 accept_payload.extend_from_slice(&1476u16.to_be_bytes());
413
414 let accept = ScMessage {
415 function: ScFunction::ConnectAccept,
416 message_id: req.message_id,
417 originating_vmac: None,
418 destination_vmac: None,
419 dest_options: Vec::new(),
420 data_options: Vec::new(),
421 payload: Bytes::from(accept_payload),
422 };
423 let mut buf = BytesMut::new();
424 encode_sc_message(&mut buf, &accept);
425 ws_hub.send(&buf).await.unwrap();
426 }
427
428 #[tokio::test]
429 async fn send_receive_apdu_unicast() {
430 let transport_a = BipTransport::new(Ipv4Addr::LOCALHOST, 0, Ipv4Addr::BROADCAST);
431 let transport_b = BipTransport::new(Ipv4Addr::LOCALHOST, 0, Ipv4Addr::BROADCAST);
432
433 let mut net_a = NetworkLayer::new(transport_a);
434 let mut net_b = NetworkLayer::new(transport_b);
435
436 let _rx_a = net_a.start().await.unwrap();
437 let mut rx_b = net_b.start().await.unwrap();
438
439 let test_apdu = vec![0x10, 0x08];
440
441 net_a
442 .send_apdu(
443 &test_apdu,
444 net_b.local_mac(),
445 false,
446 NetworkPriority::NORMAL,
447 )
448 .await
449 .unwrap();
450
451 let received = timeout(Duration::from_secs(2), rx_b.recv())
452 .await
453 .expect("Timed out waiting for APDU")
454 .expect("Channel closed");
455
456 assert_eq!(received.apdu, test_apdu);
457 assert_eq!(received.source_mac.as_slice(), net_a.local_mac());
458 assert!(received.source_network.is_none());
459
460 net_a.stop().await.unwrap();
461 net_b.stop().await.unwrap();
462 }
463
464 #[tokio::test]
465 async fn sc_data_options_reach_received_apdu_data_attributes() {
466 let (ws_client, ws_hub) = LoopbackWebSocket::pair();
467 let hub_vmac = [0x10; 6];
468 let mut net = NetworkLayer::new(ScTransport::new(ws_client, [0x01; 6]));
469
470 let hub_accept_task = tokio::spawn(async move {
471 sc_hub_accept(&ws_hub, hub_vmac).await;
472 ws_hub
473 });
474
475 let mut rx = net.start().await.unwrap();
476 let ws_hub = hub_accept_task.await.unwrap();
477
478 let apdu = Bytes::from_static(&[0x10, 0x08]);
479 let npdu = Npdu {
480 is_network_message: false,
481 expecting_reply: false,
482 priority: NetworkPriority::NORMAL,
483 destination: None,
484 source: None,
485 payload: apdu.clone(),
486 ..Npdu::default()
487 };
488 let mut npdu_buf = BytesMut::new();
489 encode_npdu(&mut npdu_buf, &npdu).unwrap();
490
491 let msg = ScMessage {
492 function: ScFunction::EncapsulatedNpdu,
493 message_id: 0x2345,
494 originating_vmac: Some(hub_vmac),
495 destination_vmac: None,
496 dest_options: Vec::new(),
497 data_options: vec![
498 ScOption {
499 option_type: 1,
500 must_understand: true,
501 data: Vec::new(),
502 },
503 ScOption {
504 option_type: 31,
505 must_understand: false,
506 data: vec![0x12, 0x34, 0x56],
507 },
508 ],
509 payload: npdu_buf.freeze(),
510 };
511 let mut sc_buf = BytesMut::new();
512 encode_sc_message(&mut sc_buf, &msg);
513 ws_hub.send(&sc_buf).await.unwrap();
514
515 let received = timeout(Duration::from_secs(1), rx.recv())
516 .await
517 .expect("timed out waiting for APDU")
518 .expect("APDU channel closed");
519
520 assert_eq!(received.apdu, apdu);
521 assert_eq!(received.source_mac.as_slice(), hub_vmac);
522 assert!(received.source_network.is_none());
523 assert_eq!(received.data_attributes.len(), 2);
524 assert_eq!(received.data_attributes[0].option_type, 1);
525 assert!(received.data_attributes[0].must_understand);
526 assert!(received.data_attributes[0].data.is_empty());
527 assert_eq!(received.data_attributes[1].option_type, 31);
528 assert!(!received.data_attributes[1].must_understand);
529 assert_eq!(received.data_attributes[1].data, vec![0x12, 0x34, 0x56]);
530
531 net.stop().await.unwrap();
532 }
533
534 #[tokio::test]
535 async fn send_apdu_data_attributes_reach_sc_data_options() {
536 let (ws_client, ws_hub) = LoopbackWebSocket::pair();
537 let hub_vmac = [0x10; 6];
538 let dest_vmac: Vmac = [0x02, 0x03, 0x04, 0x05, 0x06, 0x07];
539 let mut net = NetworkLayer::new(ScTransport::new(ws_client, [0x01; 6]));
540 let data_attributes = vec![
541 DataAttribute {
542 option_type: 1,
543 must_understand: true,
544 data: Vec::new(),
545 },
546 DataAttribute {
547 option_type: 31,
548 must_understand: false,
549 data: vec![0x12, 0x34, 0x56],
550 },
551 ];
552
553 let hub_accept_task = tokio::spawn(async move {
554 sc_hub_accept(&ws_hub, hub_vmac).await;
555 ws_hub
556 });
557
558 let _rx = net.start().await.unwrap();
559 let ws_hub = hub_accept_task.await.unwrap();
560
561 let apdu = Bytes::from_static(&[0x10, 0x08]);
562 net.send_apdu_with_data_attributes(
563 &apdu,
564 &dest_vmac,
565 false,
566 NetworkPriority::NORMAL,
567 &data_attributes,
568 )
569 .await
570 .unwrap();
571
572 let data = ws_hub.recv().await.unwrap();
573 let msg = decode_sc_message(&data).unwrap();
574 assert_eq!(msg.function, ScFunction::EncapsulatedNpdu);
575 assert_eq!(msg.destination_vmac, Some(dest_vmac));
576 assert_eq!(msg.data_options.len(), 2);
577 assert_eq!(msg.data_options[0].option_type, 1);
578 assert!(msg.data_options[0].must_understand);
579 assert_eq!(msg.data_options[1].option_type, 31);
580 assert!(!msg.data_options[1].must_understand);
581 assert_eq!(msg.data_options[1].data, vec![0x12, 0x34, 0x56]);
582
583 let npdu = decode_npdu(msg.payload).unwrap();
584 assert_eq!(npdu.payload, apdu);
585 assert!(!npdu.expecting_reply);
586 assert_eq!(npdu.priority, NetworkPriority::NORMAL);
587
588 net.stop().await.unwrap();
589 }
590
591 #[tokio::test]
592 async fn end_to_end_who_is() {
593 use bacnet_encoding::apdu::{decode_apdu, encode_apdu, Apdu, UnconfirmedRequest};
594 use bacnet_types::enums::UnconfirmedServiceChoice;
595
596 let transport_a = BipTransport::new(Ipv4Addr::LOCALHOST, 0, Ipv4Addr::BROADCAST);
597 let transport_b = BipTransport::new(Ipv4Addr::LOCALHOST, 0, Ipv4Addr::BROADCAST);
598
599 let mut net_a = NetworkLayer::new(transport_a);
600 let mut net_b = NetworkLayer::new(transport_b);
601
602 let _rx_a = net_a.start().await.unwrap();
603 let mut rx_b = net_b.start().await.unwrap();
604
605 let who_is_apdu = Apdu::UnconfirmedRequest(UnconfirmedRequest {
606 service_choice: UnconfirmedServiceChoice::WHO_IS,
607 service_request: Bytes::new(),
608 });
609 let mut apdu_buf = BytesMut::new();
610 encode_apdu(&mut apdu_buf, &who_is_apdu).expect("valid APDU encoding");
611
612 net_a
613 .send_apdu(&apdu_buf, net_b.local_mac(), false, NetworkPriority::NORMAL)
614 .await
615 .unwrap();
616
617 let received = timeout(Duration::from_secs(2), rx_b.recv())
618 .await
619 .expect("Timed out waiting for APDU")
620 .expect("Channel closed");
621
622 let decoded_apdu = decode_apdu(received.apdu.clone()).unwrap();
623 match decoded_apdu {
624 Apdu::UnconfirmedRequest(req) => {
625 assert_eq!(req.service_choice, UnconfirmedServiceChoice::WHO_IS);
626 assert!(req.service_request.is_empty());
627 }
628 other => panic!("Expected UnconfirmedRequest, got {:?}", other),
629 }
630
631 net_a.stop().await.unwrap();
632 net_b.stop().await.unwrap();
633 }
634
635 #[test]
636 fn global_broadcast_npdu_has_dnet_ffff() {
637 use bacnet_encoding::npdu::{decode_npdu, encode_npdu, Npdu, NpduAddress};
638 use bacnet_types::enums::NetworkPriority;
639
640 let npdu = Npdu {
641 is_network_message: false,
642 expecting_reply: false,
643 priority: NetworkPriority::NORMAL,
644 destination: Some(NpduAddress {
645 network: 0xFFFF,
646 mac_address: MacAddr::new(),
647 }),
648 source: None,
649 hop_count: 255,
650 payload: Bytes::from_static(&[0xAA]),
651 ..Npdu::default()
652 };
653
654 let mut buf = bytes::BytesMut::new();
655 encode_npdu(&mut buf, &npdu).unwrap();
656 let decoded = decode_npdu(Bytes::from(buf)).unwrap();
657 let dest = decoded.destination.unwrap();
658 assert_eq!(dest.network, 0xFFFF);
659 assert!(dest.mac_address.is_empty());
660 assert_eq!(decoded.hop_count, 255);
661 }
662
663 #[test]
664 fn transport_accessor() {
665 let transport = BipTransport::new(Ipv4Addr::LOCALHOST, 0, Ipv4Addr::BROADCAST);
666 let net = NetworkLayer::new(transport);
667 let mac = net.transport().local_mac();
668 assert_eq!(mac.len(), 6);
669 }
670
671 #[test]
672 fn routed_send_encodes_dnet_dadr() {
673 use bacnet_encoding::npdu::{decode_npdu, encode_npdu, Npdu, NpduAddress};
674 use bacnet_types::enums::NetworkPriority;
675
676 let npdu = Npdu {
677 is_network_message: false,
678 expecting_reply: true,
679 priority: NetworkPriority::NORMAL,
680 destination: Some(NpduAddress {
681 network: 100,
682 mac_address: MacAddr::from_slice(&[1, 2, 3, 4, 5, 6]),
683 }),
684 source: None,
685 hop_count: 255,
686 payload: Bytes::from_static(&[0xAA, 0xBB]),
687 ..Npdu::default()
688 };
689
690 let mut buf = bytes::BytesMut::new();
691 encode_npdu(&mut buf, &npdu).unwrap();
692 let decoded = decode_npdu(Bytes::from(buf)).unwrap();
693 let dest = decoded.destination.unwrap();
694 assert_eq!(dest.network, 100);
695 assert_eq!(dest.mac_address.as_slice(), &[1, 2, 3, 4, 5, 6]);
696 assert_eq!(decoded.hop_count, 255);
697 assert!(decoded.expecting_reply);
698 }
699
700 #[test]
701 fn broadcast_to_network_encodes_specific_dnet() {
702 use bacnet_encoding::npdu::{decode_npdu, encode_npdu, Npdu, NpduAddress};
703 use bacnet_types::enums::NetworkPriority;
704
705 let npdu = Npdu {
706 is_network_message: false,
707 expecting_reply: false,
708 priority: NetworkPriority::NORMAL,
709 destination: Some(NpduAddress {
710 network: 42,
711 mac_address: MacAddr::new(),
712 }),
713 source: None,
714 hop_count: 255,
715 payload: Bytes::from_static(&[0xCC]),
716 ..Npdu::default()
717 };
718
719 let mut buf = bytes::BytesMut::new();
720 encode_npdu(&mut buf, &npdu).unwrap();
721 let decoded = decode_npdu(Bytes::from(buf)).unwrap();
722 let dest = decoded.destination.unwrap();
723 assert_eq!(dest.network, 42);
724 assert!(dest.mac_address.is_empty());
725 assert_eq!(decoded.hop_count, 255);
726 assert!(!decoded.expecting_reply);
727 }
728
729 #[test]
730 fn broadcast_to_network_rejects_dnet_ffff() {
731 use bacnet_types::enums::NetworkPriority;
732
733 let transport = BipTransport::new(Ipv4Addr::LOCALHOST, 0, Ipv4Addr::BROADCAST);
734 let net = NetworkLayer::new(transport);
735
736 let rt = tokio::runtime::Builder::new_current_thread()
737 .enable_all()
738 .build()
739 .unwrap();
740 let result = rt.block_on(async {
741 net.broadcast_to_network(&[0xAA], 0xFFFF, false, NetworkPriority::NORMAL)
742 .await
743 });
744 assert!(result.is_err());
745 let err_msg = format!("{}", result.unwrap_err());
746 assert!(
747 err_msg.contains("0xFFFF"),
748 "Error should mention 0xFFFF: {err_msg}"
749 );
750 }
751}