Tunnel sush through Nexus

nexus/Cargo.toml

@@ -129,6 +129,7 @@ steno.workspace = true
 thiserror.workspace = true
 tokio = { workspace = true, features = ["full"] }
 tokio-postgres = { workspace = true, features = ["with-serde_json-1"] }
+tokio-tungstenite.workspace = true
 tokio-util = { workspace = true, features = ["codec", "rt"] }
 trust-quorum-types.workspace = true
 tufaceous.workspace = true

nexus/external-api/src/lib.rs

@@ -12,7 +12,8 @@ use dropshot::{
     HttpResponseFound, HttpResponseHeaders, HttpResponseOk,
     HttpResponseSeeOther, HttpResponseUpdatedNoContent, PaginationParams, Path,
     Query, RequestContext, ResultsPage, StreamingBody, TypedBody,
-    WebsocketChannelResult, WebsocketConnection,
+    WebsocketChannelResult, WebsocketConnection, WebsocketEndpointResult,
+    WebsocketUpgrade,
 };
 use dropshot_api_manager_types::{ValidationContext, api_versions};
 use http::Response;
@@ -7548,6 +7549,22 @@ pub trait NexusExternalApi {
         path_params: Path<latest::path_params::RackPath>,
     ) -> Result<HttpResponseOk<latest::rack::Rack>, HttpError>;
 
+    /// Tunnel to a Support Shell proxy in a switch zone
+    // This should use `channel { protocol = WEBSOCKETS, .. }`, but
+    // that does not let us return (unauthorized) errors before the
+    // connection upgrade.
+    #[endpoint {
+        method = GET,
+        path = "/v1/system/hardware/racks/{rack_id}/support-shell/tunnel",
+        tags = ["system/hardware"],
+        unpublished = true,
+    }]
+    async fn rack_support_shell_tunnel(
+        rqctx: RequestContext<Self::Context>,
+        path_params: Path<latest::path_params::RackPath>,
+        upgrade: WebsocketUpgrade,
+    ) -> WebsocketEndpointResult;
+
     /// List uninitialized sleds
     #[endpoint {
         method = GET,

nexus/src/app/mod.rs

@@ -104,6 +104,7 @@ mod snapshot;
 mod ssh_key;
 mod subnet_pool;
 pub(crate) mod support_bundles;
+mod support_shell;
 mod switch;
 mod switch_interface;
 mod switch_port;
