* feat(sse): add SseErrorOf type alias

* feat: let endpoint argument be any type that impl reqwest::IntoUrl
  moved request to RestClient trait for IMHO better separation of concerns
  moved execute to Execute trait for better type inference
* refactor: simplify qparams parsing
* security: ensure credentials are not leaked to logs

Author: b4D8

src/client/mod.rs

@@ -19,11 +19,11 @@ use bytes::Buf;
 use crypto_common::InvalidLength;
 use hmac::{Hmac, Mac};
 #[cfg(feature = "logging")]
-use log::{Level, error, log_enabled, trace};
+use log::{error, trace};
 #[cfg(feature = "metrics")]
 use prometheus::{CounterVec, opts, register_counter_vec};
 use reqwest::{
-    Method, StatusCode,
+    IntoUrl, Method, StatusCode,
     header::{self, HeaderValue},
 };
 use serde::{Deserialize, Serialize, de::DeserializeOwned};
@@ -71,40 +71,26 @@ static CLIENT_REQUEST_DURATION: LazyLock<CounterVec> = LazyLock::new(|| {
 type HmacSha512 = Hmac<Sha512>;
 
 // -----------------------------------------------------------------------------
-// Request trait
-
-pub trait Request {
-    type Error;
+// RestClient trait
 
+pub trait RestClient<X>: Execute {
     fn request<T, U>(
         &self,
         method: &Method,
-        endpoint: &str,
+        endpoint: X,
         payload: &T,
-    ) -> impl Future<Output = Result<U, Self::Error>> + Send
+    ) -> impl Future<Output = Result<U, <Self as Execute>::Error>> + Send
     where
         T: ?Sized + Serialize + Debug + Send + Sync,
         U: DeserializeOwned + Debug + Send + Sync;
 
-    fn execute(
-        &self,
-        request: reqwest::Request,
-    ) -> impl Future<Output = Result<reqwest::Response, Self::Error>> + Send + 'static;
-}
-
-// -----------------------------------------------------------------------------
-// RestClient trait
-
-pub trait RestClient: Debug {
-    type Error;
-
-    fn get<T>(&self, endpoint: &str) -> impl Future<Output = Result<T, Self::Error>> + Send
+    fn get<T>(&self, endpoint: X) -> impl Future<Output = Result<T, Self::Error>> + Send
     where
         T: DeserializeOwned + Debug + Send + Sync;
 
     fn post<T, U>(
         &self,
-        endpoint: &str,
+        endpoint: X,
         payload: &T,
     ) -> impl Future<Output = Result<U, Self::Error>> + Send
     where
@@ -113,7 +99,7 @@ pub trait RestClient: Debug {
 
     fn put<T, U>(
         &self,
-        endpoint: &str,
+        endpoint: X,
         payload: &T,
     ) -> impl Future<Output = Result<U, Self::Error>> + Send
     where
@@ -122,20 +108,20 @@ pub trait RestClient: Debug {
 
     fn patch<T, U>(
         &self,
-        endpoint: &str,
+        endpoint: X,
         payload: &T,
     ) -> impl Future<Output = Result<U, Self::Error>> + Send
     where
         T: ?Sized + Serialize + Debug + Send + Sync,
         U: DeserializeOwned + Debug + Send + Sync;
 
-    fn delete(&self, endpoint: &str) -> impl Future<Output = Result<(), Self::Error>> + Send;
+    fn delete(&self, endpoint: X) -> impl Future<Output = Result<(), Self::Error>> + Send;
 }
 
 // -----------------------------------------------------------------------------
 // ClientCredentials structure
 
-#[derive(Serialize, Deserialize, PartialEq, Eq, Clone, Debug)]
+#[derive(Serialize, Deserialize, PartialEq, Eq, Clone)]
 #[serde(untagged)]
 pub enum Credentials {
     OAuth1 {
@@ -160,6 +146,17 @@ pub enum Credentials {
     },
 }
 
+impl fmt::Debug for Credentials {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        // NOTE: ensure secrets are not leaked in logs
+        match self {
+            Self::OAuth1 { .. } => f.write_str("OAuth1"),
+            Self::Basic { .. } => f.write_str("Basic"),
+            Self::Bearer { .. } => f.write_str("Bearer"),
+        }
+    }
+}
+
 impl Default for Credentials {
     fn default() -> Self {
         Self::OAuth1 {
@@ -335,19 +332,16 @@ impl OAuth1 for Signer {
 
         if !query.is_empty() {
             for qparam in query.split('&') {
-                let (k, v) = qparam.split_at(qparam.find('=').ok_or_else(|| {
+                let (k, v) = qparam.split_once("=").ok_or_else(|| {
                     SignerError::Parse(format!("failed to parse query parameter, {qparam}"))
-                })?);
-
-                if !params.contains_key(k) {
-                    params.insert(k.to_string(), v.strip_prefix('=').unwrap_or(v).to_owned());
-                }
+                })?;
+                params.entry(k.to_owned()).or_insert(v.to_owned());
             }
         }
 
         let mut params = params
             .iter()
-            .map(|(k, v)| format!("{}={}", k, urlencoding::encode(v)))
+            .map(|(k, v)| format!("{k}={}", urlencoding::encode(v)))
             .collect::<Vec<_>>();
 
         params.sort();
@@ -429,8 +423,6 @@ pub enum ClientError {
     Digest(SignerError),
     #[error("failed to serialize signature as header value, {0}")]
     SerializeHeaderValue(header::InvalidHeaderValue),
-    #[error("failed to parse url endpoint, {0}")]
-    ParseUrlEndpoint(url::ParseError),
 }
 
 // -----------------------------------------------------------------------------
@@ -446,69 +438,24 @@ pub struct Client {
     credentials: Option<Credentials>,
 }
 
-impl Request for Client {
-    type Error = ClientError;
+pub trait Execute {
+    type Error;
 
-    #[cfg_attr(feature = "tracing", tracing::instrument)]
-    async fn request<T, U>(
+    fn execute(
         &self,
-        method: &Method,
-        endpoint: &str,
-        payload: &T,
-    ) -> Result<U, Self::Error>
-    where
-        T: ?Sized + Serialize + Debug + Send + Sync,
-        U: DeserializeOwned + Debug + Send + Sync,
-    {
-        let buf = serde_json::to_vec(payload).map_err(ClientError::Serialize)?;
-        let mut request = reqwest::Request::new(
-            method.to_owned(),
-            endpoint.parse().map_err(ClientError::ParseUrlEndpoint)?,
-        );
-
-        request
-            .headers_mut()
-            .insert(header::CONTENT_TYPE, APPLICATION_JSON);
-
-        request
-            .headers_mut()
-            .insert(header::CONTENT_LENGTH, HeaderValue::from(buf.len()));
-
-        request.headers_mut().insert(header::ACCEPT_CHARSET, UTF8);
-
-        request
-            .headers_mut()
-            .insert(header::ACCEPT, APPLICATION_JSON);
-
-        *request.body_mut() = Some(buf.into());
-
-        let res = self.execute(request).await?;
-        let status = res.status();
-        let buf = res.bytes().await.map_err(ClientError::BodyAggregation)?;
-
-        #[cfg(feature = "logging")]
-        if log_enabled!(Level::Trace) {
-            trace!(
-                "received response, endpoint: '{endpoint}', method: '{method}', status: '{}'",
-                status.as_u16()
-            );
-        }
-
-        if !status.is_success() {
-            return Err(ClientError::StatusCode(
-                status,
-                serde_json::from_reader(buf.reader()).map_err(ClientError::Deserialize)?,
-            ));
-        }
+        request: reqwest::Request,
+    ) -> impl Future<Output = Result<reqwest::Response, Self::Error>> + Send + 'static;
+}
 
-        serde_json::from_reader(buf.reader()).map_err(ClientError::Deserialize)
-    }
+impl Execute for Client {
+    type Error = ClientError;
 
+    /// Executes the given HTTP `request`.
     #[cfg_attr(feature = "tracing", tracing::instrument(skip(self)))]
     fn execute(
         &self,
         mut request: reqwest::Request,
-    ) -> impl Future<Output = Result<reqwest::Response, Self::Error>> + 'static {
+    ) -> impl Future<Output = Result<reqwest::Response, Self::Error>> + Send + 'static {
         let client = self.clone();
 
         async move {
@@ -582,18 +529,63 @@ impl Request for Client {
     }
 }
 
-impl RestClient for Client {
-    type Error = ClientError;
+impl<X: IntoUrl + fmt::Debug + Send> RestClient<X> for Client {
+    #[cfg_attr(feature = "tracing", tracing::instrument)]
+    async fn request<T, U>(
+        &self,
+        method: &Method,
+        endpoint: X,
+        payload: &T,
+    ) -> Result<U, Self::Error>
+    where
+        T: ?Sized + Serialize + Debug + Send + Sync,
+        U: DeserializeOwned + Debug + Send + Sync,
+    {
+        let buf = serde_json::to_vec(payload).map_err(ClientError::Serialize)?;
+
+        let url = endpoint.into_url().map_err(ClientError::Request)?;
+
+        #[cfg(feature = "logging")]
+        let endpoint = url.as_str().to_owned();
+
+        let mut request = reqwest::Request::new(method.to_owned(), url);
+
+        let headers = request.headers_mut();
+        headers.insert(header::CONTENT_TYPE, APPLICATION_JSON);
+        headers.insert(header::CONTENT_LENGTH, HeaderValue::from(buf.len()));
+        headers.insert(header::ACCEPT_CHARSET, UTF8);
+        headers.insert(header::ACCEPT, APPLICATION_JSON);
+
+        *request.body_mut() = Some(buf.into());
+
+        let res = self.execute(request).await?;
+        let status = res.status();
+        let buf = res.bytes().await.map_err(ClientError::BodyAggregation)?;
+
+        #[cfg(feature = "logging")]
+        trace!(
+            "received response, endpoint: '{endpoint}', method: '{method}', status: '{}'",
+            status.as_u16()
+        );
+
+        if !status.is_success() {
+            return Err(ClientError::StatusCode(
+                status,
+                serde_json::from_reader(buf.reader()).map_err(ClientError::Deserialize)?,
+            ));
+        }
+
+        serde_json::from_reader(buf.reader()).map_err(ClientError::Deserialize)
+    }
 
     #[cfg_attr(feature = "tracing", tracing::instrument)]
-    async fn get<T>(&self, endpoint: &str) -> Result<T, Self::Error>
+    async fn get<T>(&self, endpoint: X) -> Result<T, Self::Error>
     where
         T: DeserializeOwned + Debug + Send + Sync,
     {
-        let mut req = reqwest::Request::new(
-            Method::GET,
-            endpoint.parse().map_err(ClientError::ParseUrlEndpoint)?,
-        );
+        let url = endpoint.into_url().map_err(ClientError::Request)?;
+
+        let mut req = reqwest::Request::new(Method::GET, url);
 
         req.headers_mut().insert(header::ACCEPT_CHARSET, UTF8);
 
@@ -614,7 +606,7 @@ impl RestClient for Client {
     }
 
     #[cfg_attr(feature = "tracing", tracing::instrument)]
-    async fn post<T, U>(&self, endpoint: &str, payload: &T) -> Result<U, Self::Error>
+    async fn post<T, U>(&self, endpoint: X, payload: &T) -> Result<U, Self::Error>
     where
         T: ?Sized + Serialize + Debug + Send + Sync,
         U: DeserializeOwned + Debug + Send + Sync,
@@ -623,7 +615,7 @@ impl RestClient for Client {
     }
 
     #[cfg_attr(feature = "tracing", tracing::instrument)]
-    async fn put<T, U>(&self, endpoint: &str, payload: &T) -> Result<U, Self::Error>
+    async fn put<T, U>(&self, endpoint: X, payload: &T) -> Result<U, Self::Error>
     where
         T: ?Sized + Serialize + Debug + Send + Sync,
         U: DeserializeOwned + Debug + Send + Sync,
@@ -632,7 +624,7 @@ impl RestClient for Client {
     }
 
     #[cfg_attr(feature = "tracing", tracing::instrument)]
-    async fn patch<T, U>(&self, endpoint: &str, payload: &T) -> Result<U, Self::Error>
+    async fn patch<T, U>(&self, endpoint: X, payload: &T) -> Result<U, Self::Error>
     where
         T: ?Sized + Serialize + Debug + Send + Sync,
         U: DeserializeOwned + Debug + Send + Sync,
@@ -641,11 +633,9 @@ impl RestClient for Client {
     }
 
     #[cfg_attr(feature = "tracing", tracing::instrument)]
-    async fn delete(&self, endpoint: &str) -> Result<(), Self::Error> {
-        let req = reqwest::Request::new(
-            Method::DELETE,
-            endpoint.parse().map_err(ClientError::ParseUrlEndpoint)?,
-        );
+    async fn delete(&self, endpoint: X) -> Result<(), Self::Error> {
+        let url = endpoint.into_url().map_err(ClientError::Request)?;
+        let req = reqwest::Request::new(Method::DELETE, url);
 
         let res = self.execute(req).await?;
         let status = res.status();