devices/vsock: host initiation handshake

Demo-quality code, changing the host connection initiation protocol to
include a handshake (i.e. connection ack).

The new protocol looks like this:
- [host] CONNECT <port><LF>
- [guest/success] OK <assigned host port><LF>

On connection failure, the host host connection is reset without any
accompanying message.

Signed-off-by: Dan Horobeanu <dhr@amazon.com>

src/devices/src/virtio/vsock/csm/connection.rs

@@ -541,7 +541,7 @@ where
     ///
     /// Raw data can either be sent straight to the host stream, or to our TX buffer, if the
     /// former fails.
-    fn send_bytes(&mut self, buf: &[u8]) -> Result<()> {
+    pub fn send_bytes(&mut self, buf: &[u8]) -> Result<()> {
         // If there is data in the TX buffer, that means we're already registered for EPOLLOUT
         // events on the underlying stream. Therefore, there's no point in attempting a write
         // at this point. `self.notify()` will get called when EPOLLOUT arrives, and it will
@@ -576,6 +576,11 @@ where
         Ok(())
     }
 
+    /// Return the connections state.
+    pub fn state(&self) -> ConnState {
+        self.state
+    }
+
     /// Check if the credit information the peer has last received from us is outdated.
     fn peer_needs_credit_update(&self) -> bool {
         (self.fwd_cnt - self.last_fwd_cnt_to_peer).0 as usize >= defs::CONN_CREDIT_UPDATE_THRESHOLD

src/devices/src/virtio/vsock/csm/mod.rs

@@ -36,7 +36,7 @@ pub enum Error {
 type Result<T> = std::result::Result<T, Error>;
 
 /// A vsock connection state.
-#[derive(Debug, PartialEq)]
+#[derive(Clone, Copy, Debug, PartialEq)]
 pub enum ConnState {
     /// The connection has been initiated by the host end, but is yet to be confirmed by the guest.
     LocalInit,

src/devices/src/virtio/vsock/unix/muxer.rs

@@ -35,6 +35,7 @@ use std::io::Read;
 use std::os::unix::io::{AsRawFd, RawFd};
 use std::os::unix::net::{UnixListener, UnixStream};
 
+use super::super::csm::ConnState;
 use super::super::defs::uapi;
 use super::super::packet::VsockPacket;
 use super::super::{
@@ -625,9 +626,17 @@ impl VsockMuxer {
         if let Some(conn) = self.conn_map.get_mut(&key) {
             let had_rx = conn.has_pending_rx();
             let was_expiring = conn.will_expire();
+            let prev_state = conn.state();
 
             mut_fn(conn);
 
+            // If this is a host-initiated connection that has just become established, we'll have
+            // to send an ack message to the host end.
+            if prev_state == ConnState::LocalInit && conn.state() == ConnState::Established {
+                conn.send_bytes(format!("OK {}\n", key.local_port).as_bytes())
+                    .unwrap_or_else(|_| conn.kill());
+            }
+
             // If the connection wasn't previously scheduled for RX, add it to our RX queue.
             if !had_rx && conn.has_pending_rx() {
                 self.rxq.push(MuxerRx::ConnRx(key));

tests/integration_tests/functional/test_vsock.py

@@ -19,6 +19,7 @@
 from select import select
 from socket import socket, AF_UNIX, SOCK_STREAM
 from threading import Thread, Event
+import re
 
 from host_tools.network import SSHConnection
 
@@ -306,4 +307,7 @@ def _vsock_connect_to_guest(uds_path, port):
     buf = bytearray("CONNECT {}\n".format(port).encode("utf-8"))
     sock.send(buf)
 
+    ack_buf = sock.recv(32)
+    assert re.match("^OK [0-9]+\n$", ack_buf.decode('utf-8')) is not None
+
     return sock