starnix_core/fs/fuchsia/
remote_unix_domain_socket.rs

1// Copyright 2024 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 crate::fs::fuchsia::{OpenFlags, new_remote_file};
6use crate::task::{
7    CurrentTask, EventHandler, SignalHandler, SignalHandlerInner, WaitCanceler, Waiter,
8};
9use crate::vfs::buffers::{InputBuffer, OutputBuffer};
10use crate::vfs::socket::{
11    SockOptValue, Socket, SocketAddress, SocketDomain, SocketHandle, SocketMessageFlags, SocketOps,
12    SocketPeer, SocketProtocol, SocketShutdownFlags, SocketType,
13};
14use crate::vfs::{AncillaryData, FileHandle, MessageReadInfo, UnixControlData};
15use fidl::endpoints::SynchronousProxy;
16use fidl_fuchsia_io as fio;
17use fidl_fuchsia_starnix_binder as fbinder;
18use linux_uapi::{SO_LINGER, SOL_SOCKET};
19use starnix_uapi::auth::Credentials;
20use starnix_uapi::errors::Errno;
21use starnix_uapi::vfs::FdEvents;
22use starnix_uapi::{errno, error, from_status_like_fdio, uapi, ucred};
23use std::sync::Arc;
24use zerocopy::IntoBytes;
25static READABLE_SIGNAL: zx::Signals =
26    zx::Signals::from_bits_retain(fio::FileSignal::READABLE.bits());
27static WRITABLE_SIGNAL: zx::Signals =
28    zx::Signals::from_bits_retain(fio::FileSignal::WRITABLE.bits());
29
30pub struct RemoteUnixDomainSocket {
31    client: fbinder::UnixDomainSocketSynchronousProxy,
32    event: zx::EventPair,
33    remote_creds: Arc<Credentials>,
34}
35
36impl RemoteUnixDomainSocket {
37    pub fn new(channel: zx::Channel, remote_creds: Arc<Credentials>) -> Result<Self, Errno> {
38        let client = fbinder::UnixDomainSocketSynchronousProxy::from_channel(channel);
39        let response = client
40            .get_event(
41                &fbinder::UnixDomainSocketGetEventRequest::default(),
42                zx::MonotonicInstant::INFINITE,
43            )
44            .map_err(|_| errno!(ECONNREFUSED))?
45            .map_err(|e: i32| from_status_like_fdio!(zx::Status::err_from_raw(e)))?;
46        let event = response.event.ok_or_else(|| errno!(ECONNREFUSED))?;
47        Ok(Self { client, event, remote_creds })
48    }
49
50    fn get_signals_from_events(events: FdEvents) -> zx::Signals {
51        let mut signals = zx::Signals::NONE;
52        if events.contains(FdEvents::POLLIN) {
53            signals |= READABLE_SIGNAL;
54        }
55        if events.contains(FdEvents::POLLOUT) {
56            signals |= WRITABLE_SIGNAL;
57        }
58        signals
59    }
60
61    fn get_events_from_signals(signals: zx::Signals) -> FdEvents {
62        let mut events = FdEvents::empty();
63        if signals.contains(READABLE_SIGNAL) {
64            events |= FdEvents::POLLIN;
65        }
66        if signals.contains(WRITABLE_SIGNAL) {
67            events |= FdEvents::POLLOUT;
68        }
69        events
70    }
71
72    /// Perform an action using the credentials of the remote task.
73    fn with_remote_creds<F, R>(&self, current_task: &CurrentTask, f: F) -> Result<R, Errno>
74    where
75        F: FnOnce() -> Result<R, Errno>,
76    {
77        current_task.override_creds(self.remote_creds.clone(), f)
78    }
79}
80
81impl SocketOps for RemoteUnixDomainSocket {
82    fn get_socket_info(&self) -> Result<(SocketDomain, SocketType, SocketProtocol), Errno> {
83        Ok((SocketDomain::Unix, SocketType::Datagram, SocketProtocol::from_raw(0)))
84    }
85
86    fn connect(
87        &self,
88        _socket: &SocketHandle,
89        _current_task: &CurrentTask,
90        _peer: SocketPeer,
91    ) -> Result<(), Errno> {
92        error!(EISCONN)
93    }
94
95    fn listen(&self, _socket: &Socket, _backlog: i32, _credentials: ucred) -> Result<(), Errno> {
96        error!(EOPNOTSUPP)
97    }
98
99    fn accept(&self, _socket: &Socket, _current_task: &CurrentTask) -> Result<SocketHandle, Errno> {
100        error!(EOPNOTSUPP)
101    }
102
103    fn bind(
104        &self,
105        _socket: &Socket,
106        _current_task: &CurrentTask,
107        _socket_address: SocketAddress,
108    ) -> Result<(), Errno> {
109        error!(EOPNOTSUPP)
110    }
111
112    fn read(
113        &self,
114        _socket: &Socket,
115        current_task: &CurrentTask,
116        data: &mut dyn OutputBuffer,
117        flags: SocketMessageFlags,
118    ) -> Result<MessageReadInfo, Errno> {
119        if self.client.is_closed().map_err(|_| errno!(ECONNREFUSED))? {
120            return error!(ECONNREFUSED);
121        }
122        let mut read_flags = fbinder::ReadFlags::empty();
123        if flags.contains(SocketMessageFlags::PEEK) {
124            read_flags |= fbinder::ReadFlags::PEEK;
125        }
126
127        let response = self
128            .client
129            .read(
130                &fbinder::UnixDomainSocketReadRequest {
131                    count: Some(data.available() as u64),
132                    flags: Some(read_flags),
133                    ..Default::default()
134                },
135                zx::MonotonicInstant::INFINITE,
136            )
137            .map_err(|_| errno!(ECONNREFUSED))?
138            .map_err(|e: i32| {
139                let status = zx::Status::err_from_raw(e);
140                if status == zx::Status::PEER_CLOSED {
141                    errno!(ECONNRESET)
142                } else {
143                    from_status_like_fdio!(status)
144                }
145            })?;
146
147        let written =
148            if let Some(received_data) = response.data { data.write(&received_data)? } else { 0 };
149
150        let mut file_handles: Vec<FileHandle> = vec![];
151        if let Some(handles) = response.handles {
152            // Use the remote task's credentials to create the remote_file object. This ensures
153            // that the SID associated to the fd is set to the correct value.
154            self.with_remote_creds(current_task, || {
155                for handle in handles {
156                    file_handles.push(new_remote_file(current_task, handle, OpenFlags::RDWR)?);
157                }
158                Ok(())
159            })?;
160        }
161        let ancillary_data = vec![AncillaryData::Unix(UnixControlData::Rights(file_handles))];
162
163        let message_length = response.data_original_length.unwrap_or(written as u64) as usize;
164
165        Ok(MessageReadInfo {
166            bytes_read: written,
167            message_length,
168            address: None,
169            credentials: None,
170            ancillary_data,
171        })
172    }
173
174    fn write(
175        &self,
176        _socket: &Socket,
177        current_task: &CurrentTask,
178        data: &mut dyn InputBuffer,
179        _dest_address: &mut Option<SocketAddress>,
180        ancillary_data: &mut Vec<AncillaryData>,
181    ) -> Result<usize, Errno> {
182        if self.client.is_closed().map_err(|_| errno!(ECONNREFUSED))? {
183            return error!(ECONNREFUSED);
184        }
185
186        let mut handles: Vec<zx::NullableHandle> = vec![];
187        for data in ancillary_data {
188            match data {
189                AncillaryData::Unix(UnixControlData::Rights(file_handles)) => {
190                    // Access the served files with the credentials of the remote end.
191                    self.with_remote_creds(current_task, || {
192                        for file_handle in file_handles {
193                            let Some(handle) = file_handle.to_handle(current_task)? else {
194                                return error!(EINVAL);
195                            };
196                            handles.push(handle);
197                        }
198                        Ok(())
199                    })?;
200                }
201                _ => return error!(EINVAL),
202            }
203        }
204
205        let bytes = data.read_all()?;
206
207        let response = self
208            .client
209            .write(
210                fbinder::UnixDomainSocketWriteRequest {
211                    data: Some(bytes),
212                    handles: Some(handles),
213                    ..Default::default()
214                },
215                zx::MonotonicInstant::INFINITE,
216            )
217            .map_err(|_| errno!(ECONNREFUSED))?
218            .map_err(|e: i32| from_status_like_fdio!(zx::Status::err_from_raw(e)))?;
219
220        let written = response.actual_count.unwrap_or(0);
221        Ok(written as usize)
222    }
223
224    fn wait_async(
225        &self,
226        _socket: &Socket,
227        _current_task: &CurrentTask,
228        waiter: &Waiter,
229        events: FdEvents,
230        handler: EventHandler,
231    ) -> WaitCanceler {
232        let signal_handler = SignalHandler {
233            inner: SignalHandlerInner::ZxHandle(Self::get_events_from_signals),
234            event_handler: handler,
235            err_code: None,
236        };
237        let canceler = waiter
238            .wake_on_zircon_signals(
239                &self.event,
240                Self::get_signals_from_events(events),
241                signal_handler,
242            )
243            .unwrap();
244        WaitCanceler::new_port(canceler)
245    }
246
247    fn query_events(
248        &self,
249        _socket: &Socket,
250        _current_task: &CurrentTask,
251    ) -> Result<FdEvents, Errno> {
252        let signals = self
253            .event
254            .as_handle_ref()
255            .wait_one(zx::Signals::NONE, zx::MonotonicInstant::INFINITE_PAST)
256            .map_err(|e| from_status_like_fdio!(e))?;
257        Ok(Self::get_events_from_signals(signals))
258    }
259
260    fn shutdown(&self, _socket: &Socket, _how: SocketShutdownFlags) -> Result<(), Errno> {
261        Ok(())
262    }
263
264    fn close(&self, _current_task: &CurrentTask, _socket: &Socket) {
265        let _ = self.client.close(zx::MonotonicInstant::INFINITE);
266    }
267
268    fn getsockname(&self, _socket: &Socket) -> Result<SocketAddress, Errno> {
269        Ok(SocketAddress::default_for_domain(SocketDomain::Unix))
270    }
271
272    fn getpeername(&self, socket: &Socket) -> Result<SocketAddress, Errno> {
273        self.getsockname(socket)
274    }
275
276    fn setsockopt(
277        &self,
278        _socket: &Socket,
279        _current_task: &CurrentTask,
280        _level: u32,
281        _optname: u32,
282        _optval: SockOptValue,
283    ) -> Result<(), Errno> {
284        error!(EOPNOTSUPP)
285    }
286
287    fn getsockopt(
288        &self,
289        _socket: &Socket,
290        _current_task: &CurrentTask,
291        level: u32,
292        optname: u32,
293        _optlen: u32,
294    ) -> Result<Vec<u8>, Errno> {
295        if level != SOL_SOCKET {
296            return error!(EINVAL);
297        }
298        let data = match optname {
299            SO_LINGER => uapi::linger::default().as_bytes().to_vec(),
300            _ => return error!(EINVAL),
301        };
302
303        Ok(data)
304    }
305
306    fn to_handle(
307        &self,
308        _socket: &Socket,
309        _current_task: &CurrentTask,
310    ) -> Result<Option<zx::NullableHandle>, Errno> {
311        let (proxy, server) = zx::Channel::create();
312        self.client.clone(server.into()).map_err(|_| errno!(ECONNREFUSED))?;
313        Ok(Some(zx::NullableHandle::from(proxy).into()))
314    }
315}
316
317#[cfg(test)]
318mod tests {
319    use super::*;
320    use crate::testing::spawn_kernel_and_run;
321    use crate::vfs::socket::SocketFile;
322    use crate::vfs::{VecInputBuffer, VecOutputBuffer};
323    use fidl::endpoints::{DiscoverableProtocolMarker as _, RequestStream};
324    use fidl_fuchsia_unknown as funknown;
325    use fuchsia_async as fasync;
326    use futures::StreamExt;
327    use starnix_sync::{LockDepMutex, RemoteUnixDomainSocketStateLock};
328    use std::sync::Arc;
329
330    #[derive(Debug)]
331    struct Data {
332        bytes: Vec<u8>,
333        handles: Vec<zx::NullableHandle>,
334    }
335
336    impl Data {
337        fn try_clone(&mut self) -> Result<Self, zx::Status> {
338            let mut new_handles = vec![];
339            for handle in std::mem::take(&mut self.handles) {
340                let (new_handle, old_handle) = {
341                    let handle_type = handle.basic_info()?.object_type;
342                    match handle_type {
343                        zx::ObjectType::CHANNEL => {
344                            let channel = zx::Channel::from(handle);
345                            let client = funknown::CloneableSynchronousProxy::new(channel);
346                            let (proxy, server) = zx::Channel::create();
347                            let new_handle = client
348                                .clone(server.into())
349                                .map(|_| proxy.into())
350                                .map_err(|_| zx::Status::NOT_SUPPORTED);
351                            (new_handle, client.into_channel().into_handle())
352                        }
353                        _ => {
354                            let new_handle = handle.duplicate_handle(zx::Rights::SAME_RIGHTS);
355                            (new_handle, handle)
356                        }
357                    }
358                };
359                self.handles.push(old_handle);
360                new_handles.push(new_handle);
361            }
362            let new_handles =
363                new_handles.into_iter().collect::<Result<Vec<zx::NullableHandle>, zx::Status>>()?;
364            Ok(Self { bytes: self.bytes.clone(), handles: new_handles })
365        }
366    }
367
368    #[derive(Debug)]
369    struct UnixDomainSocketImplState {
370        _local_event: zx::EventPair,
371        remote_event: zx::EventPair,
372        buffer: Vec<Data>,
373    }
374
375    impl Default for UnixDomainSocketImplState {
376        fn default() -> Self {
377            let (_local_event, remote_event) = zx::EventPair::create();
378            Self { _local_event, remote_event, buffer: vec![] }
379        }
380    }
381
382    #[derive(Debug, Default)]
383    struct UnixDomainSocketImpl {
384        state: LockDepMutex<UnixDomainSocketImplState, RemoteUnixDomainSocketStateLock>,
385        close_on_read: bool,
386    }
387
388    impl UnixDomainSocketImpl {
389        fn read(
390            &self,
391            payload: fbinder::UnixDomainSocketReadRequest,
392        ) -> Result<fbinder::UnixDomainSocketReadResponse, zx::Status> {
393            if self.close_on_read {
394                return Err(zx::Status::PEER_CLOSED);
395            }
396            let Some(count) = payload.count else {
397                return Err(zx::Status::INVALID_ARGS);
398            };
399            let Some(flags) = payload.flags else {
400                return Err(zx::Status::INVALID_ARGS);
401            };
402            let mut state = self.state.lock();
403            if state.buffer.is_empty() {
404                return Err(zx::Status::SHOULD_WAIT);
405            }
406            let mut data = if flags.contains(fbinder::ReadFlags::PEEK) {
407                state.buffer[0].try_clone()?
408            } else {
409                state.buffer.remove(0)
410            };
411
412            if state.buffer.is_empty() {
413                state.remote_event.as_handle_ref().signal(READABLE_SIGNAL, WRITABLE_SIGNAL)?;
414            }
415
416            let actual_count = data.bytes.len() as u64;
417            data.bytes.truncate(count as usize);
418
419            Ok(fbinder::UnixDomainSocketReadResponse {
420                data: Some(data.bytes),
421                data_original_length: Some(actual_count),
422                handles: Some(data.handles),
423                ..Default::default()
424            })
425        }
426
427        fn write(
428            &self,
429            payload: fbinder::UnixDomainSocketWriteRequest,
430        ) -> Result<fbinder::UnixDomainSocketWriteResponse, zx::Status> {
431            let Some(bytes) = payload.data else {
432                return Err(zx::Status::INVALID_ARGS);
433            };
434            let actual_count = bytes.len() as u64;
435            let Some(handles) = payload.handles else {
436                return Err(zx::Status::INVALID_ARGS);
437            };
438            let mut state = self.state.lock();
439            state.buffer.push(Data { bytes, handles });
440            state
441                .remote_event
442                .as_handle_ref()
443                .signal(zx::Signals::NONE, READABLE_SIGNAL | WRITABLE_SIGNAL)?;
444            Ok(fbinder::UnixDomainSocketWriteResponse {
445                actual_count: Some(actual_count),
446                ..Default::default()
447            })
448        }
449
450        async fn serve(self: &Arc<Self>, channel: zx::Channel) {
451            let stream = fbinder::UnixDomainSocketRequestStream::from_channel(
452                fasync::Channel::from_channel(channel),
453            );
454            stream
455                .for_each_concurrent(None, |message| async {
456                    match message {
457                        Ok(fbinder::UnixDomainSocketRequest::GetEvent { responder, .. }) => {
458                            let state = self.state.lock();
459                            let event = state
460                                .remote_event
461                                .duplicate_handle(zx::Rights::SAME_RIGHTS)
462                                .expect("duplicate event");
463                            responder
464                                .send(Ok(fbinder::UnixDomainSocketGetEventResponse {
465                                    event: Some(event),
466                                    ..Default::default()
467                                }))
468                                .expect("respond");
469                        }
470                        Ok(fbinder::UnixDomainSocketRequest::Read {
471                            payload, responder, ..
472                        }) => {
473                            assert!(
474                                responder
475                                    .send(self.read(payload).map_err(|e| e.into_raw()))
476                                    .is_ok()
477                            );
478                        }
479                        Ok(fbinder::UnixDomainSocketRequest::Write {
480                            payload, responder, ..
481                        }) => {
482                            assert!(
483                                responder
484                                    .send(self.write(payload).as_ref().map_err(|e| e.into_raw()))
485                                    .is_ok()
486                            );
487                        }
488                        Ok(fbinder::UnixDomainSocketRequest::Query { responder }) => {
489                            assert!(
490                                responder
491                                    .send(fbinder::UnixDomainSocketMarker::PROTOCOL_NAME.as_bytes())
492                                    .is_ok()
493                            );
494                        }
495                        Ok(fbinder::UnixDomainSocketRequest::Clone { request, .. }) => {
496                            self.serve(request.into()).await;
497                        }
498                        Ok(fbinder::UnixDomainSocketRequest::Close { responder }) => {
499                            assert!(responder.send(Ok(())).is_ok());
500                        }
501                        _ => {
502                            return;
503                        }
504                    }
505                })
506                .await;
507        }
508    }
509
510    #[::fuchsia::test]
511    async fn test_remote_uds() {
512        let (client, server) = zx::Channel::create();
513        let handle = std::thread::spawn(|| {
514            let mut executor = fasync::LocalExecutor::default();
515            executor.run_singlethreaded(async move {
516                let uds_impl = UnixDomainSocketImpl::default();
517                Arc::new(uds_impl).serve(server).await;
518            });
519        });
520        spawn_kernel_and_run(async move |current_task| {
521            let original_file = new_remote_file(current_task, client.into(), OpenFlags::RDWR)
522                .expect("new_remote_file");
523            assert!(original_file.node().is_sock());
524            let file = new_remote_file(
525                current_task,
526                original_file.to_handle(current_task).expect("to_handle").expect("has_handle"),
527                OpenFlags::RDWR,
528            )
529            .expect("new_remote_file");
530            let ancillary_data =
531                vec![AncillaryData::Unix(UnixControlData::Rights(vec![original_file]))];
532            let socket_ops = file.downcast_file::<SocketFile>().unwrap();
533            let data = "HelloWorld";
534            let mut input_buffer = VecInputBuffer::new(data.as_bytes());
535            assert_eq!(
536                socket_ops.sendmsg(
537                    current_task,
538                    &file,
539                    &mut input_buffer,
540                    None,
541                    ancillary_data,
542                    SocketMessageFlags::empty()
543                ),
544                Ok(data.len())
545            );
546
547            let flags = SocketMessageFlags::CTRUNC
548                | SocketMessageFlags::TRUNC
549                | SocketMessageFlags::NOSIGNAL
550                | SocketMessageFlags::CMSG_CLOEXEC;
551
552            let mut buffer = VecOutputBuffer::new(1024);
553            let info = socket_ops
554                .recvmsg(&current_task, &file, &mut buffer, flags | SocketMessageFlags::PEEK, None)
555                .expect("recvmsg");
556
557            assert_eq!(info.ancillary_data.len(), 1);
558            assert_eq!(info.message_length, data.len());
559
560            let mut buffer = VecOutputBuffer::new(1024);
561            let info = socket_ops
562                .recvmsg(&current_task, &file, &mut buffer, flags, None)
563                .expect("recvmsg");
564
565            assert_eq!(info.ancillary_data.len(), 1);
566            assert_eq!(info.message_length, data.len());
567
568            let mut buffer = VecOutputBuffer::new(1024);
569            let err = socket_ops
570                .recvmsg(
571                    &current_task,
572                    &file,
573                    &mut buffer,
574                    flags | SocketMessageFlags::DONTWAIT,
575                    None,
576                )
577                .unwrap_err();
578            assert_eq!(err, errno!(EAGAIN));
579        })
580        .await;
581        handle.join().expect("join");
582    }
583
584    #[::fuchsia::test]
585    async fn test_remote_uds_peer_closed() {
586        let (client, server) = zx::Channel::create();
587        let handle = std::thread::spawn(move || {
588            let mut executor = fasync::LocalExecutor::default();
589            executor.run_singlethreaded(async move {
590                let uds_impl = UnixDomainSocketImpl { close_on_read: true, ..Default::default() };
591                Arc::new(uds_impl).serve(server).await;
592            });
593        });
594        spawn_kernel_and_run(async move |current_task| {
595            let file = new_remote_file(current_task, client.into(), OpenFlags::RDWR)
596                .expect("new_remote_file");
597            let socket_ops = file.downcast_file::<SocketFile>().unwrap();
598
599            let mut buffer = VecOutputBuffer::new(1024);
600            let err = socket_ops
601                .recvmsg(&current_task, &file, &mut buffer, SocketMessageFlags::empty(), None)
602                .unwrap_err();
603            assert_eq!(err, errno!(ECONNRESET));
604        })
605        .await;
606        handle.join().expect("join");
607    }
608}
starnix_core/fs/fuchsia/remote_unix_domain_socket.rs

starnix_core/fs/fuchsia/
remote_unix_domain_socket.rs
