feat(cli): added proxy support during packages installation (#2535)

Migrated to `isahc` crate that uses `libcurl`, which gives us more consistent package installation including support of `http_proxy` and `https_proxy` environment variables.

Fixes #2517

crates/fluvio-cli-common/Cargo.toml

@@ -18,14 +18,15 @@ default = ["fluvio-future","fluvio-types"]
 [dependencies]
 tracing = "0.1.19"
 semver = "1.0.0"
-async-h1 = "2.1.2"
-http-types = "2.4.0"
 home = "0.5.3"
 sha2 = "0.10.0"
 hex = "0.4.2"
 tempdir = "0.3.7"
 thiserror = "1.0.20"
+http = { version = "0.2", default-features = false}
+isahc = { version = "1.7", default-features = false, features = ["static-curl"] }
+futures = { version = "0.3" , default-features = false, features = ["std", "io-compat"]}
 
-fluvio-package-index = { path = "../fluvio-package-index" }
+fluvio-package-index = { path = "../fluvio-package-index", features = ["http_agent"] }
 fluvio-types = { path = "../fluvio-types", optional = true }
-fluvio-future = { version = "0.4.0", features = ["fs", "io", "subscriber", "native2_tls"], optional = true }
+fluvio-future = { version = "0.4.0", features = ["fs", "io", "subscriber", "native2_tls", "fixture"], optional = true }

crates/fluvio-cli-common/src/error.rs

@@ -26,13 +26,9 @@ pub enum CliError {
 }
 
 #[derive(thiserror::Error, Debug)]
-#[error("Http Error: {}", inner)]
-pub struct HttpError {
-    pub(crate) inner: http_types::Error,
-}
-
-impl From<http_types::Error> for CliError {
-    fn from(inner: http_types::Error) -> Self {
-        Self::HttpError(HttpError { inner })
-    }
+pub enum HttpError {
+    #[error("Http transport error: {0}")]
+    TransportError(#[from] isahc::Error),
+    #[error("Invalid input: {0}")]
+    InvalidInput(String),
 }

crates/fluvio-cli-common/src/http.rs

@@ -1,50 +1,95 @@
-use std::io::{ErrorKind, Error as IoError};
-use async_h1::client;
-use http_types::{Error, Request, Response, StatusCode};
+use std::fmt::Debug;
+use http::uri::Scheme;
+use isahc::{AsyncBody, Request, Response};
 use tracing::{debug, error, instrument};
+use crate::error::HttpError;
 
 #[instrument(
     skip(request),
-    fields(url = %request.url())
+    fields(uri = %request.uri())
 )]
-pub async fn execute(request: Request) -> Result<Response, Error> {
+pub async fn execute<B: Into<AsyncBody> + Debug>(
+    request: Request<B>,
+) -> Result<Response<AsyncBody>, HttpError> {
     debug!(?request, "Executing http request:");
 
-    if request.url().scheme() != "https" {
+    if request.uri().scheme() != Some(&Scheme::HTTPS) {
         error!("CLI http executor only accepts https!");
-        return Err(IoError::new(ErrorKind::InvalidInput, "Must use https").into());
+        return Err(HttpError::InvalidInput("Must use https".to_string()));
     }
 
     let host = request
-        .url()
-        .host_str()
-        .ok_or_else(|| Error::from_str(StatusCode::BadRequest, "missing hostname"))?
+        .uri()
+        .host()
+        .ok_or_else(|| HttpError::InvalidInput("missing hostname".to_string()))?
         .to_string();
     debug!(%host, "Valid hostname:");
 
-    let addr: (&str, u16) = (&host, request.url().port_or_known_default().unwrap_or(443));
-    let tcp_stream = fluvio_future::net::TcpStream::connect(addr).await?;
-    debug!("Established TCP stream");
-    let tls_connector = create_tls().await;
-    debug!("Created TLS connector");
-    let tls_stream = tls_connector.connect(host, tcp_stream).await?;
-    debug!("Opened TLS stream from TCP stream");
-    let response = client::connect(tls_stream, request).await?;
+    let response = isahc::send_async(request).await?;
 
     debug!(?response, "Http response:");
     Ok(response)
 }
-async fn create_tls() -> fluvio_future::native_tls::TlsConnector {
-    fluvio_future::native_tls::TlsConnector::default()
+
+pub async fn read_to_end(response: Response<AsyncBody>) -> std::io::Result<Vec<u8>> {
+    use futures::io::AsyncReadExt;
+    let mut async_body = response.into_body();
+    let mut body = Vec::with_capacity(async_body.len().unwrap_or_default() as usize);
+    async_body.read_to_end(&mut body).await?;
+    Ok(body)
 }
 
 #[cfg(test)]
-#[fluvio_future::test]
-async fn test_web_request() {
-    use fluvio_index::HttpAgent;
-    use http_types::StatusCode;
-    let agent = HttpAgent::default();
-    let index = agent.request_index().expect("Failed to get request index");
-    let response = execute(index).await.expect("Failed to execute request");
-    assert_eq!(response.status(), StatusCode::Ok);
+mod tests {
+    use super::*;
+
+    #[fluvio_future::test]
+    async fn test_web_request() {
+        use fluvio_index::HttpAgent;
+        use http::StatusCode;
+        let agent = HttpAgent::default();
+        let index = agent.request_index().expect("Failed to get request index");
+        let response = execute(index).await.expect("Failed to execute request");
+        assert_eq!(response.status(), StatusCode::OK);
+    }
+
+    #[fluvio_future::test]
+    async fn test_https_required() {
+        //given
+        let request = Request::get("http://fluvio.io")
+            .body(())
+            .expect("valid url");
+
+        //when
+        let result = execute(request).await;
+
+        //then
+        assert!(matches!(result, Err(HttpError::InvalidInput(_))));
+    }
+
+    #[fluvio_future::test]
+    async fn test_read_empty_body() {
+        //given
+        let body = AsyncBody::empty();
+        let response = Response::new(body);
+
+        //when
+        let result = read_to_end(response).await.expect("read body");
+
+        //then
+        assert!(result.is_empty());
+    }
+
+    #[fluvio_future::test]
+    async fn test_read_body() {
+        //given
+        let body = AsyncBody::from("content");
+        let response = Response::new(body);
+
+        //when
+        let result = read_to_end(response).await.expect("read body");
+
+        //then
+        assert_eq!(&result, b"content");
+    }
 }

crates/fluvio-cli-common/src/install.rs

@@ -74,11 +74,12 @@ pub async fn fetch_latest_version<T>(
 ) -> Result<Version> {
     let request = agent.request_package(id)?;
     debug!(
-        url = %request.url(),
+        uri = %request.uri(),
         "Requesting package manifest:",
     );
     let response = crate::http::execute(request).await?;
-    let package = agent.package_from_response(response).await?;
+    let body = crate::http::read_to_end(response).await?;
+    let package = agent.package_from_response(&body).await?;
     let latest_release = package.latest_release_for_target(target, prerelease)?;
     debug!(release = ?latest_release, "Latest release for package:");
     if !latest_release.target_exists(target) {
@@ -103,7 +104,8 @@ pub async fn fetch_package_file(
         PackageVersion::Tag(tag) => {
             let tag_request = agent.request_tag(id, tag)?;
             let tag_response = crate::http::execute(tag_request).await?;
-            agent.tag_version_from_response(tag, tag_response).await?
+            let body = crate::http::read_to_end(tag_response).await?;
+            agent.tag_version_from_response(tag, &body).await?
         }
         _ => {
             return Err(
@@ -114,7 +116,7 @@ pub async fn fetch_package_file(
 
     // Download the package file from the package registry
     let download_request = agent.request_release_download(id, &version, target)?;
-    debug!(url = %download_request.url(), "Requesting package download:");
+    debug!(uri = %download_request.uri(), "Requesting package download:");
     let response = crate::http::execute(download_request).await?;
     if !response.status().is_success() {
         return Err(CliError::PackageNotFound {
@@ -123,12 +125,14 @@ pub async fn fetch_package_file(
             target: target.clone(),
         });
     }
-    let package_file = agent.release_from_response(response).await?;
+
+    let package_file = crate::http::read_to_end(response).await?;
 
     // Download the package checksum from the package registry
     let checksum_request = agent.request_release_checksum(id, &version, target)?;
     let response = crate::http::execute(checksum_request).await?;
-    let package_checksum = agent.checksum_from_response(response).await?;
+    let body = crate::http::read_to_end(response).await?;
+    let package_checksum = String::from_utf8_lossy(&body);
 
     if !verify_checksum(&package_file, &package_checksum) {
         return Err(fluvio_index::Error::ChecksumError.into());

crates/fluvio-cli/Cargo.toml

@@ -44,8 +44,6 @@ semver = "1.0.0"
 clap = { version = "3.1.8", features = ["std", "derive"], default-features = false }
 clap_complete = "3.1.1"
 dirs = "4.0.0"
-async-h1 = "2.1.2"
-http-types = "2.4.0"
 thiserror = "1.0.20"
 eyre = "0.6.1"
 color-eyre = { version = "0.6.0", default-features = false }

crates/fluvio-cli/src/error.rs

@@ -53,8 +53,8 @@ pub enum CliError {
     IndexError(#[from] fluvio_index::Error),
     #[error("Error finding executable")]
     WhichError(#[from] which::Error),
-    #[error(transparent)]
-    HttpError(#[from] HttpError),
+    #[error("Http Error: {0}")]
+    HttpError(#[from] fluvio_cli_common::error::HttpError),
     #[error("Package {package} is not published at version {version} for target {target}")]
     PackageNotFound {
         package: PackageId,
@@ -92,18 +92,6 @@ pub enum CliError {
     ConnectorNotFound(String),
 }
 
-#[derive(thiserror::Error, Debug)]
-#[error("Http Error: {}", inner)]
-pub struct HttpError {
-    pub(crate) inner: http_types::Error,
-}
-
-impl From<http_types::Error> for CliError {
-    fn from(inner: http_types::Error) -> Self {
-        Self::HttpError(HttpError { inner })
-    }
-}
-
 impl CliError {
     pub fn invalid_arg<M: Into<String>>(reason: M) -> Self {
         Self::InvalidArg(reason.into())

crates/fluvio-cli/src/install/update.rs

@@ -232,8 +232,9 @@ impl UpdateOpt {
 pub async fn check_update_required(agent: &HttpAgent) -> Result<bool> {
     debug!("Checking for a required CLI update");
     let request = agent.request_index()?;
-    let response = crate::http::execute(request).await?;
-    let index = agent.index_from_response(response).await?;
+    let response = fluvio_cli_common::http::execute(request).await?;
+    let body = fluvio_cli_common::http::read_to_end(response).await?;
+    let index = agent.index_from_response(&body).await?;
     Ok(index.metadata.update_required())
 }
 
@@ -252,8 +253,9 @@ pub async fn check_update_available(
     debug!(%target, %id, "Checking for an available (not required) CLI update:");
 
     let request = agent.request_package(&id)?;
-    let response = crate::http::execute(request).await?;
-    let package = agent.package_from_response(response).await?;
+    let response = fluvio_cli_common::http::execute(request).await?;
+    let body = fluvio_cli_common::http::read_to_end(response).await?;
+    let package = agent.package_from_response(&body).await?;
 
     let release = package.latest_release_for_target(&target, prerelease)?;
     let latest_version = release.version.clone();

crates/fluvio-cli/src/lib.rs

@@ -2,7 +2,6 @@
 //!
 //! CLI configurations at the top of the tree
 
-mod http;
 mod error;
 pub mod client;
 pub mod install;

crates/fluvio-extension-common/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "fluvio-extension-common"
-version = "0.8.1"
+version = "0.8.2"
 edition = "2021"
 authors = ["Fluvio Contributors <[email protected]>"]
 description = "Fluvio extension common"
@@ -27,4 +27,4 @@ thiserror = "1.0.20"
 semver = { version = "1.0.0", features = ["serde"] }
 
 fluvio = { version = "0.13.0", path = "../fluvio", optional = true }
-fluvio-package-index = { version = "0.6", path = "../fluvio-package-index" }
+fluvio-package-index = { path = "../fluvio-package-index" }

crates/fluvio-package-index/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "fluvio-package-index"
-version = "0.6.1"
+version = "0.7.0"
 authors = ["Fluvio Contributors <[email protected]>"]
 description = "Fluvio Package Index"
 repository = "https://github.com/infinyon/fluvio"
@@ -12,6 +12,9 @@ build = "build.rs"
 name = "fluvio_index"
 path = "src/lib.rs"
 
+[features]
+http_agent = ["http"]
+
 [dependencies]
 tracing = "0.1.21"
 thiserror = "1.0.21"
@@ -20,4 +23,4 @@ serde_json = "1.0.59"
 semver = { version = "1.0.0", features = ["serde"] }
 url = { version = "2.1.1", features = ["serde"] }
 lazy_static = "1.4.0"
-http-types = "2.6.0"
+http = { version = "0.2", default-features = false, optional = true}