Merge pull request #313032 from microsoft/connor4312/cli-ahp-management

cli: implement client AHP control functionality

Author: connor4312

cli/Cargo.lock

@@ -15,54 +15,61 @@ name = "code"
 futures = "0.3.28"
 clap = { version = "4.3.0", features = ["derive", "env"] }
 open = "4.1.0"
-reqwest = { version = "0.11.22", default-features = false, features = ["json", "stream", "native-tls"] }
-tokio = { version = "1.38.2", features = ["full"] }
+reqwest = { version = "0.13", default-features = false, features = ["json", "stream", "native-tls"] }
+tokio = { version = "1.52", features = ["full"] }
 tokio-util = { version = "0.7.8", features = ["compat", "codec"] }
 flate2 = { version = "1.0.26", default-features = false, features = ["zlib"] }
 zip = { version = "0.6.6", default-features = false, features = ["time", "deflate-zlib"] }
 regex = "1.8.3"
-lazy_static = "1.4.0"
 sysinfo = { version = "0.29.0", default-features = false }
 serde = { version = "1.0.163", features = ["derive"] }
 serde_json = "1.0.96"
 rmp-serde = "1.1.1"
 uuid = { version = "1.4", features = ["serde", "v4"] }
 dirs = "5.0.1"
-rand = "0.9.3"
-opentelemetry = { version = "0.19.0", features = ["rt-tokio"] }
+rand = "0.10"
 serde_bytes = "0.11.9"
-chrono = { version = "0.4.26", features = ["serde", "std", "clock"], default-features = false }
+jiff = { version = "0.2", default-features = false, features = ["std", "serde"] }
+http = "1"
 gethostname = "0.4.3"
 libc = "0.2.144"
-tunnels = { git = "https://github.com/microsoft/dev-tunnels", rev = "8cae9b2a24c65c6c1958f5a0e77d72b23b5c6c30", default-features = false, features = ["connections"] }
+# temporary fork pending https://github.com/microsoft/dev-tunnels/pull/626
+tunnels = { git = "https://github.com/connor4312/dev-tunnels", rev = "4be50b3cc5ade8cb6beec4038c53ea4f2cdac5a2", default-features = false, features = ["connections"] }
 keyring = { version = "2.0.3", default-features = false, features = ["linux-secret-service-rt-tokio-crypto-openssl", "platform-windows", "platform-macos", "linux-keyutils"] }
 dialoguer = "0.10.4"
-hyper = { version = "0.14.26", features = ["server", "http1", "runtime"] }
+hyper = { version = "1", features = ["server", "http1", "client"] }
+hyper-util = { version = "0.1", features = ["tokio", "server-auto"] }
+http-body-util = "0.1"
+tokio-tungstenite = { version = "0.29", features = ["native-tls"] }
 indicatif = "0.17.4"
 tempfile = "3.5.0"
-clap_lex = "0.7.0"
+clap_lex = "1"
 url = "2.5.4"
-async-trait = "0.1.68"
 log = "0.4.18"
 const_format = "0.2.31"
 sha2 = "0.10.6"
-base64 = "0.21.2"
+base64 = "0.22"
 shell-escape = "0.1.5"
-thiserror = "1.0.40"
+thiserror = "2"
 cfg-if = "1.0.0"
 pin-project = "1.1.0"
 console = "0.15.7"
 bytes = "1.11.1"
 tar = "0.4.45"
+local-ip-address = "0.6"
+ahp = "0.1"
+ahp-types = "0.1"
+ahp-ws = "0.1"
 
 [build-dependencies]
 serde = { version="1.0.163", features = ["derive"] }
 serde_json = "1.0.96"
 winresource = "0.1"
 
 [target.'cfg(windows)'.dependencies]
-winreg = "0.50.0"
+winreg = "0.56"
 winapi = "0.3.9"
+windows-sys = { version = "0.61", features = ["Win32_System_Console", "Win32_UI_Input_KeyboardAndMouse"] }
 
 [target.'cfg(target_os = "macos")'.dependencies]
 core-foundation = "0.9.3"

cli/Cargo.toml

@@ -4,10 +4,9 @@
  *--------------------------------------------------------------------------------------------*/
 
 use crate::{constants::APPLICATION_NAME, util::errors::CodeError};
-use async_trait::async_trait;
+use std::future::Future;
 use std::path::{Path, PathBuf};
 use std::pin::Pin;
-use std::task::{Context, Poll};
 use tokio::io::{AsyncRead, AsyncWrite};
 use tokio::net::TcpListener;
 use uuid::Uuid;
