path: root/crates/core/tedge_mapper/src/sm_c8y_mapper/http_proxy.rs

use crate::sm_c8y_mapper::error::SMCumulocityMapperError;
use crate::sm_c8y_mapper::json_c8y::{
    C8yCreateEvent, C8yManagedObject, C8yUpdateSoftwareListResponse, InternalIdResponse,
};
use crate::sm_c8y_mapper::mapper::SmartRestLogEvent;
use async_trait::async_trait;
use c8y_smartrest::smartrest_deserializer::SmartRestJwtResponse;
use chrono::{DateTime, Local};
use mqtt_client::{Client, MqttClient, Topic};
use reqwest::Url;
use std::time::Duration;
use tedge_config::{C8yUrlSetting, ConfigSettingAccessorStringExt, DeviceIdSetting, TEdgeConfig};
use tracing::{error, info, instrument};

const RETRY_TIMEOUT_SECS: u64 = 60;

/// An HttpProxy handles http requests to C8y on behalf of the device.
#[async_trait]
pub trait C8YHttpProxy {
    async fn init(&mut self) -> Result<(), SMCumulocityMapperError>;

    fn url_is_in_my_tenant_domain(&self, url: &str) -> bool;

    async fn get_jwt_token(&self) -> Result<SmartRestJwtResponse, SMCumulocityMapperError>;

    async fn send_software_list_http(
        &self,
        c8y_software_list: &C8yUpdateSoftwareListResponse,
    ) -> Result<(), SMCumulocityMapperError>;

    async fn upload_log_binary(&self, log_content: &str)
        -> Result<String, SMCumulocityMapperError>;
}

/// Define a C8y endpoint
pub struct C8yEndPoint {
    c8y_host: String,
    device_id: String,
    c8y_internal_id: String,
}

impl C8yEndPoint {
    fn new(c8y_host: &str, device_id: &str, c8y_internal_id: &str) -> C8yEndPoint {
        C8yEndPoint {
            c8y_host: c8y_host.into(),
            device_id: device_id.into(),
            c8y_internal_id: c8y_internal_id.into(),
        }
    }

    fn get_url_for_sw_list(&self) -> String {
        let mut url_update_swlist = String::new();
        url_update_swlist.push_str("https://");
        url_update_swlist.push_str(&self.c8y_host);
        url_update_swlist.push_str("/inventory/managedObjects/");
        url_update_swlist.push_str(&self.c8y_internal_id);

        url_update_swlist
    }

    fn get_url_for_get_id(&self) -> String {
        let mut url_get_id = String::new();
        url_get_id.push_str("https://");
        url_get_id.push_str(&self.c8y_host);
        url_get_id.push_str("/identity/externalIds/c8y_Serial/");
        url_get_id.push_str(&self.device_id);

        url_get_id
    }

    fn get_url_for_create_event(&self) -> String {
        let mut url_create_event = String::new();
        url_create_event.push_str("https://");
        url_create_event.push_str(&self.c8y_host);
        url_create_event.push_str("/event/events/");

        url_create_event
    }

    fn get_url_for_event_binary_upload(&self, event_id: &str) -> String {
        let mut url_event_binary = self.get_url_for_create_event();
        url_event_binary.push_str(event_id);
        url_event_binary.push_str("/binaries");

        url_event_binary
    }

    fn url_is_in_my_tenant_domain(&self, url: &str) -> bool {
        // c8y URL may contain either `Tenant Name` or Tenant Id` so they can be one of following options:
        // * <tenant_name>.<domain> eg: sample.c8y.io
        // * <tenant_id>.<domain> eg: t12345.c8y.io
        // These URLs may be both equivalent and point to the same tenant.
        // We are going to remove that and only check if the domain is the same.
        let tenant_uri = &self.c8y_host;
        let url_host = match Url::parse(url) {
            Ok(url) => match url.host() {
                Some(host) => host.to_string(),
                None => return false,
            },
            Err(_err) => {
                return false;
            }
        };

        let url_domain = url_host.splitn(2, '.').collect::<Vec<&str>>();
        let tenant_domain = tenant_uri.splitn(2, '.').collect::<Vec<&str>>();
        if url_domain.get(1) == tenant_domain.get(1) {
            return true;
        }
        false
    }
}

