guest_cli/
socat.rs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
// Copyright 2022 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
use crate::platform::{GuestConsole, PlatformServices};
use fidl_fuchsia_virtualization::{
    GuestManagerProxy, GuestMarker, GuestStatus, HostVsockAcceptorMarker, HostVsockEndpointMarker,
    HostVsockEndpointProxy,
};
use futures::TryStreamExt;
use std::fmt;
use {guest_cli_args as arguments, zx_status};

#[derive(Debug, PartialEq, serde::Serialize, serde::Deserialize)]
pub enum SocatResult {
    SocatSuccess(SocatSuccess),
    SocatError(SocatError),
}

impl fmt::Display for SocatResult {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            SocatResult::SocatSuccess(v) => write!(f, "{}", v),
            SocatResult::SocatError(v) => write!(f, "{}", v),
        }
    }
}

#[derive(Debug, PartialEq, serde::Serialize, serde::Deserialize)]
pub enum SocatSuccess {
    Connected,
    Listened(u32),
}

impl fmt::Display for SocatSuccess {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            SocatSuccess::Connected => write!(f, "Disconnected after a successful connect"),
            SocatSuccess::Listened(port) => write!(f, "Stopped listening on port {}", port),
        }
    }
}

#[derive(Debug, PartialEq, serde::Serialize, serde::Deserialize)]
pub enum SocatError {
    NotRunning,
    NoVsockDevice,
    NoListener(u32),
    FailedToListen(u32),
    InternalFailure(String),
}

impl fmt::Display for SocatError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            SocatError::NotRunning => write!(f, "Can't attach to a non-running guest"),
            SocatError::NoListener(port) => write!(f, "No listener on port {}", port),
            SocatError::FailedToListen(port) => write!(f, "Already a listener on port {}", port),
            SocatError::InternalFailure(err) => write!(f, "Internal error: {}", err),
            SocatError::NoVsockDevice => write!(f, "Guest lacks the required vsock device"),
        }
    }
}

impl std::convert::From<fidl::Status> for SocatError {
    fn from(err: fidl::Status) -> SocatError {
        SocatError::InternalFailure(format!("FIDL status - {}", err))
    }
}

impl std::convert::From<fidl::Error> for SocatError {
    fn from(err: fidl::Error) -> SocatError {
        SocatError::InternalFailure(format!("FIDL error - {}", err))
    }
}

fn duplicate_socket(_socket: fidl::Socket) -> Result<(fidl::Socket, fidl::Socket), SocatError> {
    #[cfg(target_os = "fuchsia")]
    {
        use fidl::HandleBased;

        let other = _socket.duplicate_handle(fidl::Rights::SAME_RIGHTS)?;
        Ok((_socket, other))
    }

    // TODO(https://fxbug.dev/42068091): Remove when overnet supports duplicated socket handles.
    #[cfg(not(target_os = "fuchsia"))]
    unimplemented!()
}

async fn handle_socat_listen(
    vsock_endpoint: HostVsockEndpointProxy,
    host_port: u32,
) -> Result<SocatSuccess, SocatError> {
    let (vsock_accept_client, mut vsock_acceptor_stream) =
        fidl::endpoints::create_request_stream::<HostVsockAcceptorMarker>();

    vsock_endpoint
        .listen(host_port, vsock_accept_client)
        .await?
        .map_err(|_| SocatError::FailedToListen(host_port))?;

    // Try_next returns a Result<Option<T>, Err>, hence we need to unwrap our value
    let connection = vsock_acceptor_stream
        .try_next()
        .await?
        .ok_or_else(|| SocatError::InternalFailure("unexpected end of stream".to_string()))?;

    let (_src_cid, _src_port, port, responder) = connection
        .into_accept()
        .ok_or_else(|| SocatError::InternalFailure("unexpected message on stream".to_string()))?;

    if port != host_port {
        responder.send(Err(zx_status::Status::CONNECTION_REFUSED.into_raw()))?;
        return Err(SocatError::InternalFailure(
            "connection attempt on unexpected port".to_string(),
        ));
    }

    let (socket, remote_socket) = fidl::Socket::create_stream();
    responder.send(Ok(remote_socket))?;

    let (input, output) = duplicate_socket(socket)?;

    let console = GuestConsole::new(input, output)
        .map_err(|err| SocatError::InternalFailure(format!("failed to create console: {}", err)))?;
    if let Err(err) = console.run_with_stdio().await {
        Err(SocatError::InternalFailure(format!("failed to run console: {}", err)))
    } else {
        Ok(SocatSuccess::Listened(host_port))
    }
}

