feat(signature): async update certificate and validate (#973)

* feat(signature): async update certificate and validate

Author: paologallinaharbur

agent-control/src/agent_control/config.rs

@@ -1,5 +1,8 @@
 use super::http_server::config::ServerConfig;
-use crate::agent_control::defaults::{default_capabilities, AGENT_CONTROL_ID};
+use crate::agent_control::agent_control_fqn;
+use crate::agent_control::defaults::{
+    default_capabilities, default_sub_agent_custom_capabilities, AGENT_CONTROL_ID,
+};
 use crate::http::proxy::ProxyConfig;
 use crate::logging::config::LoggingConfig;
 use crate::opamp::auth::config::AuthConfig;
@@ -8,6 +11,7 @@ use crate::values::yaml_config::YAMLConfig;
 use http::HeaderMap;
 #[cfg(feature = "k8s")]
 use kube::api::TypeMeta;
+use opamp_client::opamp::proto::CustomCapabilities;
 use opamp_client::operation::capabilities::Capabilities;
 use serde::{Deserialize, Deserializer, Serialize};
 use std::ops::Deref;
@@ -372,6 +376,16 @@ impl AgentTypeFQN {
         //TODO: We should move this to EffectiveAgent
         default_capabilities()
     }
+
+    pub(crate) fn get_custom_capabilities(&self) -> Option<CustomCapabilities> {
+        //TODO: We should move this to EffectiveAgent
+        if self.eq(&agent_control_fqn()) {
+            // Agent_Control does not have custom capabilities for now
+            return None;
+        }
+
+        Some(default_sub_agent_custom_capabilities())
+    }
 }
 
 #[cfg(test)]

agent-control/src/agent_control/error.rs

