feat: prototype aggregator client query system

Author: Alenar

internal/mithril-aggregator-client/src/client.rs

@@ -0,0 +1,166 @@
+use anyhow::{Context, anyhow};
+use reqwest::{Response, Url};
+
+use serde::de::DeserializeOwned;
+use slog::Logger;
+
+use crate::AggregatorClientResult;
+use crate::error::AggregatorClientError;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum QueryMethod {
+    Get,
+    Post,
+}
+
+// Todo: wasm compatibility
+#[async_trait::async_trait]
+pub(crate) trait AggregatorQuery {
+    type Response: DeserializeOwned;
+    type Body: serde::Serialize + Sized;
+
+    fn method() -> QueryMethod;
+
+    fn route(&self) -> String;
+
+    fn body(&self) -> Option<Self::Body> {
+        None
+    }
+
+    async fn handle_response(
+        &self,
+        context: QueryContext,
+    ) -> AggregatorClientResult<Self::Response>;
+}
+
+// internal to the crate
+pub(crate) struct QueryContext {
+    pub(crate) response: Response,
+    pub(crate) logger: Logger,
+}
+
+impl QueryContext {
+    pub(crate) async fn unhandled_status_code(self) -> AggregatorClientError {
+        AggregatorClientError::from_response(self.response).await
+    }
+}
+
+pub struct AggregatorClient {
+    aggregator_endpoint: Url,
+    client: reqwest::Client,
+    logger: Logger,
+}
+
+impl AggregatorClient {
+    pub async fn send<Q: AggregatorQuery>(&self, query: Q) -> AggregatorClientResult<Q::Response> {
+        let mut request_builder = match Q::method() {
+            QueryMethod::Get => self.client.get(self.join_aggregator_endpoint(&query.route())?),
+            QueryMethod::Post => self.client.post(self.join_aggregator_endpoint(&query.route())?),
+        };
+
+        if let Some(body) = query.body() {
+            request_builder = request_builder.json(&body);
+        }
+
+        match request_builder.send().await {
+            Ok(response) => {
+                // should we always warn?
+                if !response.status().is_server_error() {
+                    // todo: import code
+                    // self.warn_if_api_version_mismatch(&response);
+                }
+
+                let context = QueryContext {
+                    response,
+                    logger: self.logger.clone(),
+                };
+                query.handle_response(context).await
+            }
+            Err(err) => Err(AggregatorClientError::RemoteServerUnreachable(anyhow!(err))),
+        }
+    }
+
+    fn join_aggregator_endpoint(&self, endpoint: &str) -> AggregatorClientResult<Url> {
+        self.aggregator_endpoint
+            .join(endpoint)
+            .with_context(|| {
+                format!(
+                    "Invalid url when joining given endpoint, '{endpoint}', to aggregator url '{}'",
+                    self.aggregator_endpoint
+                )
+            })
+            .map_err(AggregatorClientError::InvalidEndpoint)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use http::StatusCode;
+    use httpmock::MockServer;
+
+    use crate::test::TestLogger;
+
+    use super::*;
+
+    #[derive(Debug, Eq, PartialEq, serde::Deserialize)]
+    struct TestResponse {
+        foo: String,
+        bar: i32,
+    }
+
+    struct TestGetQuery;
+
+    #[async_trait::async_trait]
+    impl AggregatorQuery for TestGetQuery {
+        type Response = TestResponse;
+        type Body = ();
+
+        fn method() -> QueryMethod {
+            QueryMethod::Get
+        }
+
+        fn route(&self) -> String {
+            "/dummy-route".to_string()
+        }
+
+        async fn handle_response(
+            &self,
+            context: QueryContext,
+        ) -> AggregatorClientResult<Self::Response> {
+            match context.response.status() {
+                StatusCode::OK => context
+                    .response
+                    .json::<TestResponse>()
+                    .await
+                    .map_err(|err| AggregatorClientError::JsonParseFailed(anyhow!(err))),
+                _ => Err(context.unhandled_status_code().await),
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn test_minimal_query() {
+        let server = MockServer::start();
+        server.mock(|when, then| {
+            when.method(httpmock::Method::GET).path(TestGetQuery.route());
+            then.status(200).body(r#"{"foo": "bar", "bar": 123}"#);
+        });
+
+        let aggregator_endpoint = Url::parse(&server.url("/")).unwrap();
+        let client = AggregatorClient {
+            aggregator_endpoint,
+            client: reqwest::Client::new(),
+            logger: TestLogger::stdout(),
+        };
+
+        let response = client.send(TestGetQuery).await.unwrap();
+
+        assert_eq!(
+            response,
+            TestResponse {
+                foo: "bar".to_string(),
+                bar: 123,
+            }
+        )
+    }
+}

internal/mithril-aggregator-client/src/error.rs

@@ -0,0 +1,89 @@
+use anyhow::anyhow;
+use reqwest::{Response, StatusCode, header};
+use thiserror::Error;
+
+use mithril_common::StdError;
+use mithril_common::entities::{ClientError, ServerError};
+
+use crate::JSON_CONTENT_TYPE;
+
+/// Error structure for the Aggregator Client.
+#[derive(Error, Debug)]
+pub enum AggregatorClientError {
+    /// The aggregator host has returned a technical error.
+    #[error("remote server technical error")]
+    RemoteServerTechnical(#[source] StdError),
+
+    /// The aggregator host responded it cannot fulfill our request.
+    #[error("remote server logical error")]
+    RemoteServerLogical(#[source] StdError),
+
+    /// Could not reach aggregator.
+    #[error("Remote server unreachable")]
+    RemoteServerUnreachable(#[source] StdError),
+
+    /// Unhandled status code
+    #[error("Unhandled status code: {0}, response text: {1}")]
+    UnhandledStatusCode(StatusCode, String),
+
+    /// Could not parse response.
+    #[error("Json parsing failed")]
+    JsonParseFailed(#[source] StdError),
+
+    /// Failed to join the query endpoint to the aggregator url
+    #[error("Invalid endpoint")]
+    InvalidEndpoint(#[source] StdError),
+}
+
+impl AggregatorClientError {
+    /// Create an `AggregatorClientError` from a response.
+    ///
+    /// This method is meant to be used after handling domain-specific cases leaving only
+    /// 4xx or 5xx status codes.
+    /// Otherwise, it will return an `UnhandledStatusCode` error.
+    pub async fn from_response(response: Response) -> Self {
+        let error_code = response.status();
+
+        if error_code.is_client_error() {
+            let root_cause = Self::get_root_cause(response).await;
+            Self::RemoteServerLogical(anyhow!(root_cause))
+        } else if error_code.is_server_error() {
+            let root_cause = Self::get_root_cause(response).await;
+            Self::RemoteServerTechnical(anyhow!(root_cause))
+        } else {
+            let response_text = response.text().await.unwrap_or_default();
+            Self::UnhandledStatusCode(error_code, response_text)
+        }
+    }
+
+    async fn get_root_cause(response: Response) -> String {
+        let error_code = response.status();
+        let canonical_reason = error_code.canonical_reason().unwrap_or_default().to_lowercase();
+        let is_json = response
+            .headers()
+            .get(header::CONTENT_TYPE)
+            .is_some_and(|ct| JSON_CONTENT_TYPE == ct);
+
+        if is_json {
+            let json_value: serde_json::Value = response.json().await.unwrap_or_default();
+
+            if let Ok(client_error) = serde_json::from_value::<ClientError>(json_value.clone()) {
+                format!(
+                    "{}: {}: {}",
+                    canonical_reason, client_error.label, client_error.message
+                )
+            } else if let Ok(server_error) =
+                serde_json::from_value::<ServerError>(json_value.clone())
+            {
+                format!("{}: {}", canonical_reason, server_error.message)
+            } else if json_value.is_null() {
+                canonical_reason.to_string()
+            } else {
+                format!("{canonical_reason}: {json_value}")
+            }
+        } else {
+            let response_text = response.text().await.unwrap_or_default();
+            format!("{canonical_reason}: {response_text}")
+        }
+    }
+}

internal/mithril-aggregator-client/src/lib.rs

@@ -2,5 +2,16 @@
 //! This crate provides a client to request data from a Mithril Aggregator.
 //!
 
+mod client;
+mod error;
+pub mod query;
 #[cfg(test)]
-mod test;
+mod test;
+
+pub use client::AggregatorClient;
+pub use error::AggregatorClientError;
+
+pub(crate) const JSON_CONTENT_TYPE: reqwest::header::HeaderValue =
+    reqwest::header::HeaderValue::from_static("application/json");
+
+pub type AggregatorClientResult<T> = Result<T, error::AggregatorClientError>;

internal/mithril-aggregator-client/src/query/certificate_details.rs

@@ -0,0 +1,49 @@
+use anyhow::anyhow;
+use reqwest::StatusCode;
+use slog::debug;
+
+use mithril_common::messages::CertificateMessage;
+
+use crate::AggregatorClientResult;
+use crate::client::{AggregatorQuery, QueryContext, QueryMethod};
+use crate::error::AggregatorClientError;
+
+pub struct CertificateDetailsQuery {
+    hash: String,
+}
+
+impl CertificateDetailsQuery {
+    pub fn new(hash: String) -> Self {
+        Self { hash }
+    }
+}
+
+#[async_trait::async_trait]
+impl AggregatorQuery for CertificateDetailsQuery {
+    type Response = Option<CertificateMessage>;
+    type Body = ();
+
+    fn method() -> QueryMethod {
+        QueryMethod::Get
+    }
+
+    fn route(&self) -> String {
+        format!("certificate/{}", self.hash)
+    }
+
+    async fn handle_response(
+        &self,
+        context: QueryContext,
+    ) -> AggregatorClientResult<Self::Response> {
+        debug!(context.logger, "Retrieve certificate details"; "certificate_hash" => %self.hash);
+
+        match context.response.status() {
+            StatusCode::OK => match context.response.json::<CertificateMessage>().await {
+                Ok(message) => Ok(Some(message)),
+                Err(err) => Err(AggregatorClientError::JsonParseFailed(anyhow!(err))),
+            },
+            StatusCode::NOT_FOUND => Ok(None),
+            _ => Err(context.unhandled_status_code().await),
+        }
+    }
+}

internal/mithril-aggregator-client/src/query/mod.rs

@@ -0,0 +1,3 @@
+mod certificate_details;
+
+pub use certificate_details::*;