bt_obex/client/
mod.rs

1// Copyright 2023 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use fuchsia_bluetooth::types::Channel;
6use futures::SinkExt;
7use log::{trace, warn};
8
9pub use crate::client::get::GetOperation;
10pub use crate::client::put::PutOperation;
11use crate::error::Error;
12use crate::header::{
13    ConnectionIdentifier, Header, HeaderIdentifier, HeaderSet, SingleResponseMode,
14};
15use crate::operation::{OpCode, RequestPacket, ResponseCode, ResponsePacket, SetPathFlags};
16pub use crate::transport::TransportType;
17use crate::transport::{ObexTransportManager, max_packet_size_from_transport};
18use fuchsia_sync::Mutex;
19
20/// Implements the OBEX PUT operation.
21mod put;
22
23/// Implements the OBEX GET operation.
24mod get;
25
26/// An interface for the Single Response Mode (SRM) feature for an OBEX client operation.
27pub(crate) trait SrmOperation {
28    const OPERATION_TYPE: OpCode;
29
30    /// Returns the current SRM mode.
31    fn get_srm(&self) -> SingleResponseMode;
32
33    /// Sets SRM to the provided `mode`.
34    fn set_srm(&mut self, mode: SingleResponseMode);
35
36    /// Attempts to enable SRM for the operation by updating the provided `headers` with the SRM
37    /// header.
38    /// Returns Error if `headers` couldn't be updated with SRM, Ok otherwise.
39    fn try_enable_srm(&mut self, headers: &mut HeaderSet) -> Result<(), Error> {
40        let requested_srm = headers.try_add_srm(self.get_srm())?;
41        self.set_srm(requested_srm);
42        trace!(operation:? = Self::OPERATION_TYPE; "Requesting SRM {requested_srm:?}");
43        Ok(())
44    }
45
46    /// Checks the provided response `headers` for the SRM flag and updates the local SRM state
47    /// for the operation.
48    fn check_response_for_srm(&mut self, headers: &HeaderSet) {
49        let srm_response = if let Some(Header::SingleResponseMode(srm)) =
50            headers.get(&HeaderIdentifier::SingleResponseMode)
51        {
52            *srm
53        } else {
54            // No SRM indication from peer defaults to disabled.
55            trace!(operation:? = Self::OPERATION_TYPE; "Response doesn't contain SRM header");
56            SingleResponseMode::Disable
57        };
58
59        trace!(current_status:? = self.get_srm(), operation:? = Self::OPERATION_TYPE; "Peer responded with {srm_response:?}");
60        match (srm_response, self.get_srm()) {
61            (SingleResponseMode::Enable, SingleResponseMode::Disable) => {
62                warn!("SRM stays disabled");
63            }
64            (SingleResponseMode::Disable, SingleResponseMode::Enable) => {
65                trace!("SRM is disabled");
66                self.set_srm(SingleResponseMode::Disable);
67            }
68            _ => {} // Otherwise, both sides agree on the SRM status.
69        }
70        trace!(status:? = self.get_srm(), operation:? = Self::OPERATION_TYPE; "SRM status");
71    }
72}
73
74#[derive(Clone, Copy, Debug, PartialEq, Default)]
75enum ConnectionStatus {
76    /// The transport is created but the CONNECT operation has not been completed.
77    #[default]
78    Initialized,
79    /// The CONNECT operation has been completed and the transport is considered connected.
80    /// `id` contains the optional identifier for this connection. This is configured during the
81    /// CONNECT operation and is a unique value assigned by the remote OBEX server.
82    Connected { id: Option<ConnectionIdentifier> },
83    /// The transport is considered disconnected -- the OBEX client has Disconnected the session.
84    Disconnected,
85}
86
87impl ConnectionStatus {
88    #[cfg(test)]
89    fn connected_no_id() -> Self {
90        Self::Connected { id: None }
91    }
92}
93
94/// The Client role for an OBEX session.
95/// Provides an interface for connecting to a remote OBEX server and initiating the operations
96/// specified in OBEX 1.5.
97#[derive(Debug)]
98pub struct ObexClient {
99    /// Whether the CONNECT operation has completed.
100    connected: Mutex<ConnectionStatus>,
101    /// The maximum OBEX packet length for this OBEX session.
102    max_packet_size: Mutex<u16>,
103    /// Manages the RFCOMM or L2CAP transport and provides a reservation system for starting
104    /// new operations.
105    transport: ObexTransportManager,
106}
107
108impl ObexClient {
109    pub fn new(channel: Channel, type_: TransportType) -> Self {
110        let max_packet_size = max_packet_size_from_transport(channel.max_tx_size());
111        let transport = ObexTransportManager::new(channel, type_);
112        Self {
113            connected: Mutex::new(ConnectionStatus::default()),
114            max_packet_size: Mutex::new(max_packet_size),
115            transport,
116        }
117    }
118
119    pub fn is_transport_connected(&self) -> bool {
120        !self.transport.is_transport_closed()
121    }
122
123    fn set_connection_status(&self, status: ConnectionStatus) {
124        *self.connected.lock() = status;
125    }
126
127    fn connection_status(&self) -> ConnectionStatus {
128        *self.connected.lock()
129    }
130
131    pub fn is_connected(&self) -> bool {
132        matches!(*self.connected.lock(), ConnectionStatus::Connected { .. })
133    }
134
135    pub fn connection_id(&self) -> Option<ConnectionIdentifier> {
136        match self.connection_status() {
137            ConnectionStatus::Connected { id } => id.clone(),
138            _ => None,
139        }
140    }
141
142    fn set_max_packet_size(&self, peer_max_packet_size: u16) {
143        // We have no opinion on the preferred max packet size, so just use the peer's.
144        *self.max_packet_size.lock() = peer_max_packet_size;
145        trace!("Max packet size set to {peer_max_packet_size}");
146    }
147
148    fn max_packet_size(&self) -> u16 {
149        *self.max_packet_size.lock()
150    }
151
152    fn handle_connect_response(&self, response: ResponsePacket) -> Result<HeaderSet, Error> {
153        let request = OpCode::Connect;
154        let response = response.expect_code(request, ResponseCode::Ok)?;
155
156        // Expect the 4 bytes of additional data. We negotiate the max packet length based on what
157        // the peer requests. See OBEX 1.5 Section 3.4.1.
158        if response.data().len() != request.response_data_length() {
159            return Err(Error::response(request, "Invalid CONNECT data"));
160        }
161        let peer_max_packet_size = u16::from_be_bytes(response.data()[2..4].try_into().unwrap());
162        self.set_max_packet_size(peer_max_packet_size);
163
164        // Check to see if the response headers contains a Connection Identifier. If so, this should
165        // be included in all subsequent operations.
166        let headers: HeaderSet = response.into();
167        if let Some(Header::ConnectionId(id)) = headers.get(&HeaderIdentifier::ConnectionId) {
168            trace!(id:? = id; "Found Connection Identifier in CONNECT response");
169            self.set_connection_status(ConnectionStatus::Connected { id: Some(*id) });
170        }
171        Ok(headers)
172    }
173
174    /// Initiates a CONNECT request to the remote peer.
175    /// Returns the Headers associated with the response on success.
176    /// Returns Error if the CONNECT operation could not be completed.
177    pub async fn connect(&self, headers: HeaderSet) -> Result<HeaderSet, Error> {
178        if self.is_connected() {
179            return Err(Error::operation(OpCode::Connect, "already connected"));
180        }
181
182        let response = {
183            let request = RequestPacket::new_connect(self.max_packet_size(), headers);
184            let mut transport = self.transport.try_new_operation()?;
185            transport.send(request).await?;
186            trace!("Successfully made CONNECT request");
187            transport.receive_response(OpCode::Connect).await?
188        };
189
190        let response_headers = self.handle_connect_response(response)?;
191        Ok(response_headers)
192    }
193
194    /// Initiates a DISCONNECT request to the remote peer.
195    /// Returns the Headers associated with the response on success.
196    /// Returns Error if the DISCONNECT operation couldn't be completed or was rejected by the peer.
197    /// The OBEX Session with the peer is considered terminated, regardless.
198    pub async fn disconnect(self, mut headers: HeaderSet) -> Result<HeaderSet, Error> {
199        let opcode = OpCode::Disconnect;
200        if !self.is_connected() {
201            return Err(Error::operation(opcode, "session not connected"));
202        }
203        headers.try_add_connection_id(&self.connection_id())?;
204        let response = {
205            let request = RequestPacket::new_disconnect(headers);
206            let mut transport = self.transport.try_new_operation()?;
207            trace!("Making outgoing DISCONNECT request: {request:?}");
208            transport.send(request).await?;
209            trace!("Successfully made DISCONNECT request");
210            transport.receive_response(opcode).await?
211        };
212        self.set_connection_status(ConnectionStatus::Disconnected);
213        response.expect_code(opcode, ResponseCode::Ok).map(Into::into)
214    }
215
216    /// Initializes a GET Operation to retrieve data from the remote OBEX Server.
217    /// Returns a `GetOperation` on success, Error if the new operation couldn't be started.
218    pub fn get(&self) -> Result<GetOperation<'_>, Error> {
219        // A GET can only be initiated after the OBEX session is connected.
220        if !self.is_connected() {
221            return Err(Error::operation(OpCode::Get, "session not connected"));
222        }
223
224        let mut headers = HeaderSet::new();
225        headers.try_add_connection_id(&self.connection_id())?;
226
227        // Only one operation can be active at a time.
228        let transport = self.transport.try_new_operation()?;
229        Ok(GetOperation::new(headers, transport))
230    }
231
232    /// Initializes a PUT Operation to write data to the remote OBEX Server.
233    /// Returns a `PutOperation` on success, Error if the new operation couldn't be started.
234    pub fn put(&self) -> Result<PutOperation<'_>, Error> {
235        // A PUT can only be initiated after the OBEX session is connected.
236        if !self.is_connected() {
237            return Err(Error::operation(OpCode::Put, "session not connected"));
238        }
239
240        let mut headers = HeaderSet::new();
241        headers.try_add_connection_id(&self.connection_id())?;
242
243        // Only one operation can be active at a time.
244        let transport = self.transport.try_new_operation()?;
245        Ok(PutOperation::new(headers, transport))
246    }
247
248    /// Initializes a SETPATH Operation to set the current folder on the remote OBEX Server.
249    /// Returns the Headers associated with the response on success.
250    /// Returns `Error::NotImplemented` if the remote server does not support SETPATH.
251    /// Returns Error for all other errors.
252    pub async fn set_path(
253        &self,
254        flags: SetPathFlags,
255        mut headers: HeaderSet,
256    ) -> Result<HeaderSet, Error> {
257        let opcode = OpCode::SetPath;
258        // A SETPATH can only be initiated after the OBEX session is connected.
259        if !self.is_connected() {
260            return Err(Error::operation(opcode, "session not connected"));
261        }
262        headers.try_add_connection_id(&self.connection_id())?;
263        let request = RequestPacket::new_set_path(flags, headers)?;
264        let response = {
265            let mut transport = self.transport.try_new_operation()?;
266            trace!("Making outgoing SETPATH request: {request:?}");
267            transport.send(request).await?;
268            trace!("Successfully made SETPATH request");
269            transport.receive_response(opcode).await?
270        };
271
272        // Per OBEX Section 3.4.6, the server may respond with BadRequest or Forbidden if it does
273        // not support the operation.
274        if *response.code() == ResponseCode::BadRequest
275            || *response.code() == ResponseCode::Forbidden
276        {
277            return Err(Error::not_implemented(opcode));
278        }
279        response.expect_code(opcode, ResponseCode::Ok).map(Into::into)
280    }
281}
282
283#[cfg(test)]
284mod tests {
285    use super::*;
286    use bt_channel_test_support::{Transport, create_test_channels};
287    use test_case::test_case;
288
289    use assert_matches::assert_matches;
290    use async_utils::PollExt;
291    use fuchsia_async as fasync;
292    use std::pin::pin;
293
294    use crate::transport::test_utils::{expect_code, expect_request_and_reply};
295
296    #[fuchsia::test]
297    fn max_packet_size_calculation() {
298        // Value in [255, 65535] should be kept.
299        let transport_max = 1000;
300        assert_eq!(max_packet_size_from_transport(transport_max), 1000);
301
302        // Lower bound should be enforced.
303        let transport_max_small = 40;
304        assert_eq!(max_packet_size_from_transport(transport_max_small), 255);
305
306        // Upper bound should be enforced.
307        let transport_max_large = 700000;
308        assert_eq!(max_packet_size_from_transport(transport_max_large), std::u16::MAX);
309    }
310
311    /// Returns a new ObexClient and the remote end of the transport.
312    /// If `connected` is set, returns an ObexClient in the connected state, indicating the
313    /// completion of the OBEX CONNECT procedure.
314    fn new_obex_client(connected: ConnectionStatus, transport: Transport) -> (ObexClient, Channel) {
315        let (local, remote) = create_test_channels(transport);
316        let client = ObexClient::new(local, TransportType::Rfcomm);
317        client.set_connection_status(connected);
318        (client, remote)
319    }
320
321    #[test_case(Transport::Socket ; "socket")]
322    #[test_case(Transport::Fidl ; "fidl")]
323    #[fuchsia::test]
324    fn client_connect_success(transport: Transport) {
325        let mut exec = fasync::TestExecutor::new();
326        let (client, mut remote) = new_obex_client(ConnectionStatus::default(), transport);
327
328        assert!(!client.is_connected());
329        assert_eq!(client.max_packet_size(), Channel::DEFAULT_MAX_TX as u16);
330        assert_eq!(client.connection_id(), None);
331
332        {
333            let connect_fut = client.connect(HeaderSet::new());
334            let mut connect_fut = pin!(connect_fut);
335            exec.run_until_stalled(&mut connect_fut).expect_pending("waiting for response");
336
337            // Expect the Connect request on the remote and reply positively.
338            let response_headers =
339                HeaderSet::from_headers(vec![Header::ConnectionId(ConnectionIdentifier(1))])
340                    .unwrap();
341            let response = ResponsePacket::new(
342                ResponseCode::Ok,
343                vec![0x10, 0x00, 0xff, 0xff], // Version = 1.0, Flags = 0, Max packet = 0xffff
344                response_headers.clone(),
345            );
346            expect_request_and_reply(
347                &mut exec,
348                &mut remote,
349                expect_code(OpCode::Connect),
350                response,
351            );
352
353            let connect_result = exec
354                .run_until_stalled(&mut connect_fut)
355                .expect("received response")
356                .expect("response is ok");
357            assert_eq!(connect_result, response_headers);
358        }
359
360        // Should be connected with the max packet size specified by the peer.
361        assert!(client.is_connected());
362        assert_eq!(client.max_packet_size(), 0xffff);
363        assert_eq!(client.connection_id(), Some(ConnectionIdentifier(1)));
364    }
365
366    #[test_case(Transport::Socket ; "socket")]
367    #[test_case(Transport::Fidl ; "fidl")]
368    #[fuchsia::test]
369    async fn multiple_connect_is_error(transport: Transport) {
370        let (client, _remote) = new_obex_client(ConnectionStatus::connected_no_id(), transport);
371
372        // Trying to connect again is an Error since it can only be done once.
373        let result = client.connect(HeaderSet::new()).await;
374        assert_matches!(result, Err(Error::OperationError { .. }));
375    }
376
377    #[test_case(Transport::Socket ; "socket")]
378    #[test_case(Transport::Fidl ; "fidl")]
379    #[fuchsia::test]
380    fn get_before_connect_is_error(transport: Transport) {
381        let _exec = fasync::TestExecutor::new();
382        let (client, _remote) = new_obex_client(ConnectionStatus::default(), transport);
383
384        let get_result = client.get();
385        assert_matches!(get_result, Err(Error::OperationError { .. }));
386    }
387
388    #[test_case(Transport::Socket ; "socket")]
389    #[test_case(Transport::Fidl ; "fidl")]
390    #[fuchsia::test]
391    fn sequential_get_operations_is_ok(transport: Transport) {
392        let _exec = fasync::TestExecutor::new();
393        let (client, _remote) = new_obex_client(ConnectionStatus::connected_no_id(), transport);
394
395        // Creating the first GET operation should succeed.
396        let _get_operation1 = client.get().expect("can initialize first get");
397
398        // After the first one "completes" (e.g. no longer held), it's okay to initiate a GET.
399        drop(_get_operation1);
400        let _get_operation2 = client.get().expect("can initialize second get");
401    }
402
403    #[test_case(Transport::Socket ; "socket")]
404    #[test_case(Transport::Fidl ; "fidl")]
405    #[fuchsia::test]
406    fn disconnect_success(transport: Transport) {
407        let mut exec = fasync::TestExecutor::new();
408        let (client, mut remote) = new_obex_client(ConnectionStatus::connected_no_id(), transport);
409
410        let headers = HeaderSet::from_header(Header::Description("finished".into()));
411        let disconnect_fut = client.disconnect(headers);
412        let mut disconnect_fut = pin!(disconnect_fut);
413        exec.run_until_stalled(&mut disconnect_fut).expect_pending("waiting for response");
414
415        // Expect the Disconnect request on the remote. The typical response is a positive `Ok`.
416        let response_headers = HeaderSet::from_header(Header::Description("accepted".into()));
417        let response = ResponsePacket::new(ResponseCode::Ok, vec![], response_headers.clone());
418        expect_request_and_reply(&mut exec, &mut remote, expect_code(OpCode::Disconnect), response);
419
420        let disconnect_result = exec
421            .run_until_stalled(&mut disconnect_fut)
422            .expect("received response")
423            .expect("response is ok");
424        assert_eq!(disconnect_result, response_headers);
425    }
426
427    #[test_case(Transport::Socket ; "socket")]
428    #[test_case(Transport::Fidl ; "fidl")]
429    #[fuchsia::test]
430    async fn disconnect_before_connect_error(transport: Transport) {
431        let (client, _remote) = new_obex_client(ConnectionStatus::default(), transport);
432
433        let headers = HeaderSet::from_header(Header::Description("finished".into()));
434        let disconnect_result = client.disconnect(headers).await;
435        assert_matches!(disconnect_result, Err(Error::OperationError { .. }))
436    }
437
438    #[test_case(Transport::Socket ; "socket")]
439    #[test_case(Transport::Fidl ; "fidl")]
440    #[fuchsia::test]
441    fn disconnect_error_response_error(transport: Transport) {
442        let mut exec = fasync::TestExecutor::new();
443        let (client, mut remote) = new_obex_client(ConnectionStatus::connected_no_id(), transport);
444
445        let disconnect_fut = client.disconnect(HeaderSet::new());
446        let mut disconnect_fut = pin!(disconnect_fut);
447        exec.run_until_stalled(&mut disconnect_fut).expect_pending("waiting for response");
448
449        // Expect the Disconnect request on the remote. An Error response still results in
450        // disconnection.
451        let response_headers = HeaderSet::from_header(Header::Description("accepted".into()));
452        let response = ResponsePacket::new(
453            ResponseCode::InternalServerError,
454            vec![],
455            response_headers.clone(),
456        );
457        expect_request_and_reply(&mut exec, &mut remote, expect_code(OpCode::Disconnect), response);
458
459        let disconnect_result =
460            exec.run_until_stalled(&mut disconnect_fut).expect("received response");
461        assert_matches!(disconnect_result, Err(Error::PeerRejected { .. }));
462    }
463
464    #[test_case(Transport::Socket ; "socket")]
465    #[test_case(Transport::Fidl ; "fidl")]
466    #[fuchsia::test]
467    fn setpath_success(transport: Transport) {
468        let mut exec = fasync::TestExecutor::new();
469        let (client, mut remote) = new_obex_client(ConnectionStatus::connected_no_id(), transport);
470
471        let headers = HeaderSet::from_header(Header::name("myfolder"));
472        let setpath_fut = client.set_path(SetPathFlags::empty(), headers);
473        let mut setpath_fut = pin!(setpath_fut);
474        exec.run_until_stalled(&mut setpath_fut).expect_pending("waiting for response");
475
476        // Expect the SetPath request on the remote. The typical response is a positive `Ok`.
477        let response_headers =
478            HeaderSet::from_header(Header::Description("updated current folder".into()));
479        let response = ResponsePacket::new(ResponseCode::Ok, vec![], response_headers.clone());
480        let expectation = |request: RequestPacket| {
481            assert_eq!(*request.code(), OpCode::SetPath);
482            assert_eq!(request.data(), &[0, 0]);
483        };
484        expect_request_and_reply(&mut exec, &mut remote, expectation, response);
485
486        let setpath_result = exec
487            .run_until_stalled(&mut setpath_fut)
488            .expect("received response")
489            .expect("response is ok");
490        assert_eq!(setpath_result, response_headers);
491    }
492
493    #[test_case(Transport::Socket ; "socket")]
494    #[test_case(Transport::Fidl ; "fidl")]
495    #[fuchsia::test]
496    fn setpath_error_response_is_error(transport: Transport) {
497        let mut exec = fasync::TestExecutor::new();
498        let (client, mut remote) = new_obex_client(ConnectionStatus::connected_no_id(), transport);
499
500        // Peer doesn't support SetPath.
501        {
502            let setpath_fut = client.set_path(SetPathFlags::BACKUP, HeaderSet::new());
503            let mut setpath_fut = pin!(setpath_fut);
504            exec.run_until_stalled(&mut setpath_fut).expect_pending("waiting for response");
505
506            // Expect the SetPath request on the remote - peer doesn't support SetPath.
507            let response_headers =
508                HeaderSet::from_header(Header::Description("not implemented".into()));
509            let response = ResponsePacket::new(ResponseCode::BadRequest, vec![], response_headers);
510            expect_request_and_reply(
511                &mut exec,
512                &mut remote,
513                expect_code(OpCode::SetPath),
514                response,
515            );
516
517            let setpath_result =
518                exec.run_until_stalled(&mut setpath_fut).expect("received response");
519            assert_matches!(setpath_result, Err(Error::NotImplemented { .. }));
520        }
521
522        // Peer rejects SetPath.
523        let headers = HeaderSet::from_header(Header::name("file"));
524        let setpath_fut = client.set_path(SetPathFlags::DONT_CREATE, headers);
525        let mut setpath_fut = pin!(setpath_fut);
526        exec.run_until_stalled(&mut setpath_fut).expect_pending("waiting for response");
527
528        // Expect the SetPath request on the remote - peer responds with error.
529        let response_headers =
530            HeaderSet::from_header(Header::Description("not implemented".into()));
531        let response =
532            ResponsePacket::new(ResponseCode::InternalServerError, vec![], response_headers);
533        expect_request_and_reply(&mut exec, &mut remote, expect_code(OpCode::SetPath), response);
534
535        let setpath_result = exec.run_until_stalled(&mut setpath_fut).expect("received response");
536        assert_matches!(setpath_result, Err(Error::PeerRejected { .. }));
537    }
538}
bt_obex/client/mod.rs

bt_obex/client/
mod.rs