async fn handle_socat_connect(
    vsock_endpoint: HostVsockEndpointProxy,
    port: u32,
) -> Result<SocatSuccess, SocatError> {
    let Ok(socket) = vsock_endpoint.connect(port).await? else {
        return Err(SocatError::NoListener(port));
    };

    let (input, output) = duplicate_socket(socket)?;
    let console = GuestConsole::new(input, output)
        .map_err(|err| SocatError::InternalFailure(format!("failed to create console: {}", err)))?;

    if let Err(err) = console.run_with_stdio().await {
        Err(SocatError::InternalFailure(format!("failed to run console: {}", err)))
    } else {
        Ok(SocatSuccess::Connected)
    }
}

async fn connect_to_vsock_endpoint(
    manager: GuestManagerProxy,
) -> Result<HostVsockEndpointProxy, SocatError> {
    let (guest_endpoint, guest_server_end) = fidl::endpoints::create_proxy::<GuestMarker>();
    manager
        .connect(guest_server_end)
        .await?
        .map_err(|err| SocatError::InternalFailure(format!("failed to connect: {:?}", err)))?;

    let (vsock_endpoint, vsock_server_end) =
        fidl::endpoints::create_proxy::<HostVsockEndpointMarker>();

    guest_endpoint
        .get_host_vsock_endpoint(vsock_server_end)
        .await?
        .map_err(|_| SocatError::NoVsockDevice)?;

    Ok(vsock_endpoint)
}

async fn get_manager<P: PlatformServices>(
    services: &P,
    guest_type: arguments::GuestType,
) -> Result<GuestManagerProxy, SocatError> {
    if guest_type == arguments::GuestType::Zircon {
        // We don't have a Zircon vsock device.
        return Err(SocatError::NoVsockDevice);
    }

    let guest_manager = services.connect_to_manager(guest_type).await.map_err(|err| {
        SocatError::InternalFailure(format!("failed to connect to manager: {}", err))
    })?;

    let guest_info = guest_manager.get_info().await?;
    let status = guest_info.guest_status.expect("guest status should always be set");
    if status != GuestStatus::Starting && status != GuestStatus::Running {
        return Err(SocatError::NotRunning);
    }

    Ok(guest_manager)
}

async fn handle_impl<P: PlatformServices>(
    services: &P,
    args: &arguments::socat_args::SocatArgs,
) -> Result<SocatSuccess, SocatError> {
    match &args.socat_cmd {
        arguments::socat_args::SocatCommands::Listen(args) => {
            let manager = get_manager(services, args.guest_type).await?;
            let endpoint = connect_to_vsock_endpoint(manager).await?;
            handle_socat_listen(endpoint, args.host_port).await
        }
        arguments::socat_args::SocatCommands::Connect(args) => {
            let manager = get_manager(services, args.guest_type).await?;
            let endpoint = connect_to_vsock_endpoint(manager).await?;
            handle_socat_connect(endpoint, args.guest_port).await
        }
    }
}

pub async fn handle_socat<P: PlatformServices>(
    services: &P,
    args: &arguments::socat_args::SocatArgs,
) -> SocatResult {
    match handle_impl(services, args).await {
        Err(err) => SocatResult::SocatError(err),
        Ok(ok) => SocatResult::SocatSuccess(ok),
    }
}

#[cfg(test)]
mod test {
    use super::*;
    use fidl::endpoints::create_proxy_and_stream;
    use fuchsia_async as fasync;
    use futures::future::join;
    use futures::StreamExt;

    #[fasync::run_until_stalled(test)]
    async fn socat_listen_invalid_host_returns_err() {
        let (proxy, mut stream) = create_proxy_and_stream::<HostVsockEndpointMarker>();
        let server = async move {
            let (port, _acceptor, responder) = stream
                .next()
                .await
                .expect("Failed to read from stream")
                .expect("Failed to parse request")
                .into_listen()
                .expect("Unexpected call to Guest Proxy");
            assert_eq!(port, 0);
            responder
                .send(Err(zx_status::Status::CONNECTION_REFUSED.into_raw()))
                .expect("Failed to send status code to client");
        };

        let client = handle_socat_listen(proxy, 0);
        let (_server_res, client_res) = join(server, client).await;
        assert_eq!(client_res.unwrap_err().to_string(), "Already a listener on port 0");
    }

    #[fasync::run_until_stalled(test)]
    async fn socat_listen_mismatched_ports_returns_err() {
        let (proxy, mut stream) = create_proxy_and_stream::<HostVsockEndpointMarker>();
        let server = async move {
            let (port, acceptor, responder) = stream
                .next()
                .await
                .expect("Failed to read from stream")
                .expect("Failed to parse request")
                .into_listen()
                .expect("Unexpected call to Guest Proxy");
            assert_eq!(port, 0);
            responder.send(Ok(())).expect("Failed to send status code to client");
            let _ = acceptor.into_proxy().accept(0, 0, 1).await.expect("Failed to accept listener");
        };

        let client = handle_socat_listen(proxy, 0);
        let (_server_res, client_res) = join(server, client).await;
        assert_eq!(
            client_res.unwrap_err().to_string(),
            "Internal error: connection attempt on unexpected port"
        );
    }
}