@@ -99,4 +99,7 @@ pub enum AgentError {
 
     #[error("parsing remote config into YAMLConfig: `{0}`")]
     YAMLConfigError(#[from] YAMLConfigError),
+
+    #[error("failed to initialize the signature validator: `{0}`")]
+    InitialiseSignatureValidator(String),
 }

agent-control/src/agent_control/run/k8s.rs

@@ -14,6 +14,7 @@ use crate::opamp::effective_config::loader::DefaultEffectiveConfigLoaderBuilder;
 use crate::opamp::instance_id::getter::InstanceIDWithIdentifiersGetter;
 use crate::opamp::instance_id::Identifiers;
 use crate::opamp::operations::build_opamp_with_channel;
+use crate::opamp::remote_config::validators::signature::SignatureValidator;
 use crate::sub_agent::effective_agents_assembler::LocalEffectiveAgentsAssembler;
 use crate::{
     agent_control::error::AgentError,
@@ -91,6 +92,11 @@ impl AgentControlRunner {
 
         let hash_repository = Arc::new(HashRepositoryConfigMap::new(k8s_store.clone()));
 
+        let signature_validator = Arc::new(
+            SignatureValidator::try_new()
+                .map_err(|e| AgentError::InitialiseSignatureValidator(e.to_string()))?,
+        );
+
         info!("Creating the k8s sub_agent builder");
         let sub_agent_builder = K8sSubAgentBuilder::new(
             opamp_client_builder.as_ref(),
@@ -100,6 +106,7 @@ impl AgentControlRunner {
             agents_assembler,
             self.k8s_config.clone(),
             yaml_config_repository.clone(),
+            signature_validator,
         );
 
         let additional_identifying_attributes =

agent-control/src/agent_control/run/on_host.rs

@@ -12,6 +12,7 @@ use crate::opamp::effective_config::loader::DefaultEffectiveConfigLoaderBuilder;
 use crate::opamp::instance_id::getter::InstanceIDWithIdentifiersGetter;
 use crate::opamp::instance_id::{Identifiers, Storer};
 use crate::opamp::operations::build_opamp_with_channel;
+use crate::opamp::remote_config::validators::signature::SignatureValidator;
 use crate::sub_agent::effective_agents_assembler::LocalEffectiveAgentsAssembler;
 use crate::{agent_control::error::AgentError, opamp::client_builder::DefaultOpAMPClientBuilder};
 use crate::{
@@ -104,13 +105,19 @@ impl AgentControlRunner {
             template_renderer,
         ));
 
+        let signature_validator = Arc::new(
+            SignatureValidator::try_new()
+                .map_err(|e| AgentError::InitialiseSignatureValidator(e.to_string()))?,
+        );
+
         let sub_agent_builder = OnHostSubAgentBuilder::new(
             opamp_client_builder.as_ref(),
             &instance_id_getter,
             sub_agent_hash_repository,
             agents_assembler,
             self.base_paths.log_dir.join(SUB_AGENT_DIR),
             yaml_config_repository.clone(),
+            signature_validator,
         );
 
         let (maybe_client, maybe_sa_opamp_consumer) = opamp_client_builder

agent-control/src/opamp/client_builder.rs

@@ -1,3 +1,10 @@
+use super::callbacks::AgentCallbacks;
+use super::effective_config::loader::EffectiveConfigLoaderBuilder;
+use super::http::builder::{HttpClientBuilder, HttpClientBuilderError};
+use crate::agent_control::config::AgentID;
+use crate::event::channel::EventPublisher;
+use crate::event::OpAMPEvent;
+use crate::opamp::instance_id;
 use opamp_client::http::{NotStartedHttpClient, StartedHttpClient};
 use opamp_client::operation::callbacks::Callbacks;
 use opamp_client::operation::settings::StartSettings;
@@ -6,15 +13,6 @@ use std::time::Duration;
 use thiserror::Error;
 use tracing::{error, info};
 
-use crate::agent_control::config::AgentID;
-use crate::event::channel::EventPublisher;
-use crate::event::OpAMPEvent;
-use crate::opamp::instance_id;
-
-use super::callbacks::AgentCallbacks;
-use super::effective_config::loader::EffectiveConfigLoaderBuilder;
-use super::http::builder::{HttpClientBuilder, HttpClientBuilderError};
-
 /// Default poll interval for the OpAMP http managed client
 pub const DEFAULT_POLL_INTERVAL: Duration = Duration::from_secs(30);
 

agent-control/src/opamp/operations.rs

@@ -1,14 +1,10 @@
-use std::collections::HashMap;
-use std::ops::Not;
-
 use super::instance_id::InstanceID;
 use super::{
     client_builder::{OpAMPClientBuilder, OpAMPClientBuilderError},
     instance_id::getter::InstanceIDGetter,
 };
 use crate::agent_control::defaults::{
-    default_sub_agent_custom_capabilities, OPAMP_SERVICE_NAME, OPAMP_SERVICE_NAMESPACE,
-    PARENT_AGENT_ID_ATTRIBUTE_KEY,
+    OPAMP_SERVICE_NAME, OPAMP_SERVICE_NAMESPACE, PARENT_AGENT_ID_ATTRIBUTE_KEY,
 };
 use crate::{
     agent_control::config::{AgentID, AgentTypeFQN},
@@ -25,6 +21,7 @@ use opamp_client::{
     },
     StartedClient,
 };
+use std::collections::HashMap;
 use tracing::info;
 
 pub fn build_sub_agent_opamp<CB, OB, IG>(
@@ -74,7 +71,6 @@ where
     let (opamp_publisher, opamp_consumer) = pub_sub();
     let start_settings = start_settings(
         instance_id_getter.get(&agent_id)?,
-        &agent_id,
         agent_type,
         additional_identifying_attributes,
         non_identifying_attributes,
@@ -88,7 +84,6 @@ where
 /// Builds the OpAMP StartSettings corresponding to the provided arguments for any sub agent and agent control.
 pub fn start_settings(
     instance_id: InstanceID,
-    agent_id: &AgentID,
     agent_fqn: &AgentTypeFQN,
     additional_identifying_attributes: HashMap<String, DescriptionValueType>,
     non_identifying_attributes: HashMap<String, DescriptionValueType>,
@@ -103,16 +98,10 @@ pub fn start_settings(
 
     identifying_attributes.extend(additional_identifying_attributes);
 
-    // Agent control does not have custom capabilities
-    let custom_capabilities = agent_id
-        .is_agent_control_id()
-        .not()
-        .then(default_sub_agent_custom_capabilities);
-
     StartSettings {
         instance_id: instance_id.into(),
         capabilities: agent_fqn.get_capabilities(),
-        custom_capabilities,
+        custom_capabilities: agent_fqn.get_custom_capabilities(),
         agent_description: AgentDescription {
             identifying_attributes,
             non_identifying_attributes,

agent-control/src/opamp/remote_config.rs

@@ -8,6 +8,7 @@ use thiserror::Error;
 pub mod hash;
 pub mod report;
 pub mod signature;
+pub mod validators;
 
 /// This structure represents the remote configuration that we would retrieve from a server via OpAMP.
 /// Contains identifying metadata and the actual configuration values
@@ -73,6 +74,10 @@ impl RemoteConfig {
             )),
         }
     }
+
+    pub fn get_signature(&self) -> &Option<Signature> {
+        &self.signature
+    }
 }
 
 impl ConfigurationMap {

agent-control/src/opamp/remote_config/validators.rs

@@ -0,0 +1,3 @@
+mod certificate_fetcher;
+pub mod regexes;
+pub mod signature;

agent-control/src/opamp/remote_config/validators/certificate_fetcher.rs

@@ -0,0 +1,43 @@
+use crate::opamp::remote_config::validators::signature::Certificate;
+use std::sync::Mutex;
+use std::time::SystemTime;
+use thiserror::Error;
+use tracing::debug;
+use tracing::log::error;
+
+#[derive(Error, Debug)]
+pub enum CertificateFetchError {}
+
+/// The CertificateFetcher is responsible for returning the certificate.
+pub struct CertificateFetcher {
+    certificate: Mutex<Certificate>,
+}
+
+impl CertificateFetcher {
+    pub fn try_new() -> Result<Self, CertificateFetchError> {
+        CertificateFetcher::fetch_certificate().map(|certificate| Self {
+            certificate: Mutex::new(certificate),
+        })
+    }
+
+    pub fn get_certificate(&self) -> Result<Certificate, CertificateFetchError> {
+        let mut certificate = self
+            .certificate
+            .lock()
+            .expect("failed to acquire certificate lock");
+
+        debug!("Updating Certificate");
+        CertificateFetcher::fetch_certificate()
+            .inspect_err(|e| error!("error fetching certificate: {:?}", e))
+            .map(|c| {
+                *certificate = c;
+            })?;
+
+        Ok(certificate.clone())
+    }
+
+    //TODO this is a stub
+    fn fetch_certificate() -> Result<Certificate, CertificateFetchError> {
+        Ok(Certificate)
+    }
+}