use crate::network_number::LocalNetworkNumber;
use bacnet_encoding::npdu::{encode_npdu, Npdu, NpduAddress};
use bacnet_transport::port::{DataAttribute, TransportPort, TransportProvenance};
use bacnet_types::enums::{NetworkPriority, PduType};
use bacnet_types::error::Error;
use bacnet_types::MacAddr;
use bytes::{Bytes, BytesMut};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use tokio::sync::oneshot;
use tokio::task::JoinHandle;
#[path = "layer_issuance.rs"]
mod issuance;
pub use issuance::IssuedApdu;
#[path = "layer_admission.rs"]
mod admission;
pub(crate) use admission::{
broadcast_carries_unconfirmed, destination_is_coherent, link_source_fits, AdmissionSender,
};
pub use admission::{AdmissionReceiver, QueueAdmissionCounters, QueueAdmissionSnapshot};
pub struct ReceivedApdu {
pub apdu: Bytes,
pub source_mac: MacAddr,
pub ingress_network: Option<u16>,
pub source_network: Option<NpduAddress>,
pub link_layer_group: bool,
pub is_group: bool,
pub global_broadcast: bool,
pub data_attributes: Vec<DataAttribute>,
pub provenance: TransportProvenance,
pub direct_response: Option<bacnet_transport::port::DirectResponse>,
pub reply_tx: Option<oneshot::Sender<Bytes>>,
}
#[derive(Clone)]
pub struct ReceivedNetworkControl {
pub npdu: Npdu,
pub source_mac: MacAddr,
pub link_layer_group: bool,
pub data_attributes: Vec<DataAttribute>,
pub provenance: TransportProvenance,
pub ingress_sequence: u64,
}
impl std::fmt::Debug for ReceivedNetworkControl {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ReceivedNetworkControl")
.field("npdu", &self.npdu)
.field("source_mac_len", &self.source_mac.len())
.field("link_layer_group", &self.link_layer_group)
.field("data_attributes", &self.data_attributes)
.field("provenance", &self.provenance)
.field("ingress_sequence", &self.ingress_sequence)
.finish_non_exhaustive()
}
}
impl Clone for ReceivedApdu {
fn clone(&self) -> Self {
Self {
apdu: self.apdu.clone(),
source_mac: self.source_mac.clone(),
ingress_network: self.ingress_network,
source_network: self.source_network.clone(),
link_layer_group: self.link_layer_group,
is_group: self.is_group,
global_broadcast: self.global_broadcast,
data_attributes: self.data_attributes.clone(),
provenance: self.provenance,
direct_response: self.direct_response.clone(),
reply_tx: None,
}
}
}
impl std::fmt::Debug for ReceivedApdu {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let source_network_len = self
.source_network
.as_ref()
.map(|source| (source.network, source.mac_address.len()));
f.debug_struct("ReceivedApdu")
.field("apdu_len", &self.apdu.len())
.field("source_mac_len", &self.source_mac.len())
.field("ingress_network", &self.ingress_network)
.field("source_network", &source_network_len)
.field("link_layer_group", &self.link_layer_group)
.field("is_group", &self.is_group)
.field("global_broadcast", &self.global_broadcast)
.field("data_attributes", &self.data_attributes)
.field("provenance", &self.provenance)
.field("reply_tx", &self.reply_tx.as_ref().map(|_| "Some(...)"))
.finish_non_exhaustive()
}
}
pub(crate) fn is_global_broadcast(destination: Option<&NpduAddress>) -> bool {
destination.is_some_and(|destination| destination.network == 0xFFFF)
}
pub(crate) fn check_destination(network: u16, local_form: &str) -> Result<(), Error> {
match network {
0 => Err(Error::Encoding(format!(
"dest_network 0 is not a network number; {local_form}"
))),
0xFFFF => Err(Error::Encoding(
"dest_network 0xFFFF is the global broadcast; \
use broadcast_global_apdu for a global broadcast"
.into(),
)),
_ => Ok(()),
}
}
pub(crate) fn check_broadcast_apdu(apdu: &[u8], what: &str) -> Result<(), Error> {
let pdu_type = match apdu.first() {
Some(&first) if PduType::from_raw(first >> 4) == PduType::UNCONFIRMED_REQUEST => {
return Ok(())
}
Some(&first) => format!("PDU type {}", PduType::from_raw(first >> 4)),
None => "an empty APDU".to_string(),
};
Err(Error::Encoding(format!(
"{what} carries only an UNCONFIRMED_REQUEST APDU (Clause 6.3), not {pdu_type}; \
send anything else to one device"
)))
}
pub(crate) fn is_group_delivery(link_layer_group: bool, destination: Option<&NpduAddress>) -> bool {
match destination {
None => link_layer_group,
Some(destination) => destination.network == 0xFFFF || destination.mac_address.is_empty(),
}
}
#[derive(Debug, Clone, Copy)]
pub struct RoutedTarget<'a> {
pub network: u16,
pub mac: &'a [u8],
pub router_mac: &'a [u8],
}
pub struct NetworkLayer<T: TransportPort> {
transport: T,
response_scope: bacnet_transport::port::DirectResponseScope,
dispatch_task: Option<JoinHandle<()>>,
network_control_tx: Option<AdmissionSender<ReceivedNetworkControl>>,
network_control_ingress_sequence: Arc<AtomicU64>,
address_length_drops: Arc<AtomicU64>,
global_broadcast_dadr_drops: Arc<AtomicU64>,
broadcast_pdu_type_drops: Arc<AtomicU64>,
local_network_number: LocalNetworkNumber,
}
impl<T: TransportPort + 'static> NetworkLayer<T> {
pub fn new(transport: T) -> Self {
Self {
transport,
response_scope: Default::default(),
dispatch_task: None,
network_control_tx: None,
network_control_ingress_sequence: Arc::new(AtomicU64::new(0)),
address_length_drops: Arc::new(AtomicU64::new(0)),
global_broadcast_dadr_drops: Arc::new(AtomicU64::new(0)),
broadcast_pdu_type_drops: Arc::new(AtomicU64::new(0)),
local_network_number: LocalNetworkNumber::default(),
}
}
pub async fn send_apdu(
&self,
apdu: &[u8],
destination_mac: &[u8],
expecting_reply: bool,
priority: NetworkPriority,
) -> Result<(), Error> {
self.send_apdu_with_data_attributes(apdu, destination_mac, expecting_reply, priority, &[])
.await
}
pub async fn send_apdu_with_data_attributes(
&self,
apdu: &[u8],
destination_mac: &[u8],
expecting_reply: bool,
priority: NetworkPriority,
data_attributes: &[DataAttribute],
) -> Result<(), Error> {
let npdu = Npdu {
is_network_message: false,
expecting_reply,
priority,
destination: None,
source: None,
payload: Bytes::copy_from_slice(apdu),
..Npdu::default()
};
let mut buf = BytesMut::with_capacity(2 + apdu.len());
encode_npdu(&mut buf, &npdu)?;
self.transport
.send_unicast_with_data_attributes(&buf, destination_mac, data_attributes)
.await
}
pub async fn broadcast_apdu(
&self,
apdu: &[u8],
expecting_reply: bool,
priority: NetworkPriority,
) -> Result<(), Error> {
self.broadcast_apdu_with_data_attributes(apdu, expecting_reply, priority, &[])
.await
}
pub async fn broadcast_apdu_with_data_attributes(
&self,
apdu: &[u8],
expecting_reply: bool,
priority: NetworkPriority,
data_attributes: &[DataAttribute],
) -> Result<(), Error> {
check_broadcast_apdu(apdu, "a local broadcast")?;
let npdu = Npdu {
is_network_message: false,
expecting_reply,
priority,
destination: None,
source: None,
payload: Bytes::copy_from_slice(apdu),
..Npdu::default()
};
let mut buf = BytesMut::with_capacity(2 + apdu.len());
encode_npdu(&mut buf, &npdu)?;
self.transport
.send_broadcast_with_data_attributes(&buf, data_attributes)
.await
}
pub async fn broadcast_global_apdu(
&self,
apdu: &[u8],
expecting_reply: bool,
priority: NetworkPriority,
) -> Result<(), Error> {
self.broadcast_global_apdu_with_data_attributes(apdu, expecting_reply, priority, &[])
.await
}
pub async fn broadcast_global_apdu_with_data_attributes(
&self,
apdu: &[u8],
expecting_reply: bool,
priority: NetworkPriority,
data_attributes: &[DataAttribute],
) -> Result<(), Error> {
check_broadcast_apdu(apdu, "a global broadcast")?;
let npdu = Npdu {
is_network_message: false,
expecting_reply,
priority,
destination: Some(NpduAddress {
network: 0xFFFF,
mac_address: MacAddr::new(),
}),
source: None,
hop_count: 255,
payload: Bytes::copy_from_slice(apdu),
..Npdu::default()
};
let mut buf = BytesMut::with_capacity(8 + apdu.len());
encode_npdu(&mut buf, &npdu)?;
self.transport
.send_broadcast_with_data_attributes(&buf, data_attributes)
.await
}
pub async fn broadcast_to_network(
&self,
apdu: &[u8],
dest_network: u16,
expecting_reply: bool,
priority: NetworkPriority,
) -> Result<(), Error> {
self.broadcast_to_network_with_data_attributes(
apdu,
dest_network,
expecting_reply,
priority,
&[],
)
.await
}
pub async fn broadcast_to_network_with_data_attributes(
&self,
apdu: &[u8],
dest_network: u16,
expecting_reply: bool,
priority: NetworkPriority,
data_attributes: &[DataAttribute],
) -> Result<(), Error> {
check_destination(dest_network, "use broadcast_apdu for the local network")?;
check_broadcast_apdu(apdu, "a broadcast to one network")?;
let npdu = Npdu {
is_network_message: false,
expecting_reply,
priority,
destination: Some(NpduAddress {
network: dest_network,
mac_address: MacAddr::new(),
}),
source: None,
hop_count: 255,
payload: Bytes::copy_from_slice(apdu),
..Npdu::default()
};
let mut buf = BytesMut::with_capacity(8 + apdu.len());
encode_npdu(&mut buf, &npdu)?;
self.transport
.send_broadcast_with_data_attributes(&buf, data_attributes)
.await
}
pub async fn send_apdu_routed(
&self,
apdu: &[u8],
dest_network: u16,
dest_mac: &[u8],
router_mac: &[u8],
expecting_reply: bool,
priority: NetworkPriority,
) -> Result<(), Error> {
self.send_apdu_routed_with_data_attributes(
apdu,
RoutedTarget {
network: dest_network,
mac: dest_mac,
router_mac,
},
expecting_reply,
priority,
&[],
)
.await
}
pub async fn send_apdu_routed_with_data_attributes(
&self,
apdu: &[u8],
target: RoutedTarget<'_>,
expecting_reply: bool,
priority: NetworkPriority,
data_attributes: &[DataAttribute],
) -> Result<(), Error> {
check_destination(target.network, "use send_apdu for a local device")?;
if target.mac.is_empty() {
check_broadcast_apdu(
apdu,
"a routed send with an empty dest_mac (a remote broadcast)",
)?;
}
let buf = Self::encode_routed_npdu_buf(
apdu,
target.network,
target.mac,
expecting_reply,
priority,
)?;
self.transport
.send_unicast_with_data_attributes(&buf, target.router_mac, data_attributes)
.await
}
pub async fn send_apdu_routed_via_local_broadcast(
&self,
apdu: &[u8],
dest_network: u16,
dest_mac: &[u8],
expecting_reply: bool,
priority: NetworkPriority,
) -> Result<(), Error> {
self.send_apdu_routed_via_local_broadcast_with_data_attributes(
apdu,
dest_network,
dest_mac,
expecting_reply,
priority,
&[],
)
.await
}
pub async fn send_apdu_routed_via_local_broadcast_with_data_attributes(
&self,
apdu: &[u8],
dest_network: u16,
dest_mac: &[u8],
expecting_reply: bool,
priority: NetworkPriority,
data_attributes: &[DataAttribute],
) -> Result<(), Error> {
check_destination(dest_network, "use send_apdu for a local device")?;
if dest_mac.is_empty() {
return Err(Error::Encoding(format!(
"an empty dest_mac addresses the broadcast on network {dest_network}, \
not one device; use broadcast_to_network for a remote broadcast"
)));
}
let buf =
Self::encode_routed_npdu_buf(apdu, dest_network, dest_mac, expecting_reply, priority)?;
self.transport
.send_broadcast_with_data_attributes(&buf, data_attributes)
.await
}
#[doc(hidden)]
pub fn retain_network_port_lease_internal(&mut self, lease: Arc<()>) -> Result<(), Error> {
self.transport.retain_network_port_lease_internal(lease)
}
pub fn transport(&self) -> &T {
&self.transport
}
pub fn local_mac(&self) -> &[u8] {
self.transport.local_mac()
}
pub fn local_network_number(&self) -> &LocalNetworkNumber {
&self.local_network_number
}
pub fn network_control_ingress_sequence(&self) -> u64 {
self.network_control_ingress_sequence.load(Ordering::SeqCst)
}
pub fn address_length_drops(&self) -> u64 {
self.address_length_drops.load(Ordering::Relaxed)
}
pub fn global_broadcast_dadr_drops(&self) -> u64 {
self.global_broadcast_dadr_drops.load(Ordering::Relaxed)
}
pub fn broadcast_pdu_type_drops(&self) -> u64 {
self.broadcast_pdu_type_drops.load(Ordering::Relaxed)
}
fn encode_routed_npdu_buf(
apdu: &[u8],
dest_network: u16,
dest_mac: &[u8],
expecting_reply: bool,
priority: NetworkPriority,
) -> Result<BytesMut, Error> {
let npdu = Npdu {
is_network_message: false,
expecting_reply,
priority,
destination: Some(NpduAddress {
network: dest_network,
mac_address: MacAddr::from_slice(dest_mac),
}),
source: None,
hop_count: 255,
payload: Bytes::copy_from_slice(apdu),
..Npdu::default()
};
let mut buf = BytesMut::with_capacity(8 + dest_mac.len() + apdu.len());
encode_npdu(&mut buf, &npdu)?;
Ok(buf)
}
pub async fn stop(&mut self) -> Result<(), Error> {
self.seal_responses();
if let Some(task) = self.abort_dispatch_task() {
let _ = task.await;
}
self.transport.stop().await
}
}
pub(crate) fn next_ingress_sequence(sequence: &AtomicU64) -> u64 {
#[allow(deprecated, reason = "try_update needs Rust 1.95; the MSRV is 1.93")]
let previous = sequence
.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |value| {
value.checked_add(1)
})
.unwrap_or(u64::MAX);
previous.saturating_add(1)
}
pub(crate) fn count_drop(counter: &AtomicU64) {
#[allow(deprecated, reason = "try_update needs Rust 1.95; the MSRV is 1.93")]
let _ = counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |n| n.checked_add(1));
}
impl<T: TransportPort> NetworkLayer<T> {
fn abort_dispatch_task(&mut self) -> Option<JoinHandle<()>> {
let task = self.dispatch_task.take()?;
task.abort();
Some(task)
}
}
impl<T: TransportPort> Drop for NetworkLayer<T> {
fn drop(&mut self) {
self.response_scope.seal();
let _ = self.abort_dispatch_task();
self.transport.abort();
}
}
#[cfg(test)]
#[path = "issuance_tests.rs"]
mod issuance_tests;
#[cfg(test)]
mod tests {
use super::*;
use bacnet_encoding::npdu::decode_npdu;
use bacnet_transport::bip::BipTransport;
use bacnet_transport::sc::{LoopbackWebSocket, ScTransport, WebSocketPort};
use bacnet_transport::sc_frame::{
decode_sc_message, encode_sc_message, ScFunction, ScMessage, Vmac,
};
use std::net::Ipv4Addr;
use tokio::time::{timeout, Duration};
async fn sc_hub_accept(ws_hub: &LoopbackWebSocket, hub_vmac: Vmac) {
let data = ws_hub.recv().await.unwrap();
let req = decode_sc_message(&data).unwrap();
assert_eq!(req.function, ScFunction::ConnectRequest);
let mut accept_payload = Vec::with_capacity(26);
accept_payload.extend_from_slice(&hub_vmac);
accept_payload.extend_from_slice(&[0x33; 16]);
accept_payload.extend_from_slice(&1476u16.to_be_bytes());
accept_payload.extend_from_slice(&1476u16.to_be_bytes());
let accept = ScMessage {
function: ScFunction::ConnectAccept,
message_id: req.message_id,
originating_vmac: None,
destination_vmac: None,
dest_options: Vec::new(),
data_options: Vec::new(),
payload: Bytes::from(accept_payload),
};
let mut buf = BytesMut::new();
encode_sc_message(&mut buf, &accept);
ws_hub.send(&buf).await.unwrap();
}
async fn assert_sc_socket_closed_after_drop(ws_hub: &LoopbackWebSocket, context: &str) {
timeout(Duration::from_secs(1), async {
while let Ok(data) = ws_hub.recv().await {
let msg = decode_sc_message(&data).unwrap();
assert_ne!(
msg.function,
ScFunction::HeartbeatAck,
"{context} must not leave SC answering heartbeats"
);
}
})
.await
.unwrap_or_else(|_| panic!("{context} did not close the SC WebSocket"));
let heartbeat = ScMessage {
function: ScFunction::HeartbeatRequest,
message_id: 0x66,
originating_vmac: None,
destination_vmac: None,
dest_options: Vec::new(),
data_options: Vec::new(),
payload: Bytes::new(),
};
let mut buf = BytesMut::new();
encode_sc_message(&mut buf, &heartbeat);
assert!(
ws_hub.send(&buf).await.is_err(),
"{context} must reject post-drop Heartbeat-Request on the closed socket"
);
}
#[tokio::test]
async fn network_layer_drop_releases_sc_transport_socket() {
let (ws_client, ws_hub) = LoopbackWebSocket::pair();
let hub_vmac = [0x10; 6];
let mut net =
NetworkLayer::new(ScTransport::new(ws_client, [0x01; 6]).with_device_uuid([1; 16]));
let hub_accept_task = tokio::spawn(async move {
sc_hub_accept(&ws_hub, hub_vmac).await;
ws_hub
});
let _rx = net.start().await.unwrap();
let ws_hub = hub_accept_task.await.unwrap();
drop(net);
assert_sc_socket_closed_after_drop(&ws_hub, "dropped NetworkLayer").await;
}
#[tokio::test]
async fn send_apdu_data_attributes_reach_sc_data_options() {
let (ws_client, ws_hub) = LoopbackWebSocket::pair();
let hub_vmac = [0x10; 6];
let dest_vmac: Vmac = [0x02, 0x03, 0x04, 0x05, 0x06, 0x07];
let mut net =
NetworkLayer::new(ScTransport::new(ws_client, [0x01; 6]).with_device_uuid([1; 16]));
let data_attributes = vec![
DataAttribute {
option_type: 1,
must_understand: true,
data: Vec::new(),
},
DataAttribute {
option_type: 31,
must_understand: false,
data: vec![0x12, 0x34, 0x56],
},
];
let hub_accept_task = tokio::spawn(async move {
sc_hub_accept(&ws_hub, hub_vmac).await;
ws_hub
});
let _rx = net.start().await.unwrap();
let ws_hub = hub_accept_task.await.unwrap();
let apdu = Bytes::from_static(&[0x10, 0x08]);
net.send_apdu_with_data_attributes(
&apdu,
&dest_vmac,
false,
NetworkPriority::NORMAL,
&data_attributes,
)
.await
.unwrap();
let data = ws_hub.recv().await.unwrap();
let msg = decode_sc_message(&data).unwrap();
assert_eq!(msg.function, ScFunction::EncapsulatedNpdu);
assert_eq!(msg.destination_vmac, Some(dest_vmac));
assert_eq!(msg.data_options.len(), 2);
assert_eq!(msg.data_options[0].option_type, 1);
assert!(msg.data_options[0].must_understand);
assert_eq!(msg.data_options[1].option_type, 31);
assert!(!msg.data_options[1].must_understand);
assert_eq!(msg.data_options[1].data, vec![0x12, 0x34, 0x56]);
let npdu = decode_npdu(msg.payload).unwrap();
assert_eq!(npdu.payload, apdu);
assert!(!npdu.expecting_reply);
assert_eq!(npdu.priority, NetworkPriority::NORMAL);
net.stop().await.unwrap();
}
#[tokio::test]
async fn end_to_end_who_is() {
use bacnet_encoding::apdu::{decode_apdu, encode_apdu, Apdu, UnconfirmedRequest};
use bacnet_types::enums::UnconfirmedServiceChoice;
let transport_a = BipTransport::new(Ipv4Addr::LOCALHOST, 0, Ipv4Addr::BROADCAST);
let transport_b = BipTransport::new(Ipv4Addr::LOCALHOST, 0, Ipv4Addr::BROADCAST);
let mut net_a = NetworkLayer::new(transport_a);
let mut net_b = NetworkLayer::new(transport_b);
let _rx_a = net_a.start().await.unwrap();
let mut rx_b = net_b.start().await.unwrap();
let who_is_apdu = Apdu::UnconfirmedRequest(UnconfirmedRequest {
service_choice: UnconfirmedServiceChoice::WHO_IS,
service_request: Bytes::new(),
});
let mut apdu_buf = BytesMut::new();
encode_apdu(&mut apdu_buf, &who_is_apdu).expect("valid APDU encoding");
net_a
.send_apdu(&apdu_buf, net_b.local_mac(), false, NetworkPriority::NORMAL)
.await
.unwrap();
let received = timeout(Duration::from_secs(2), rx_b.recv())
.await
.expect("Timed out waiting for APDU")
.expect("Channel closed");
let decoded_apdu = decode_apdu(received.apdu.clone()).unwrap();
match decoded_apdu {
Apdu::UnconfirmedRequest(req) => {
assert_eq!(req.service_choice, UnconfirmedServiceChoice::WHO_IS);
assert!(req.service_request.is_empty());
}
other => panic!("Expected UnconfirmedRequest, got {:?}", other),
}
net_a.stop().await.unwrap();
net_b.stop().await.unwrap();
}
#[test]
fn transport_accessor() {
let transport = BipTransport::new(Ipv4Addr::LOCALHOST, 0, Ipv4Addr::BROADCAST);
let net = NetworkLayer::new(transport);
let mac = net.transport().local_mac();
assert_eq!(mac.len(), 6);
}
}
#[cfg(test)]
#[path = "layer_delivery_tests.rs"]
mod delivery_tests;
#[cfg(test)]
#[path = "layer_admission_tests.rs"]
mod admission_tests;
#[cfg(test)]
#[path = "layer_sc_data_options_tests.rs"]
mod sc_data_options_tests;
#[cfg(test)]
#[path = "broadcast_pdu_tests.rs"]
mod broadcast_pdu_tests;