1use crate::transport::TransportError;
8
9use super::addr::BleAddr;
10
11pub trait BleStream: Send + Sync {
17 fn send(
19 &self,
20 data: &[u8],
21 ) -> impl std::future::Future<Output = Result<(), TransportError>> + Send;
22
23 fn recv(
27 &self,
28 buf: &mut [u8],
29 ) -> impl std::future::Future<Output = Result<usize, TransportError>> + Send;
30
31 fn send_mtu(&self) -> u16;
33
34 fn recv_mtu(&self) -> u16;
36
37 fn remote_addr(&self) -> &BleAddr;
39}
40
41pub trait BleAcceptor: Send {
43 type Stream: BleStream + 'static;
45
46 fn accept(
48 &mut self,
49 ) -> impl std::future::Future<Output = Result<Self::Stream, TransportError>> + Send;
50}
51
52pub trait BleScanner: Send {
54 fn next(&mut self) -> impl std::future::Future<Output = Option<BleAddr>> + Send;
58}
59
60pub trait BleIo: Send + Sync + 'static {
66 type Stream: BleStream + 'static;
68 type Acceptor: BleAcceptor<Stream = Self::Stream> + 'static;
70 type Scanner: BleScanner + 'static;
72
73 fn listen(
75 &self,
76 psm: u16,
77 ) -> impl std::future::Future<Output = Result<Self::Acceptor, TransportError>> + Send;
78
79 fn connect(
81 &self,
82 addr: &BleAddr,
83 psm: u16,
84 ) -> impl std::future::Future<Output = Result<Self::Stream, TransportError>> + Send;
85
86 fn start_advertising(
88 &self,
89 ) -> impl std::future::Future<Output = Result<(), TransportError>> + Send;
90
91 fn stop_advertising(
93 &self,
94 ) -> impl std::future::Future<Output = Result<(), TransportError>> + Send;
95
96 fn start_scanning(
98 &self,
99 ) -> impl std::future::Future<Output = Result<Self::Scanner, TransportError>> + Send;
100
101 fn local_addr(&self) -> Result<BleAddr, TransportError>;
103
104 fn adapter_name(&self) -> &str;
106}
107
108#[cfg(bluer_available)]
113mod bluer_impl {
114 use super::*;
115 use crate::transport::TransportError;
116
117 use bluer::l2cap::{SeqPacket, SeqPacketListener, Socket, SocketAddr};
118 use bluer::{
119 AdapterEvent, AddressType, DiscoveryFilter, DiscoveryTransport, adv::Advertisement,
120 };
121 use futures::StreamExt;
122 use std::collections::{BTreeSet, HashSet};
123 use std::pin::Pin;
124 use tokio::sync::Mutex;
125 use tracing::{debug, trace};
126
127 pub const FIPS_SERVICE_UUID: bluer::Uuid =
132 bluer::Uuid::from_u128(0x9c90_b790_2cc5_42c0_9f87_c9cc_4064_8f4c);
133
134 fn map_err(context: &str, e: bluer::Error) -> TransportError {
136 TransportError::Io(std::io::Error::other(format!("{}: {}", context, e)))
137 }
138
139 fn map_io_err(context: &str, e: std::io::Error) -> TransportError {
141 TransportError::Io(std::io::Error::new(e.kind(), format!("{}: {}", context, e)))
142 }
143
144 pub struct BluerStream {
150 conn: SeqPacket,
151 remote: BleAddr,
152 send_mtu: u16,
153 recv_mtu: u16,
154 }
155
156 impl BluerStream {
157 pub fn new(conn: SeqPacket, remote: BleAddr) -> Result<Self, TransportError> {
159 let send_mtu = conn.send_mtu().map_err(|e| map_io_err("send_mtu", e))? as u16;
160 let recv_mtu = conn.recv_mtu().map_err(|e| map_io_err("recv_mtu", e))? as u16;
161
162 match conn.as_ref().phy() {
164 Ok(phy) => {
165 debug!(addr = %remote, phy, send_mtu, recv_mtu, "BLE connection established")
166 }
167 Err(_) => {
168 debug!(addr = %remote, send_mtu, recv_mtu, "BLE connection established (PHY query unsupported)")
169 }
170 }
171
172 Ok(Self {
173 conn,
174 remote,
175 send_mtu,
176 recv_mtu,
177 })
178 }
179 }
180
181 impl BleStream for BluerStream {
182 async fn send(&self, data: &[u8]) -> Result<(), TransportError> {
183 self.conn
184 .send(data)
185 .await
186 .map(|_| ())
187 .map_err(|e| TransportError::SendFailed(format!("{}", e)))
188 }
189
190 async fn recv(&self, buf: &mut [u8]) -> Result<usize, TransportError> {
191 self.conn
192 .recv(buf)
193 .await
194 .map_err(|e| TransportError::RecvFailed(format!("{}", e)))
195 }
196
197 fn send_mtu(&self) -> u16 {
198 self.send_mtu
199 }
200
201 fn recv_mtu(&self) -> u16 {
202 self.recv_mtu
203 }
204
205 fn remote_addr(&self) -> &BleAddr {
206 &self.remote
207 }
208 }
209
210 pub struct BluerAcceptor {
216 listener: SeqPacketListener,
217 adapter_name: String,
218 }
219
220 impl BleAcceptor for BluerAcceptor {
221 type Stream = BluerStream;
222
223 async fn accept(&mut self) -> Result<BluerStream, TransportError> {
224 let (conn, peer_sa) = self
225 .listener
226 .accept()
227 .await
228 .map_err(|e| map_io_err("accept", e))?;
229
230 let remote = BleAddr::from_bluer(peer_sa.addr, &self.adapter_name);
231 BluerStream::new(conn, remote)
232 }
233 }
234
235 pub struct BluerScanner {
241 events: Pin<Box<dyn futures::Stream<Item = AdapterEvent> + Send>>,
242 adapter: bluer::Adapter,
243 adapter_name: String,
244 }
245
246 impl BleScanner for BluerScanner {
247 async fn next(&mut self) -> Option<BleAddr> {
248 loop {
249 match self.events.next().await {
250 Some(AdapterEvent::DeviceAdded(addr)) => {
251 if let Ok(device) = self.adapter.device(addr) {
253 match device.uuids().await {
254 Ok(Some(uuids)) if uuids.contains(&FIPS_SERVICE_UUID) => {
255 let ble_addr = BleAddr::from_bluer(addr, &self.adapter_name);
256 debug!(addr = %ble_addr, "BLE scanner: FIPS peer found");
257 return Some(ble_addr);
258 }
259 Ok(_) => {
260 trace!(addr = %addr, "BLE scanner: device without FIPS UUID");
261 }
262 Err(e) => {
263 trace!(addr = %addr, error = %e, "BLE scanner: failed to read UUIDs");
264 }
265 }
266 }
267 }
268 Some(_) => continue,
269 None => return None,
270 }
271 }
272 }
273 }
274
275 pub struct BluerIo {
281 _session: bluer::Session,
282 adapter: bluer::Adapter,
283 adapter_name: String,
284 adv_handle: Mutex<Option<bluer::adv::AdvertisementHandle>>,
285 mtu: u16,
286 }
287
288 impl BluerIo {
289 pub async fn new(adapter_name: &str, mtu: u16) -> Result<Self, TransportError> {
293 let session = bluer::Session::new()
294 .await
295 .map_err(|e| map_err("Session::new", e))?;
296
297 let adapter = if adapter_name == "default" {
298 session
299 .default_adapter()
300 .await
301 .map_err(|e| map_err("default_adapter", e))?
302 } else {
303 session
304 .adapter(adapter_name)
305 .map_err(|e| map_err("adapter", e))?
306 };
307
308 adapter
309 .set_powered(true)
310 .await
311 .map_err(|e| map_err("set_powered", e))?;
312
313 let name = adapter.name().to_string();
314 debug!(adapter = %name, "BluerIo initialized");
315
316 Ok(Self {
317 _session: session,
318 adapter,
319 adapter_name: name,
320 adv_handle: Mutex::new(None),
321 mtu,
322 })
323 }
324 }
325
326 impl BleIo for BluerIo {
327 type Stream = BluerStream;
328 type Acceptor = BluerAcceptor;
329 type Scanner = BluerScanner;
330
331 async fn listen(&self, psm: u16) -> Result<Self::Acceptor, TransportError> {
332 let local_addr = self
333 .adapter
334 .address()
335 .await
336 .map_err(|e| map_err("address", e))?;
337
338 let sa = SocketAddr::new(local_addr, AddressType::LePublic, psm);
339 let listener = SeqPacketListener::bind(sa)
340 .await
341 .map_err(|e| map_io_err("bind", e))?;
342
343 listener
345 .as_ref()
346 .set_recv_mtu(self.mtu)
347 .map_err(|e| map_io_err("set_recv_mtu", e))?;
348
349 if let Err(e) = listener.as_ref().set_power_forced_active(true) {
351 debug!(error = %e, "BLE listener: set_power_forced_active not supported");
352 }
353
354 debug!(psm, mtu = self.mtu, "BLE listener bound");
355
356 Ok(BluerAcceptor {
357 listener,
358 adapter_name: self.adapter_name.clone(),
359 })
360 }
361
362 async fn connect(&self, addr: &BleAddr, psm: u16) -> Result<Self::Stream, TransportError> {
363 let target_sa = addr.to_socket_addr(psm);
364
365 let socket = Socket::<SeqPacket>::new_seq_packet()
366 .map_err(|e| map_io_err("new_seq_packet", e))?;
367 socket
368 .bind(SocketAddr::any_le())
369 .map_err(|e| map_io_err("bind", e))?;
370 socket
371 .set_recv_mtu(self.mtu)
372 .map_err(|e| map_io_err("set_recv_mtu", e))?;
373
374 if let Err(e) = socket.set_power_forced_active(true) {
376 debug!(error = %e, "BLE connect: set_power_forced_active not supported");
377 }
378
379 let conn = socket
380 .connect(target_sa)
381 .await
382 .map_err(|e| map_io_err("connect", e))?;
383
384 let remote = addr.clone();
385 BluerStream::new(conn, remote)
386 }
387
388 async fn start_advertising(&self) -> Result<(), TransportError> {
389 let adv = Advertisement {
390 advertisement_type: bluer::adv::Type::Peripheral,
391 service_uuids: {
392 let mut s = BTreeSet::new();
393 s.insert(FIPS_SERVICE_UUID);
394 s
395 },
396 local_name: Some("fips".to_string()),
397 min_interval: Some(std::time::Duration::from_millis(400)),
398 max_interval: Some(std::time::Duration::from_millis(600)),
399 ..Default::default()
400 };
401
402 let handle = self
403 .adapter
404 .advertise(adv)
405 .await
406 .map_err(|e| map_err("advertise", e))?;
407
408 *self.adv_handle.lock().await = Some(handle);
409 debug!("BLE advertising started");
410 Ok(())
411 }
412
413 async fn stop_advertising(&self) -> Result<(), TransportError> {
414 let _ = self.adv_handle.lock().await.take();
415 debug!("BLE advertising stopped");
416 Ok(())
417 }
418
419 async fn start_scanning(&self) -> Result<Self::Scanner, TransportError> {
420 if let Ok(cached) = self.adapter.device_addresses().await {
426 let count = cached.len();
427 for addr in cached {
428 let _ = self.adapter.remove_device(addr).await;
429 }
430 if count > 0 {
431 debug!(count, "BLE scanner: cleared cached devices");
432 }
433 }
434
435 let filter = DiscoveryFilter {
437 transport: DiscoveryTransport::Le,
438 uuids: {
439 let mut s = HashSet::new();
440 s.insert(FIPS_SERVICE_UUID);
441 s
442 },
443 ..Default::default()
444 };
445
446 self.adapter
447 .set_discovery_filter(filter)
448 .await
449 .map_err(|e| map_err("set_discovery_filter", e))?;
450
451 let events = self
452 .adapter
453 .discover_devices()
454 .await
455 .map_err(|e| map_err("discover_devices", e))?;
456
457 debug!("BLE scanning started");
458
459 Ok(BluerScanner {
460 events: Box::pin(events),
461 adapter: self.adapter.clone(),
462 adapter_name: self.adapter_name.clone(),
463 })
464 }
465
466 fn local_addr(&self) -> Result<BleAddr, TransportError> {
467 let addr = futures::executor::block_on(self.adapter.address())
471 .map_err(|e| map_err("address", e))?;
472 Ok(BleAddr::from_bluer(addr, &self.adapter_name))
473 }
474
475 fn adapter_name(&self) -> &str {
476 &self.adapter_name
477 }
478 }
479
480 const _: () = {
482 fn require<T: Send + Sync>() {}
483 let _ = require::<BluerIo>;
484 };
485}
486
487#[cfg(bluer_available)]
488pub use bluer_impl::{BluerAcceptor, BluerIo, BluerScanner, BluerStream, FIPS_SERVICE_UUID};
489
490pub struct MockBleStream {
496 addr: BleAddr,
497 send_mtu: u16,
498 recv_mtu: u16,
499 tx: tokio::sync::mpsc::Sender<Vec<u8>>,
500 rx: tokio::sync::Mutex<tokio::sync::mpsc::Receiver<Vec<u8>>>,
501}
502
503impl MockBleStream {
504 pub fn pair(addr_a: BleAddr, addr_b: BleAddr, mtu: u16) -> (Self, Self) {
506 let (tx_a, rx_a) = tokio::sync::mpsc::channel(64);
507 let (tx_b, rx_b) = tokio::sync::mpsc::channel(64);
508 let stream_a = Self {
509 addr: addr_b.clone(),
510 send_mtu: mtu,
511 recv_mtu: mtu,
512 tx: tx_a,
513 rx: tokio::sync::Mutex::new(rx_b),
514 };
515 let stream_b = Self {
516 addr: addr_a,
517 send_mtu: mtu,
518 recv_mtu: mtu,
519 tx: tx_b,
520 rx: tokio::sync::Mutex::new(rx_a),
521 };
522 (stream_a, stream_b)
523 }
524}
525
526impl BleStream for MockBleStream {
527 async fn send(&self, data: &[u8]) -> Result<(), TransportError> {
528 self.tx
529 .send(data.to_vec())
530 .await
531 .map_err(|_| TransportError::SendFailed("channel closed".into()))
532 }
533
534 async fn recv(&self, buf: &mut [u8]) -> Result<usize, TransportError> {
535 let mut rx = self.rx.lock().await;
536 match rx.recv().await {
537 Some(data) => {
538 let len = data.len().min(buf.len());
539 buf[..len].copy_from_slice(&data[..len]);
540 Ok(len)
541 }
542 None => Ok(0), }
544 }
545
546 fn send_mtu(&self) -> u16 {
547 self.send_mtu
548 }
549
550 fn recv_mtu(&self) -> u16 {
551 self.recv_mtu
552 }
553
554 fn remote_addr(&self) -> &BleAddr {
555 &self.addr
556 }
557}
558
559pub struct MockBleAcceptor {
561 rx: tokio::sync::mpsc::Receiver<MockBleStream>,
562}
563
564impl BleAcceptor for MockBleAcceptor {
565 type Stream = MockBleStream;
566
567 async fn accept(&mut self) -> Result<MockBleStream, TransportError> {
568 self.rx
569 .recv()
570 .await
571 .ok_or(TransportError::RecvFailed("acceptor channel closed".into()))
572 }
573}
574
575pub struct MockBleScanner {
577 rx: tokio::sync::mpsc::Receiver<BleAddr>,
578}
579
580impl BleScanner for MockBleScanner {
581 async fn next(&mut self) -> Option<BleAddr> {
582 self.rx.recv().await
583 }
584}
585
586type ConnectHandler =
588 Box<dyn Fn(&BleAddr, u16) -> Result<MockBleStream, TransportError> + Send + Sync>;
589
590pub struct MockBleIo {
595 adapter: String,
596 local_addr: BleAddr,
597 accept_tx: tokio::sync::mpsc::Sender<MockBleStream>,
598 accept_rx: std::sync::Mutex<Option<tokio::sync::mpsc::Receiver<MockBleStream>>>,
599 scan_tx: tokio::sync::mpsc::Sender<BleAddr>,
600 scan_rx: std::sync::Mutex<Option<tokio::sync::mpsc::Receiver<BleAddr>>>,
601 connect_handler: std::sync::Mutex<Option<ConnectHandler>>,
602}
603
604impl MockBleIo {
605 pub fn new(adapter: &str, local_addr: BleAddr) -> Self {
607 let (accept_tx, accept_rx) = tokio::sync::mpsc::channel(16);
608 let (scan_tx, scan_rx) = tokio::sync::mpsc::channel(64);
609 Self {
610 adapter: adapter.to_string(),
611 local_addr,
612 accept_tx,
613 accept_rx: std::sync::Mutex::new(Some(accept_rx)),
614 scan_tx,
615 scan_rx: std::sync::Mutex::new(Some(scan_rx)),
616 connect_handler: std::sync::Mutex::new(None),
617 }
618 }
619
620 pub async fn inject_inbound(&self, stream: MockBleStream) {
622 let _ = self.accept_tx.send(stream).await;
623 }
624
625 pub async fn inject_scan_result(&self, addr: BleAddr) {
627 let _ = self.scan_tx.send(addr).await;
628 }
629
630 pub fn set_connect_handler<F>(&self, handler: F)
632 where
633 F: Fn(&BleAddr, u16) -> Result<MockBleStream, TransportError> + Send + Sync + 'static,
634 {
635 *self
636 .connect_handler
637 .lock()
638 .unwrap_or_else(|e| e.into_inner()) = Some(Box::new(handler));
639 }
640}
641
642impl BleIo for MockBleIo {
643 type Stream = MockBleStream;
644 type Acceptor = MockBleAcceptor;
645 type Scanner = MockBleScanner;
646
647 async fn listen(&self, _psm: u16) -> Result<Self::Acceptor, TransportError> {
648 let rx = self
649 .accept_rx
650 .lock()
651 .unwrap_or_else(|e| e.into_inner())
652 .take()
653 .ok_or_else(|| TransportError::NotSupported("acceptor already taken".into()))?;
654 Ok(MockBleAcceptor { rx })
655 }
656
657 async fn connect(&self, addr: &BleAddr, psm: u16) -> Result<Self::Stream, TransportError> {
658 let handler = self
659 .connect_handler
660 .lock()
661 .unwrap_or_else(|e| e.into_inner());
662 match handler.as_ref() {
663 Some(f) => f(addr, psm),
664 None => Err(TransportError::ConnectionRefused),
665 }
666 }
667
668 async fn start_advertising(&self) -> Result<(), TransportError> {
669 Ok(())
670 }
671
672 async fn stop_advertising(&self) -> Result<(), TransportError> {
673 Ok(())
674 }
675
676 async fn start_scanning(&self) -> Result<Self::Scanner, TransportError> {
677 let rx = self
678 .scan_rx
679 .lock()
680 .unwrap_or_else(|e| e.into_inner())
681 .take()
682 .ok_or_else(|| TransportError::NotSupported("scanner already taken".into()))?;
683 Ok(MockBleScanner { rx })
684 }
685
686 fn local_addr(&self) -> Result<BleAddr, TransportError> {
687 Ok(self.local_addr.clone())
688 }
689
690 fn adapter_name(&self) -> &str {
691 &self.adapter
692 }
693}
694
695#[cfg(test)]
700mod tests {
701 use super::*;
702
703 fn test_addr(n: u8) -> BleAddr {
704 BleAddr {
705 adapter: "hci0".to_string(),
706 device: [0xAA, 0xBB, 0xCC, 0xDD, 0xEE, n],
707 }
708 }
709
710 #[tokio::test]
711 async fn test_mock_stream_pair_send_recv() {
712 let (a, b) = MockBleStream::pair(test_addr(1), test_addr(2), 2048);
713
714 a.send(b"hello").await.unwrap();
715 let mut buf = [0u8; 64];
716 let n = b.recv(&mut buf).await.unwrap();
717 assert_eq!(&buf[..n], b"hello");
718
719 b.send(b"world").await.unwrap();
720 let n = a.recv(&mut buf).await.unwrap();
721 assert_eq!(&buf[..n], b"world");
722 }
723
724 #[tokio::test]
725 async fn test_mock_stream_mtu() {
726 let (a, b) = MockBleStream::pair(test_addr(1), test_addr(2), 512);
727 assert_eq!(a.send_mtu(), 512);
728 assert_eq!(a.recv_mtu(), 512);
729 assert_eq!(b.send_mtu(), 512);
730 assert_eq!(b.recv_mtu(), 512);
731 }
732
733 #[tokio::test]
734 async fn test_mock_stream_remote_addr() {
735 let (a, b) = MockBleStream::pair(test_addr(1), test_addr(2), 2048);
736 assert_eq!(a.remote_addr(), &test_addr(2));
737 assert_eq!(b.remote_addr(), &test_addr(1));
738 }
739
740 #[tokio::test]
741 async fn test_mock_io_listen_accept() {
742 let io = MockBleIo::new("hci0", test_addr(1));
743 let mut acceptor = io.listen(0x0085).await.unwrap();
744
745 let (stream_a, _stream_b) = MockBleStream::pair(test_addr(1), test_addr(2), 2048);
746 io.inject_inbound(stream_a).await;
747
748 let accepted = acceptor.accept().await.unwrap();
749 assert_eq!(accepted.remote_addr(), &test_addr(2));
751 }
752
753 #[tokio::test]
754 async fn test_mock_io_connect() {
755 let io = MockBleIo::new("hci0", test_addr(1));
756 let local = test_addr(1);
757 io.set_connect_handler(move |addr, _psm| {
758 let (stream, _peer) = MockBleStream::pair(local.clone(), addr.clone(), 2048);
759 Ok(stream)
760 });
761
762 let stream = io.connect(&test_addr(2), 0x0085).await.unwrap();
763 assert_eq!(stream.remote_addr(), &test_addr(2));
764 }
765
766 #[tokio::test]
767 async fn test_mock_io_connect_no_handler() {
768 let io = MockBleIo::new("hci0", test_addr(1));
769 let result = io.connect(&test_addr(2), 0x0085).await;
770 assert!(result.is_err());
771 }
772
773 #[tokio::test]
774 async fn test_mock_io_scan() {
775 let io = MockBleIo::new("hci0", test_addr(1));
776 let mut scanner = io.start_scanning().await.unwrap();
777
778 io.inject_scan_result(test_addr(2)).await;
779 io.inject_scan_result(test_addr(3)).await;
780
781 assert_eq!(scanner.next().await, Some(test_addr(2)));
782 assert_eq!(scanner.next().await, Some(test_addr(3)));
783 }
784
785 #[tokio::test]
786 async fn test_mock_io_local_addr() {
787 let io = MockBleIo::new("hci0", test_addr(1));
788 assert_eq!(io.local_addr().unwrap(), test_addr(1));
789 assert_eq!(io.adapter_name(), "hci0");
790 }
791
792 #[tokio::test]
793 async fn test_mock_io_advertising_noop() {
794 let io = MockBleIo::new("hci0", test_addr(1));
795 io.start_advertising().await.unwrap();
796 io.stop_advertising().await.unwrap();
797 }
798
799 #[tokio::test]
800 async fn test_mock_io_listen_twice_fails() {
801 let io = MockBleIo::new("hci0", test_addr(1));
802 let _acceptor = io.listen(0x0085).await.unwrap();
803 assert!(io.listen(0x0085).await.is_err());
804 }
805}