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

starnix_core/vfs/socket/
socket_vsock.rs