@@ -1432,7 +1433,7 @@ pub(crate) async fn lldpd_clients(
 /// # Errors
 /// If we fail to resolve the ipv6 addresses of the Dendrite service we
 /// return an error
-async fn switch_zone_address_mappings(
+pub(crate) async fn switch_zone_address_mappings(
     resolver: &internal_dns_resolver::Resolver,
     log: &slog::Logger,
 ) -> Result<HashMap<SwitchSlot, Ipv6Addr>, String> {

nexus/src/app/support_shell.rs

@@ -0,0 +1,201 @@
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at https://mozilla.org/MPL/2.0/.
+
+//! An authenticated byte tunnel to the rack's Support Shell proxy.
+//!
+//! Nexus copies bytes and nothing else. The client's platform TLS
+//! terminates in the switch zone and every support action is signed
+//! off-rack, so this endpoint decides who may reach the support
+//! plane without being able to read or alter what crosses it.
+//!
+//! The mirror image of this proxy lives in wicketd's `nexus_proxy`,
+//! which carries techport users the other way.
+
+use std::net::{SocketAddr, SocketAddrV6};
+use std::time::Duration;
+
+use dropshot::{
+    WebsocketConnectionRaw, WebsocketEndpointResult, WebsocketUpgrade,
+};
+use futures::{SinkExt, StreamExt};
+use nexus_db_queries::authz;
+use nexus_db_queries::context::OpContext;
+use omicron_common::address::SUSH_PROXY_PORT;
+use omicron_common::api::external::Error;
+use omicron_uuid_kinds::RackUuid;
+use slog::{Logger, info, o, warn};
+use slog_error_chain::InlineErrorChain;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpStream;
+use tokio::time::timeout;
+use tokio_tungstenite::WebSocketStream;
+use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::tungstenite::protocol::{Role, WebSocketConfig};
+
+use crate::app::switch_zone_address_mappings;
+
+/// How long to wait for one switch before trying the other.
+const CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
+
+/// The client's pipe sends 8 KiB frames; anything much bigger is not
+/// our client, and the default cap is a 64 MiB allocation.
+const MAX_WS_MESSAGE_SIZE: usize = 0x2_0000;
+
+impl super::Nexus {
+    /// Tunnel a support client's connection to a sush proxy.
+    /// Errors are returned before the websocket upgrade so
+    /// the client learns why the connection failed.
+    pub(crate) async fn support_shell_tunnel(
+        &self,
+        opctx: &OpContext,
+        rack_id: RackUuid,
+        upgrade: WebsocketUpgrade,
+    ) -> WebsocketEndpointResult {
+        let log = opctx.log.new(o!(
+            "component" => "SupportShellTunnel",
+            "rack_id" => rack_id.to_string(),
+            "actor" => format!("{:?}", opctx.authn.actor()),
+        ));
+        let proxy = self.support_shell_proxy(opctx, &rack_id, &log).await?;
+        let log = match proxy.peer_addr() {
+            Ok(addr) => log.new(o!("proxy_addr" => addr)),
+            Err(_) => log,
+        };
+        upgrade.handle(move |conn| async move {
+            let config = WebSocketConfig {
+                max_message_size: Some(MAX_WS_MESSAGE_SIZE),
+                max_frame_size: Some(MAX_WS_MESSAGE_SIZE),
+                ..Default::default()
+            };
+            let client = WebSocketStream::from_raw_socket(
+                conn.into_inner(),
+                Role::Server,
+                Some(config),
+            )
+            .await;
+            info!(log, "tunnel opened");
+            let PipeSummary { reason, to_proxy, to_client } =
+                pipe(client, proxy).await;
+            info!(
+                log, "tunnel closed";
+                "reason" => reason,
+                "bytes_sent_to_proxy" => to_proxy,
+                "bytes_sent_to_client" => to_client,
+            );
+            Ok(())
+        })
+    }
+
+    /// Authorize the tunnel and connect to a sush proxy, on whichever
+    /// switch answers first.
+    ///
+    /// The proxies have no directory entry of their own; like lldpd,
+    /// they answer on the switch zone addresses dendrite advertises.
+    async fn support_shell_proxy(
+        &self,
+        opctx: &OpContext,
+        rack_id: &RackUuid,
+        log: &Logger,
+    ) -> Result<TcpStream, Error> {
+        // The lookup comes first so an unauthorized caller gets the
+        // same 404 as for a rack that does not exist. It otherwise
+        // only validates existence: like lldpd_clients, we assume the
+        // single rack is ours (omicron#1276).
+        self.rack_lookup(opctx, rack_id).await?;
+        opctx.authorize(authz::Action::Modify, &authz::FLEET).await?;
+        let proxy_addrs =
+            switch_zone_address_mappings(&self.internal_resolver, log)
+                .await
+                .map_err(|e| Error::unavail(&e))?
+                .into_values()
+                .map(|ip| {
+                    SocketAddr::V6(SocketAddrV6::new(ip, SUSH_PROXY_PORT, 0, 0))
+                })
+                .collect::<Vec<_>>();
+        for addr in &proxy_addrs {
+            match timeout(CONNECT_TIMEOUT, TcpStream::connect(addr)).await {
+                Ok(Ok(stream)) => {
+                    let _ = stream.set_nodelay(true);
+                    return Ok(stream);
+                }
+                Ok(Err(error)) => warn!(
+                    log, "sush proxy unreachable";
+                    "addr" => %addr,
+                    InlineErrorChain::new(&error),
+                ),
+                Err(_) => warn!(
+                    log, "sush proxy connect timed out";
+                    "addr" => %addr,
+                ),
+            }
+        }
+        Err(Error::unavail("no sush proxy reachable on either switch"))
+    }
+}
+
+/// Why a tunnel ended and how much it carried.
+struct PipeSummary {
+    reason: &'static str,
+    to_proxy: u64,
+    to_client: u64,
+}
+
+/// Copy bytes both ways until either side finishes, returning why
+/// and how much. The directions are independent pipes; when either
+/// ends, both are torn down, since half-open service would only
+/// delay the client's error.
+async fn pipe(
+    client: WebSocketStream<WebsocketConnectionRaw>,
+    proxy: TcpStream,
+) -> PipeSummary {
+    let (mut ws_sink, mut ws_source) = client.split();
+    let (mut proxy_read, mut proxy_write) = proxy.into_split();
+    let mut to_proxy = 0;
+    let mut to_client = 0;
+
+    let inbound = async {
+        while let Some(message) = ws_source.next().await {
+            match message {
+                Ok(Message::Binary(data)) => {
+                    if proxy_write.write_all(&data).await.is_err() {
+                        return "proxy write failed";
+                    }
+                    to_proxy += data.len() as u64;
+                }
+                Ok(Message::Close(_)) | Err(_) => break,
+                // Tungstenite answers pings itself.
+                Ok(Message::Ping(_)) | Ok(Message::Pong(_)) => {}
+                // Anything else would truncate the byte stream
+                // invisibly; fail loudly instead.
+                Ok(_) => return "unexpected frame",
+            }
+        }
+        let _ = proxy_write.shutdown().await;
+        "client closed"
+    };
+
+    let outbound = async {
+        let mut buf = [0; 0x2000];
+        loop {
+            match proxy_read.read(&mut buf).await {
+                Ok(0) | Err(_) => break,
+                Ok(n) => {
+                    let data = buf[..n].to_vec();
+                    if ws_sink.send(Message::Binary(data)).await.is_err() {
+                        return "client send failed";
+                    }
+                    to_client += n as u64;
+                }
+            }
+        }
+        let _ = ws_sink.send(Message::Close(None)).await;
+        "proxy closed"
+    };
+
+    let reason = tokio::select! {
+        reason = inbound => reason,
+        reason = outbound => reason,
+    };
+    PipeSummary { reason, to_proxy, to_client }
+}

nexus/src/external_api/http_entrypoints.rs

@@ -27,7 +27,10 @@ use dropshot::WhichPage;
 use dropshot::{ApiDescription, StreamingBody};
 use dropshot::{HttpResponseAccepted, HttpResponseFound, HttpResponseSeeOther};
 use dropshot::{HttpResponseCreated, HttpResponseHeaders};
-use dropshot::{WebsocketChannelResult, WebsocketConnection};
+use dropshot::{
+    WebsocketChannelResult, WebsocketConnection, WebsocketEndpointResult,
+    WebsocketUpgrade,
+};
 use dropshot::{http_response_found, http_response_see_other};
 use http::{Response, StatusCode, header};
 use ipnetwork::IpNetwork;
@@ -6663,6 +6666,24 @@ impl NexusExternalApi for NexusExternalApiImpl {
             .await
     }
 
+    async fn rack_support_shell_tunnel(
+        rqctx: RequestContext<ApiContext>,
+        path_params: Path<path_params::RackPath>,
+        upgrade: WebsocketUpgrade,
+    ) -> WebsocketEndpointResult {
+        let apictx = rqctx.context();
+        let nexus = &apictx.context.nexus;
+        let path = path_params.into_inner();
+        let opctx = crate::context::op_context_for_external_api(&rqctx).await?;
+        nexus
+            .support_shell_tunnel(
+                &opctx,
+                RackUuid::from_untyped_uuid(path.rack_id),
+                upgrade,
+            )
+            .await
+    }
+
     // This request isn't currently paginated. The query is somewhat complex
     // underneath and doesn't support pagination right now. We would need a way
     // to order filtered `InvPhysicalDisk` objects to support pagination, and

nexus/test-utils/src/http_testing.rs

@@ -295,6 +295,19 @@ impl<'a> RequestBuilder<'a> {
             )
     }
 
+    /// Tells the request to initiate a WebSocket upgrade handshake
+    /// without expecting it to succeed, for endpoints that refuse in
+    /// HTTP before upgrading. This also sets the request method to
+    /// GET. The caller supplies the expected failure status.
+    pub fn websocket_handshake_headers(mut self) -> Self {
+        const TEST_WEBSOCKET_REQUEST_KEY: &str = "SEFDSyBUSEUgUExBTkVUIQ==";
+        self.method = http::method::Method::GET;
+        self.header(http::header::CONNECTION, "Upgrade")
+            .header(http::header::UPGRADE, "websocket")
+            .header(http::header::SEC_WEBSOCKET_VERSION, "13")
+            .header(http::header::SEC_WEBSOCKET_KEY, TEST_WEBSOCKET_REQUEST_KEY)
+    }
+
     /// Expect a successful console asset response.
     pub fn expect_console_asset(mut self) -> Self {
         let headers = [
@@ -646,6 +659,13 @@ impl<'a> NexusRequest<'a> {
         self
     }
 
+    /// See [`RequestBuilder::websocket_handshake_headers()`].
+    pub fn websocket_handshake_headers(mut self) -> Self {
+        self.request_builder =
+            self.request_builder.websocket_handshake_headers();
+        self
+    }
+
     /// Allow non-Dropshot error responses (e.g., SCIM endpoints).
     pub fn allow_non_dropshot_errors(mut self) -> Self {
         self.request_builder = self.request_builder.allow_non_dropshot_errors();

nexus/tests/integration_tests/audit_log.rs

@@ -515,6 +515,7 @@ async fn test_audit_log_coverage(ctx: &ControlPlaneTestContext) {
                 | AllowedMethod::GetUnimplemented
                 | AllowedMethod::GetVolatile
                 | AllowedMethod::GetWebsocket
+                | AllowedMethod::GetWebsocketUnavailable
                 | AllowedMethod::Head
                 | AllowedMethod::HeadNonexistent => false,
             };