/// An HttpProxy that uses MQTT to retrieve JWT tokens and authenticate the device
///
/// - Keep the connection info to c8y and the internal Id of the device
/// - Handle JWT requests
pub struct JwtAuthHttpProxy {
    mqtt_con: Client,
    http_con: reqwest::Client,
    end_point: C8yEndPoint,
}

impl JwtAuthHttpProxy {
    pub fn new(
        mqtt_con: Client,
        http_con: reqwest::Client,
        c8y_host: &str,
        device_id: &str,
    ) -> JwtAuthHttpProxy {
        JwtAuthHttpProxy {
            mqtt_con,
            http_con,
            end_point: C8yEndPoint {
                c8y_host: c8y_host.into(),
                device_id: device_id.into(),
                c8y_internal_id: "".into(),
            },
        }
    }

    pub fn try_new(
        mqtt_con: Client,
        tedge_config: &TEdgeConfig,
    ) -> Result<JwtAuthHttpProxy, SMCumulocityMapperError> {
        let c8y_host = tedge_config.query_string(C8yUrlSetting)?;
        let device_id = tedge_config.query_string(DeviceIdSetting)?;
        let http_con = reqwest::ClientBuilder::new().build()?;
        Ok(JwtAuthHttpProxy::new(
            mqtt_con, http_con, &c8y_host, &device_id,
        ))
    }

    async fn try_get_and_set_internal_id(&mut self) -> Result<(), SMCumulocityMapperError> {
        let token = self.get_jwt_token().await?;
        let url_get_id = self.end_point.get_url_for_get_id();

        self.end_point.c8y_internal_id = self
            .try_get_internal_id(&url_get_id, &token.token())
            .await?;

        Ok(())
    }

    async fn try_get_internal_id(
        &self,
        url_get_id: &str,
        token: &str,
    ) -> Result<String, SMCumulocityMapperError> {
        let internal_id = self
            .http_con
            .get(url_get_id)
            .bearer_auth(token)
            .send()
            .await?;
        let internal_id_response = internal_id.json::<InternalIdResponse>().await?;

        let internal_id = internal_id_response.id();
        Ok(internal_id)
    }

    /// Make a POST request to /event/events and return the event id from response body.
    /// The event id is used to upload the binary.
    fn create_log_event(&self) -> C8yCreateEvent {
        let local: DateTime<Local> = Local::now();

        let c8y_managed_object = C8yManagedObject {
            id: self.end_point.c8y_internal_id.clone(),
        };

        C8yCreateEvent::new(
            c8y_managed_object.to_owned(),
            "c8y_Logfile",
            &local.format("%Y-%m-%dT%H:%M:%SZ").to_string(),
            "software-management",
        )
    }

    async fn get_event_id(
        &self,
        c8y_event: C8yCreateEvent,
    ) -> Result<String, SMCumulocityMapperError> {
        let token = self.get_jwt_token().await?;
        let create_event_url = self.end_point.get_url_for_create_event();

        let request = self
            .http_con
            .post(create_event_url)
            .json(&c8y_event)
            .bearer_auth(token.token())
            .header("Accept", "application/json")
            .timeout(Duration::from_millis(10000))
            .build()?;

        let response = self.http_con.execute(request).await?;
        let event_response_body = response.json::<SmartRestLogEvent>().await?;

        Ok(event_response_body.id)
    }
}

#[async_trait]
impl C8YHttpProxy for JwtAuthHttpProxy {
    fn url_is_in_my_tenant_domain(&self, url: &str) -> bool {
        self.end_point.url_is_in_my_tenant_domain(url)
    }

    #[instrument(skip(self), name = "init"<