starnix_core/vfs/socket/
socket_vsock.rs

1// Copyright 2022 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::task::{CurrentTask, EventHandler, WaitCanceler, WaitQueue, Waiter};
6use crate::vfs::FileHandle;
7use crate::vfs::buffers::{AncillaryData, InputBuffer, MessageReadInfo, OutputBuffer};
8use crate::vfs::socket::{
9    AcceptQueue, DEFAULT_LISTEN_BACKLOG, Socket, SocketAddress, SocketDomain, SocketHandle,
10    SocketMessageFlags, SocketOps, SocketPeer, SocketProtocol, SocketShutdownFlags, SocketType,
11};
12use starnix_sync::{
13    FileOpsCore, LockDepGuard, LockDepMutex, LockEqualOrBefore, Locked, VsockSocketInnerLock,
14    allow_subclass,
15};
16use starnix_uapi::auth::Credentials;
17use starnix_uapi::errors::Errno;
18use starnix_uapi::open_flags::OpenFlags;
19use starnix_uapi::vfs::FdEvents;
20use starnix_uapi::{errno, error, ucred};
21
22// An implementation of AF_VSOCK.
23// See https://man7.org/linux/man-pages/man7/vsock.7.html
24
25pub struct VsockSocket {
26    inner: LockDepMutex<VsockSocketInner, VsockSocketInnerLock>,
27}
28
29struct VsockSocketInner {
30    /// The address that this socket has been bound to, if it has been bound.
31    address: Option<SocketAddress>,
32
33    // WaitQueue for listening sockets.
34    waiters: WaitQueue,
35
36    // state of the vsock. Contains a handle to a ZxioBackedSocket when connected.
37    state: VsockSocketState,
38}
39
40enum VsockSocketState {
41    /// The socket has not been connected.
42    Disconnected,
43
44    /// The socket has had `listen` called and can accept incoming connections.
45    Listening(AcceptQueue),
46
47    /// The socket is connected to a ZxioBackedSocket.
48    Connected { file: FileHandle, peer_addr: SocketAddress },
49
50    /// The socket is closed.
51    Closed,
52}
53
54fn downcast_socket_to_vsock(socket: &Socket) -> &VsockSocket {
55    // It is a programing error if we are downcasting
56    // a different type of socket as sockets from different families
57    // should not communicate, so unwrapping here
58    // will let us know that.
59    socket.downcast_socket::<VsockSocket>().unwrap()
60}
61
62impl VsockSocket {
63    pub fn new(_socket_type: SocketType) -> VsockSocket {
64        VsockSocket {
65            inner: VsockSocketInner {
66                address: None,
67                waiters: WaitQueue::default(),
68                state: VsockSocketState::Disconnected,
69            }
70            .into(),
71        }
72    }
73
74    /// Locks and returns the inner state of the Socket.
75    fn lock(&self) -> LockDepGuard<'_, VsockSocketInner> {
76        self.inner.lock()
77    }
78}
79
80impl SocketOps for VsockSocket {
81    // Connect with Vsock sockets is not allowed as
82    // we only connect from the enclosing OK.
83    fn connect(
84        &self,
85        _locked: &mut Locked<FileOpsCore>,
86        _socket: &SocketHandle,
87        _current_task: &CurrentTask,
88        _peer: SocketPeer,
89    ) -> Result<(), Errno> {
90        error!(EPROTOTYPE)
91    }
92
93    fn listen(
94        &self,
95        _locked: &mut Locked<FileOpsCore>,
96        _socket: &Socket,
97        backlog: i32,
98        _credentials: ucred,
99    ) -> Result<(), Errno> {
100        let mut inner = self.lock();
101        let is_bound = inner.address.is_some();
102        let backlog = if backlog < 0 { DEFAULT_LISTEN_BACKLOG } else { backlog as usize };
103        match &mut inner.state {
104            VsockSocketState::Disconnected if is_bound => {
105                inner.state = VsockSocketState::Listening(AcceptQueue::new(backlog));
106                Ok(())
107            }
108            VsockSocketState::Listening(queue) => {
109                queue.set_backlog(backlog)?;
110                Ok(())
111            }
112            _ => error!(EINVAL),
113        }
114    }
115
116    fn accept(
117        &self,
118        _locked: &mut Locked<FileOpsCore>,
119        socket: &Socket,
120        _current_task: &CurrentTask,
121    ) -> Result<SocketHandle, Errno> {
122        match socket.socket_type {
123            SocketType::Stream | SocketType::SeqPacket => {}
124            _ => return error!(EOPNOTSUPP),
125        }
126        let mut inner = self.lock();
127        let queue = match &mut inner.state {
128            VsockSocketState::Listening(queue) => queue,
129            _ => return error!(EINVAL),
130        };
131        let socket = queue.sockets.pop_front().ok_or_else(|| errno!(EAGAIN))?;
132        Ok(socket)
133    }
134
135    fn bind(
136        &self,
137        _locked: &mut Locked<FileOpsCore>,
138        _socket: &Socket,
139        _current_task: &CurrentTask,
140        socket_address: SocketAddress,
141    ) -> Result<(), Errno> {
142        match socket_address {
143            SocketAddress::Vsock { .. } => {}
144            _ => return error!(EINVAL),
145        }
146        let mut inner = self.lock();
147        if inner.address.is_some() {
148            return error!(EINVAL);
149        }
150        inner.address = Some(socket_address);
151        Ok(())
152    }
153
154    fn read(
155        &self,
156        locked: &mut Locked<FileOpsCore>,
157        _socket: &Socket,
158        current_task: &CurrentTask,
159        data: &mut dyn OutputBuffer,
160        _flags: SocketMessageFlags,
161    ) -> Result<MessageReadInfo, Errno> {
162        let (address, file) = {
163            let inner = self.lock();
164            let address = inner.address.clone();
165
166            match &inner.state {
167                VsockSocketState::Connected { file, .. } => (address, file.clone()),
168                _ => return error!(EBADF),
169            }
170        };
171        let bytes_read = current_task
172            .override_creds(Credentials::root(), || file.read(locked, current_task, data))?;
173        Ok(MessageReadInfo {
174            bytes_read,
175            message_length: bytes_read,
176            address,
177            ancillary_data: vec![],
178        })
179    }
180
181    fn write(
182        &self,
183        locked: &mut Locked<FileOpsCore>,
184        _socket: &Socket,
185        current_task: &CurrentTask,
186        data: &mut dyn InputBuffer,
187        _dest_address: &mut Option<SocketAddress>,
188        _ancillary_data: &mut Vec<AncillaryData>,
189    ) -> Result<usize, Errno> {
190        let file = {
191            let inner = self.lock();
192            match &inner.state {
193                VsockSocketState::Connected { file, .. } => file.clone(),
194                _ => return error!(EBADF),
195            }
196        };
197        current_task.override_creds(Credentials::root(), || file.write(locked, current_task, data))
198    }
199
200    fn wait_async(
201        &self,
202        locked: &mut Locked<FileOpsCore>,
203        _socket: &Socket,
204        current_task: &CurrentTask,
205        waiter: &Waiter,
206        events: FdEvents,
207        handler: EventHandler,
208    ) -> WaitCanceler {
209        let inner = self.lock();
210        match &inner.state {
211            VsockSocketState::Connected { file, .. } => file
212                .wait_async(locked, current_task, waiter, events, handler)
213                .expect("vsock socket should be connected to a file that can be waited on"),
214            _ => inner.waiters.wait_async_fd_events(waiter, events, handler),
215        }
216    }
217
218    fn query_events(
219        &self,
220        locked: &mut Locked<FileOpsCore>,
221        _socket: &Socket,
222        current_task: &CurrentTask,
223    ) -> Result<FdEvents, Errno> {
224        self.lock().query_events(locked, current_task)
225    }
226
227    fn shutdown(
228        &self,
229        _locked: &mut Locked<FileOpsCore>,
230        _socket: &Socket,
231        _how: SocketShutdownFlags,
232    ) -> Result<(), Errno> {
233        self.lock().state = VsockSocketState::Closed;
234        Ok(())
235    }
236
237    fn close(
238        &self,
239        locked: &mut Locked<FileOpsCore>,
240        _current_task: &CurrentTask,
241        socket: &Socket,
242    ) {
243        // Call to shutdown should never fail, so unwrap is OK
244        self.shutdown(locked, socket, SocketShutdownFlags::READ | SocketShutdownFlags::WRITE)
245            .unwrap();
246    }
247
248    fn getsockname(
249        &self,
250        _locked: &mut Locked<FileOpsCore>,
251        socket: &Socket,
252    ) -> Result<SocketAddress, Errno> {
253        let inner = self.lock();
254        if let Some(address) = &inner.address {
255            Ok(address.clone())
256        } else {
257            Ok(SocketAddress::default_for_domain(socket.domain))
258        }
259    }
260
261    fn getpeername(
262        &self,
263        _locked: &mut Locked<FileOpsCore>,
264        _socket: &Socket,
265    ) -> Result<SocketAddress, Errno> {
266        let inner = self.lock();
267        match &inner.state {
268            VsockSocketState::Connected { peer_addr, .. } => Ok(peer_addr.clone()),
269            _ => {
270                error!(ENOTCONN)
271            }
272        }
273    }
274}
275
276impl VsockSocket {
277    pub fn remote_connection<L>(
278        &self,
279        locked: &mut Locked<L>,
280        socket: &Socket,
281        current_task: &CurrentTask,
282        file: FileHandle,
283    ) -> Result<(), Errno>
284    where
285        L: LockEqualOrBefore<FileOpsCore>,
286    {
287        // we only allow non-blocking files here, so that
288        // read and write on file can return EAGAIN.
289        assert!(file.flags().contains(OpenFlags::NONBLOCK));
290        if socket.socket_type != SocketType::Stream {
291            return error!(ENOTSUP);
292        }
293        if socket.domain != SocketDomain::Vsock {
294            return error!(EINVAL);
295        }
296
297        let mut inner = self.lock();
298        match &mut inner.state {
299            VsockSocketState::Listening(queue) => {
300                if queue.sockets.len() >= queue.backlog {
301                    return error!(EAGAIN);
302                }
303                let remote_socket = Socket::new(
304                    locked,
305                    current_task,
306                    SocketDomain::Vsock,
307                    SocketType::Stream,
308                    SocketProtocol::default(),
309                    /* kernel_private = */ true,
310                )?;
311                // The lock on remote_socket is safe because the socket is not shared with any other
312                // thread yet.
313                let _token = allow_subclass();
314                downcast_socket_to_vsock(&remote_socket).lock().state =
315                    VsockSocketState::Connected {
316                        file,
317                        peer_addr: SocketAddress::Vsock {
318                            port: u32::MAX,
319                            cid: starnix_uapi::VMADDR_CID_HOST,
320                        },
321                    };
322                queue.sockets.push_back(remote_socket);
323                inner.waiters.notify_fd_events(FdEvents::POLLIN);
324                Ok(())
325            }
326            _ => error!(EINVAL),
327        }
328    }
329}
330
331impl VsockSocketInner {
332    fn query_events<L>(
333        &self,
334        locked: &mut Locked<L>,
335        current_task: &CurrentTask,
336    ) -> Result<FdEvents, Errno>
337    where
338        L: LockEqualOrBefore<FileOpsCore>,
339    {
340        Ok(match &self.state {
341            VsockSocketState::Disconnected => FdEvents::empty(),
342            VsockSocketState::Connected { file, .. } => current_task
343                .override_creds(Credentials::root(), || file.query_events(locked, current_task))?,
344            VsockSocketState::Listening(queue) => {
345                if !queue.sockets.is_empty() {
346                    FdEvents::POLLIN
347                } else {
348                    FdEvents::empty()
349                }
350            }
351            VsockSocketState::Closed => FdEvents::POLLHUP,
352        })
353    }
354}
355
356#[cfg(test)]
357mod tests {
358    use super::*;
359    use crate::fs::fuchsia::create_fuchsia_pipe;
360    use crate::mm::PAGE_SIZE;
361    use crate::task::dynamic_thread_spawner::SpawnRequestBuilder;
362    use crate::testing::spawn_kernel_and_run;
363    use crate::vfs::EpollFileObject;
364    use crate::vfs::buffers::{VecInputBuffer, VecOutputBuffer};
365    use crate::vfs::socket::SocketFile;
366    use futures::executor::block_on;
367    use starnix_sync::Unlocked;
368    use starnix_uapi::vfs::EpollEvent;
369    use syncio::Zxio;
370
371    #[::fuchsia::test]
372    async fn test_vsock_socket() {
373        spawn_kernel_and_run(async |locked, current_task| {
374            let (fs1, fs2) = fidl::Socket::create_stream();
375            const VSOCK_PORT: u32 = 5555;
376
377            let listen_socket = Socket::new(
378                locked,
379                &current_task,
380                SocketDomain::Vsock,
381                SocketType::Stream,
382                SocketProtocol::default(),
383                /* kernel_private = */ false,
384            )
385            .expect("Failed to create socket.");
386            let vsock_ns = current_task.running_state().abstract_vsock_namespace.clone();
387            vsock_ns
388                .bind(locked, &current_task, VSOCK_PORT, &listen_socket)
389                .expect("Failed to bind socket.");
390            listen_socket.listen(locked, &current_task, 10).expect("Failed to listen.");
391
392            let listen_socket =
393                vsock_ns.lookup(&VSOCK_PORT).expect("Failed to look up listening socket.");
394            let remote = create_fuchsia_pipe(
395                locked,
396                &current_task,
397                fs2,
398                OpenFlags::RDWR | OpenFlags::NONBLOCK,
399            )
400            .unwrap();
401            listen_socket
402                .downcast_socket::<VsockSocket>()
403                .unwrap()
404                .remote_connection(locked, &listen_socket, &current_task, remote)
405                .unwrap();
406
407            let server_socket = listen_socket.accept(locked, &current_task).unwrap();
408
409            let test_bytes_in: [u8; 5] = [0, 1, 2, 3, 4];
410            assert_eq!(fs1.write(&test_bytes_in[..]).unwrap(), test_bytes_in.len());
411            let mut buffer_iterator = VecOutputBuffer::new(*PAGE_SIZE as usize);
412            let read_message_info = server_socket
413                .read(locked, &current_task, &mut buffer_iterator, SocketMessageFlags::empty())
414                .unwrap();
415            assert_eq!(read_message_info.bytes_read, test_bytes_in.len());
416            assert_eq!(buffer_iterator.data(), test_bytes_in);
417
418            let test_bytes_out: [u8; 10] = [9, 8, 7, 6, 5, 4, 3, 2, 1, 0];
419            let mut buffer_iterator = VecInputBuffer::new(&test_bytes_out);
420            server_socket
421                .write(locked, &current_task, &mut buffer_iterator, &mut None, &mut vec![])
422                .unwrap();
423            assert_eq!(buffer_iterator.bytes_read(), test_bytes_out.len());
424
425            let mut read_back_buf = [0u8; 100];
426            assert_eq!(test_bytes_out.len(), fs1.read(&mut read_back_buf).unwrap());
427            assert_eq!(&read_back_buf[..test_bytes_out.len()], &test_bytes_out);
428
429            server_socket.close(locked, &current_task);
430            listen_socket.close(locked, &current_task);
431        })
432        .await;
433    }
434
435    #[::fuchsia::test]
436    async fn test_vsock_write_while_read() {
437        spawn_kernel_and_run(async |locked, current_task| {
438            let kernel = current_task.kernel();
439            let (fs1, fs2) = fidl::Socket::create_stream();
440            let socket = Socket::new(
441                locked,
442                &current_task,
443                SocketDomain::Vsock,
444                SocketType::Stream,
445                SocketProtocol::default(),
446                /* kernel_private = */ false,
447            )
448            .expect("Failed to create socket.");
449            let remote = create_fuchsia_pipe(
450                locked,
451                &current_task,
452                fs2,
453                OpenFlags::RDWR | OpenFlags::NONBLOCK,
454            )
455            .unwrap();
456            downcast_socket_to_vsock(&socket).lock().state = VsockSocketState::Connected {
457                file: remote,
458                peer_addr: SocketAddress::Vsock {
459                    port: u32::MAX,
460                    cid: starnix_uapi::VMADDR_CID_HOST,
461                },
462            };
463            let socket_file =
464                SocketFile::from_socket(locked, &current_task, socket, OpenFlags::RDWR, false)
465                    .expect("Failed to create socket file.");
466
467            const XFER_SIZE: usize = 42;
468
469            let socket_clone = socket_file.clone();
470            let closure = move |locked: &mut Locked<Unlocked>, current_task: &CurrentTask| {
471                let bytes_read = socket_clone
472                    .read(locked, current_task, &mut VecOutputBuffer::new(XFER_SIZE))
473                    .unwrap();
474                assert_eq!(XFER_SIZE, bytes_read);
475            };
476            let (result, req) =
477                SpawnRequestBuilder::new().with_sync_closure(closure).build_with_async_result();
478            kernel.kthreads.spawner().spawn_from_request(req);
479
480            // Wait for the thread to become blocked on the read.
481            std::thread::sleep(std::time::Duration::from_secs(2));
482
483            socket_file
484                .write(locked, &current_task, &mut VecInputBuffer::new(&[0; XFER_SIZE]))
485                .unwrap();
486
487            let mut buffer = [0u8; 1024];
488            assert_eq!(XFER_SIZE, fs1.read(&mut buffer).unwrap());
489            assert_eq!(XFER_SIZE, fs1.write(&buffer[..XFER_SIZE]).unwrap());
490            block_on(result).unwrap();
491        })
492        .await;
493    }
494
495    #[::fuchsia::test]
496    async fn test_vsock_poll() {
497        spawn_kernel_and_run(async |locked, current_task| {
498            let (client, server) = zx::Socket::create_stream();
499            let pipe = create_fuchsia_pipe(locked, &current_task, client, OpenFlags::RDWR)
500                .expect("create_fuchsia_pipe");
501            let server_zxio = Zxio::create(server.into_handle()).expect("Zxio::create");
502            let socket_object = Socket::new(
503                locked,
504                &current_task,
505                SocketDomain::Vsock,
506                SocketType::Stream,
507                SocketProtocol::default(),
508                /* kernel_private = */ false,
509            )
510            .expect("Failed to create socket.");
511            downcast_socket_to_vsock(&socket_object).lock().state = VsockSocketState::Connected {
512                file: pipe,
513                peer_addr: SocketAddress::Vsock {
514                    port: u32::MAX,
515                    cid: starnix_uapi::VMADDR_CID_HOST,
516                },
517            };
518            let socket = SocketFile::from_socket(
519                locked,
520                &current_task,
521                socket_object.clone(),
522                OpenFlags::RDWR,
523                false,
524            )
525            .expect("Failed to create socket file.");
526
527            assert_eq!(
528                socket.query_events(locked, &current_task),
529                Ok(FdEvents::POLLOUT | FdEvents::POLLWRNORM)
530            );
531
532            let epoll_object = EpollFileObject::new_file(locked, &current_task);
533            let epoll_file = epoll_object.downcast_file::<EpollFileObject>().unwrap();
534            let event = EpollEvent::new(FdEvents::POLLIN, 0);
535            epoll_file
536                .add(locked, &current_task, &socket, &epoll_object, event)
537                .expect("poll_file.add");
538
539            let fds = epoll_file
540                .wait(locked, &current_task, 1, zx::MonotonicInstant::ZERO)
541                .expect("wait");
542            assert!(fds.is_empty());
543
544            assert_eq!(server_zxio.write(&[0]).expect("write"), 1);
545
546            assert_eq!(
547                socket.query_events(locked, &current_task),
548                Ok(FdEvents::POLLOUT
549                    | FdEvents::POLLWRNORM
550                    | FdEvents::POLLIN
551                    | FdEvents::POLLRDNORM)
552            );
553            let fds = epoll_file
554                .wait(locked, &current_task, 1, zx::MonotonicInstant::ZERO)
555                .expect("wait");
556            assert_eq!(fds.len(), 1);
557
558            assert_eq!(
559                socket.read(locked, &current_task, &mut VecOutputBuffer::new(64)).expect("read"),
560                1
561            );
562
563            assert_eq!(
564                socket.query_events(locked, &current_task),
565                Ok(FdEvents::POLLOUT | FdEvents::POLLWRNORM)
566            );
567            let fds = epoll_file
568                .wait(locked, &current_task, 1, zx::MonotonicInstant::ZERO)
569                .expect("wait");
570            assert!(fds.is_empty());
571        })
572        .await;
573    }
574}
starnix_core/vfs/socket/socket_vsock.rs

starnix_core/vfs/socket/
socket_vsock.rs