@@ -46,6 +45,7 @@ cfg_if::cfg_if! {
 	} else {
 		use tokio::{time::sleep, io::ReadBuf};
 		use tokio::net::windows::named_pipe::{ClientOptions, ServerOptions, NamedPipeClient, NamedPipeServer};
+		use std::task::{Context, Poll};
 		use std::{time::Duration, io};
 		use pin_project::pin_project;
 
@@ -176,57 +176,6 @@ cfg_if::cfg_if! {
 	}
 }
 
-impl AsyncPipeListener {
-	pub fn into_pollable(self) -> PollableAsyncListener {
-		PollableAsyncListener {
-			listener: Some(self),
-			write_fut: tokio_util::sync::ReusableBoxFuture::new(make_accept_fut(None)),
-		}
-	}
-}
-
-pub struct PollableAsyncListener {
-	listener: Option<AsyncPipeListener>,
-	write_fut: tokio_util::sync::ReusableBoxFuture<
-		'static,
-		(AsyncPipeListener, Result<AsyncPipe, CodeError>),
-	>,
-}
-
-async fn make_accept_fut(
-	data: Option<AsyncPipeListener>,
-) -> (AsyncPipeListener, Result<AsyncPipe, CodeError>) {
-	match data {
-		Some(mut l) => {
-			let c = l.accept().await;
-			(l, c)
-		}
-		None => unreachable!("this future should not be pollable in this state"),
-	}
-}
-
-impl hyper::server::accept::Accept for PollableAsyncListener {
-	type Conn = AsyncPipe;
-	type Error = CodeError;
-
-	fn poll_accept(
-		mut self: Pin<&mut Self>,
-		cx: &mut Context<'_>,
-	) -> Poll<Option<Result<Self::Conn, Self::Error>>> {
-		if let Some(l) = self.listener.take() {
-			self.write_fut.set(make_accept_fut(Some(l)))
-		}
-
-		match self.write_fut.poll(cx) {
-			Poll::Ready((l, cnx)) => {
-				self.listener = Some(l);
-				Poll::Ready(Some(cnx))
-			}
-			Poll::Pending => Poll::Pending,
-		}
-	}
-}
-
 /// Gets a random name for a pipe/socket on the platform
 pub fn get_socket_name() -> PathBuf {
 	cfg_if::cfg_if! {
@@ -243,28 +192,41 @@ pub type AcceptedRW = (
 	Box<dyn AsyncWrite + Send + Unpin>,
 );
 
-#[async_trait]
 pub trait AsyncRWAccepter {
-	async fn accept_rw(&mut self) -> Result<AcceptedRW, CodeError>;
+	fn accept_rw(
+		&mut self,
+	) -> Pin<Box<dyn Future<Output = Result<AcceptedRW, CodeError>> + Send + '_>>;
 }
 
-#[async_trait]
 impl AsyncRWAccepter for AsyncPipeListener {
-	async fn accept_rw(&mut self) -> Result<AcceptedRW, CodeError> {
-		let pipe = self.accept().await?;
-		let (read, write) = socket_stream_split(pipe);
-		Ok((Box::new(read), Box::new(write)))
+	fn accept_rw(
+		&mut self,
+	) -> Pin<Box<dyn Future<Output = Result<AcceptedRW, CodeError>> + Send + '_>> {
+		Box::pin(async move {
+			let pipe = self.accept().await?;
+			let (read, write) = socket_stream_split(pipe);
+			Ok((
+				Box::new(read) as Box<dyn AsyncRead + Send + Unpin>,
+				Box::new(write) as Box<dyn AsyncWrite + Send + Unpin>,
+			))
+		})
 	}
 }
 
-#[async_trait]
 impl AsyncRWAccepter for TcpListener {
-	async fn accept_rw(&mut self) -> Result<AcceptedRW, CodeError> {
-		let (stream, _) = self
-			.accept()
-			.await
-			.map_err(CodeError::AsyncPipeListenerFailed)?;
-		let (read, write) = tokio::io::split(stream);
-		Ok((Box::new(read), Box::new(write)))
+	fn accept_rw(
+		&mut self,
+	) -> Pin<Box<dyn Future<Output = Result<AcceptedRW, CodeError>> + Send + '_>> {
+		Box::pin(async move {
+			let (stream, _) = self
+				.accept()
+				.await
+				.map_err(CodeError::AsyncPipeListenerFailed)?;
+			let (read, write) = tokio::io::split(stream);
+			Ok((
+				Box::new(read) as Box<dyn AsyncRead + Send + Unpin>,
+				Box::new(write) as Box<dyn AsyncWrite + Send + Unpin>,
+			))
+		})
 	}
 }