tedge connect c8y checks configured URL and connected URL are the same (#661)

* tedge connect c8y checks configured URL and connected URL are the same
- Close #600

Signed-off-by: Rina Fujino <18257209+rina23q@users.noreply.github.com>

5 files changed, 172 insertions, 3 deletions

diff --git a/crates/core/tedge/Cargo.toml b/crates/core/tedge/Cargo.toml
index 026a0a85..7af49afe 100644
--- a/crates/core/tedge/Cargo.toml
+++ b/crates/core/tedge/Cargo.toml
@@ -13,6 +13,7 @@ maintainer-scripts = "configuration/debian/tedge"
 
 [dependencies]
 anyhow = "1.0"
+base64 = "0.13"
 certificate = { path = "../../common/certificate" }
 chrono = "0.4"
 futures = "0.3"
@@ -22,6 +23,7 @@ rpassword = "5.0"
 rumqttc = "0.10"
 rustls = "0.20"
 serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
 structopt = "0.3"
 tedge_config = { path = "../../common/tedge_config" }
 tedge_users = { path = "../../common/tedge_users" }
@@ -38,6 +40,7 @@ mockito = "0.30"
 pem = "1.0"
 predicates = "2.0"
 tempfile = "3.2"
+test-case = "1.2"
 
 [features]
 integration-test = []
diff --git a/crates/core/tedge/src/cli/connect/command.rs b/crates/core/tedge/src/cli/connect/command.rs
index b2813963..3fa35c36 100644
--- a/crates/core/tedge/src/cli/connect/command.rs
+++ b/crates/core/tedge/src/cli/connect/command.rs
@@ -1,4 +1,6 @@
-use crate::{cli::connect::*, command::Command, system_services::*, ConfigError};
+use crate::{
+    cli::connect::jwt_token::*, cli::connect::*, command::Command, system_services::*, ConfigError,
+};
 use rumqttc::QoS::AtLeastOnce;
 use rumqttc::{Event, Incoming, MqttOptions, Outgoing, Packet};
 use std::path::{Path, PathBuf};
@@ -8,11 +10,11 @@ use tedge_config::*;
 use tedge_utils::paths::{create_directories, ok_if_not_found, DraftFile};
 use which::which;
 
-const DEFAULT_HOST: &str = "localhost";
+pub(crate) const DEFAULT_HOST: &str = "localhost";
 const WAIT_FOR_CHECK_SECONDS: u64 = 10;
 const C8Y_CONFIG_FILENAME: &str = "c8y-bridge.conf";
 const AZURE_CONFIG_FILENAME: &str = "az-bridge.conf";
-const RESPONSE_TIMEOUT: Duration = Duration::from_secs(10);
+pub(crate) const RESPONSE_TIMEOUT: Duration = Duration::from_secs(10);
 const MOSQUITTO_RESTART_TIMEOUT_SECONDS: u64 = 5;
 const MQTT_TLS_PORT: u16 = 8883;
 const TEDGE_BRIDGE_CONF_DIR_PATH: &str = "mosquitto-conf";
@@ -141,6 +143,10 @@ impl Command for ConnectCommand {
         }
 
         if let Cloud::C8y = self.cloud {
+            check_connected_c8y_tenant_as_configured(
+                &config.query_string(C8yUrlSetting)?,
+                config.query(MqttPortSetting)?.into(),
+            );
             enable_software_management(&bridge_config, self.service_manager.as_ref());
         }
 
@@ -498,6 +504,7 @@ fn enable_software_management(
         }
     }
 }
+
 // To preserve error chain and not discard other errors we need to ignore error here
 // (don't use '?' with the call to this function to preserve original error).
 fn clean_up(
@@ -567,3 +574,16 @@ fn get_common_mosquitto_config_file_path(
         .join(TEDGE_BRIDGE_CONF_DIR_PATH)
         .join(&common_mosquitto_config.config_file)
 }
+
+// To confirm the connected c8y tenant is the one that user configured.
+fn check_connected_c8y_tenant_as_configured(configured_url: &str, port: u16) {
+    match get_connected_c8y_url(port) {
+        Ok(url) if url == configured_url => {}
+        Ok(url) => println!(
+            "Warning: Connecting to {}, but the configured URL is {}.\n\
+            The device certificate has to be removed from the former tenant.\n",
+            url, configured_url
+        ),
+        Err(_) => println!("Failed to get the connected tenant URL from Cumulocity.\n"),
+    }
+}
diff --git a/crates/core/tedge/src/cli/connect/error.rs b/crates/core/tedge/src/cli/connect/error.rs
index 3d5cf94f..9a85682c 100644
--- a/crates/core/tedge/src/cli/connect/error.rs
+++ b/crates/core/tedge/src/cli/connect/error.rs
@@ -38,4 +38,9 @@ pub enum ConnectError {
 
     #[error("Device is not connected to {cloud} cloud")]
     DeviceNotConnected { cloud: String },
+
+    #[error(
+        "The JWT token received from Cumulocity is invalid.\nToken: {token}\nReason: {reason}"
+    )]
+    InvalidJWTToken { token: String, reason: String },
 }
diff --git a/crates/core/tedge/src/cli/connect/jwt_token.rs b/crates/core/tedge/src/cli/connect/jwt_token.rs
new file mode 100644
index 00000000..b81e26ed
--- /dev/null
+++ b/crates/core/tedge/src/cli/connect/jwt_token.rs
@@ -0,0 +1,140 @@
+use crate::cli::connect::{ConnectError, DEFAULT_HOST, RESPONSE_TIMEOUT};
+use rumqttc::QoS::AtLeastOnce;
+use rumqttc::{Event, Incoming, MqttOptions, Outgoing, Packet};
+
+pub(crate) fn get_connected_c8y_url(port: u16) -> Result<String, ConnectError> {
+    const C8Y_TOPIC_BUILTIN_JWT_TOKEN_UPSTREAM: &str = "c8y/s/uat";
+    const C8Y_TOPIC_BUILTIN_JWT_TOKEN_DOWNSTREAM: &str = "c8y/s/dat";
+    const CLIENT_ID: &str = "get_jwt_token_c8y";
+
+    let mut options = MqttOptions::new(CLIENT_ID, DEFAULT_HOST, port);
+    options.set_keep_alive(RESPONSE_TIMEOUT);
+
+    let (mut client, mut connection) = rumqttc::Client::new(options, 10);
+    let mut acknowledged = false;
+
+    client.subscribe(C8Y_TOPIC_BUILTIN_JWT_TOKEN_DOWNSTREAM, AtLeastOnce)?;
+
+    for event in connection.iter() {
+        match event {
+            Ok(Event::Incoming(Packet::SubAck(_))) => {
+                // We are ready to get the response, hence send the request
+                client.publish(C8Y_TOPIC_BUILTIN_JWT_TOKEN_UPSTREAM, AtLeastOnce, false, "")?;
+            }
+            Ok(Event::Incoming(Packet::PubAck(_))) => {
+                // The request has been sent
+                acknowledged = true;
+            }
+            Ok(Event::Incoming(Packet::Publish(response))) => {
+                // We got a response
+                let token = String::from_utf8(response.payload.to_vec()).unwrap();
+                let connected_url = decode_jwt_token(token.as_str())?;
+                return Ok(connected_url);
+            }
+            Ok(Event::Outgoing(Outgoing::PingReq)) => {
+                // No messages have been received for a while
+                println!("Local MQTT publish has timed out.");
+                break;
+            }
+            Ok(Event::Incoming(Incoming::Disconnect)) => {
+                eprintln!("ERROR: Disconnected");
+                break;
+            }
+            Err(err) => {
+                eprintln!("ERROR: {:?}", err);
+                break;
+            }
+            _ => {}
+        }
+    }
+
+    if acknowledged {
+        // The request has been sent but without a response
+        println!("\nThe request has been sent, however, no response.");
+        Err(ConnectError::TimeoutElapsedError)
+    } else {
+        // The request has not even been sent
+        println!("\nMake sure mosquitto is running.");
+        Err(ConnectError::TimeoutElapsedError)
+    }
+}
+
+pub(crate) fn decode_jwt_token(token: &str) -> Result<String, ConnectError> {
+    // JWT token format: <header>.<payload>.<signature>. Thus, we want only <paylaod>.
+    let payload = token
+        .split_terminator('.')
+        .nth(1)
+        .ok_or(ConnectError::InvalidJWTToken {
+            token: token.to_string(),
+            reason: "JWT token format must be <header>.<payload>.<signature>.".to_string(),
+        })?;
+
+    let decoded = base64::decode(payload).map_err(|_| ConnectError::InvalidJWTToken {
+        token: token.to_string(),
+        reason: "Cannot decode the payload of JWT token by Base64.".to_string(),
+    })?;
+
+    let json: serde_json::Value =
+        serde_json::from_slice(decoded.as_slice()).map_err(|_| ConnectError::InvalidJWTToken {
+            token: token.to_string(),
+            reason: "The payload of JWT token is not JSON.".to_string(),
+        })?;
+
+    let tenant_url = json["iss"].as_str().ok_or(ConnectError::InvalidJWTToken {
+        token: token.to_string(),
+        reason: "The JSON decoded from JWT token doesn't contain 'iss' field.".to_string(),
+    })?;
+
+    Ok(tenant_url.to_string())
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use test_case::test_case;
+
+    #[test]
+    fn check_decode_valid_jwt_token() {
+        let token = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJqdGkiOm51bGwsImlzcyI6InRlc3QuY3VtdWxvY2l0eS5jb20iLCJhdWQiOiJ0ZXN0LmN1bXVsb2NpdHkuY29tIiwic3ViIjoiZGV2aWNlX3Rlc3QwMDA1IiwidGNpIjoiZGV2aWNlX3Rva2VuX2NvbmZpZyIsImlhdCI6MTYzODQ0Mjk5NywibmJmIjoxNjM4NDQyOTk3LCJleHAiOjE2Mzg0NDY1OTcsInRmYSI6ZmFsc2UsInRlbiI6InQzMTcwNDgiLCJ4c3JmVG9rZW4iOiJLc2VBVUZBTGF1aUplZFFNR2ZzRiJ9.JUYtU9FVWlOWUPJXawFzKNiHD4HoEEWmvKdU1k9L2UF2ofRA2zAdcLH4mxaaspt4suyyZbPL6cS6c9MROG3YCsnqle2NSoYw8mxqncFECWsDS8lwCRTG4402iPTETfWpo9uXw2pFryBoJMAvNzt1qsXXn8EXSYxjzgj0YyxSANypm7PL1kMaprdLuUML_9Cwxf7Z6CRyWkZWWmnQ3lYgV5KMGW7HznkkqcmUCvuXKrHhVL5RkmzE1WyL4ndpGEPFEv9VYmEvFYA8wVHSuw5iVZIFp5lQldDdy_8U-N80xnf3fqZ6Q_wnVm8cga77vIgcf9zK5rSCdehvolM48uM4_w";
+        let expected_url = "test.cumulocity.com";
+        assert_eq!(decode_jwt_token(token).unwrap(), expected_url.to_string());
+    }
+
+    #[test_case(
+    "dGVzdC5jdW11bG9jaXR5LmNvbQ==",
+    "The JWT token received from Cumulocity is invalid.\n\
+    Token: dGVzdC5jdW11bG9jaXR5LmNvbQ==\n\
+    Reason: JWT token format must be <header>.<payload>.<signature>."
+    ; "not jwt token"
+    )]
+    #[test_case(
+    "aaa.bbb.ccc",
+    "The JWT token received from Cumulocity is invalid.\n\
+    Token: aaa.bbb.ccc\n\
+    Reason: Cannot decode the payload of JWT token by Base64."
+    ; "payload is not base64 encoded"
+    )]
+    #[test_case(
+    "aaa.dGVzdC5jdW11bG9jaXR5LmNvbQ==.ccc",
+    "The JWT token received from Cumulocity is invalid.\n\
+    Token: aaa.dGVzdC5jdW11bG9jaXR5LmNvbQ==.ccc\n\
+    Reason: The payload of JWT token is not JSON."
+    ; "payload is not json"
+    )]
+    #[test_case(
+    "aaa.eyJqdGkiOm51bGwsImF1ZCI6InRlc3QuY3VtdWxvY2l0eS5jb20iLCJzdWIiOiJkZXZpY2VfdGVzdDAwMDUiLCJ0Y2kiOiJkZXZpY2VfdG9rZW5fY29uZmlnIiwiaWF0IjoxNjM4NDQyOTk3LCJuYmYiOjE2Mzg0NDI5OTcsImV4cCI6MTYzODQ0NjU5NywidGZhIjpmYWxzZSwidGVuIjoidDMxNzA0OCIsInhzcmZUb2tlbiI6IktzZUFVRkFMYXVpSmVkUU1HZnNGIn0=.ccc",
+    "The JWT token received from Cumulocity is invalid.\n\
+    Token: aaa.eyJqdGkiOm51bGwsImF1ZCI6InRlc3QuY3VtdWxvY2l0eS5jb20iLCJzdWIiOiJkZXZpY2VfdGVzdDAwMDUiLCJ0Y2kiOiJkZXZpY2VfdG9rZW5fY29uZmlnIiwiaWF0IjoxNjM4NDQyOTk3LCJuYmYiOjE2Mzg0NDI5OTcsImV4cCI6MTYzODQ0NjU5NywidGZhIjpmYWxzZSwidGVuIjoidDMxNzA0OCIsInhzcmZUb2tlbiI6IktzZUFVRkFMYXVpSmVkUU1HZnNGIn0=.ccc\n\
+    Reason: The JSON decoded from JWT token doesn't contain 'iss' field."
+    ; "payload is json but not contains iss field"
+    )]
+    fn check_decode_invalid_jwt_token(input: &str, expected_error_msg: &str) {
+        match decode_jwt_token(input) {
+            Ok(_) => panic!("This test should result in an error"),
+            Err(err) => {
+                let error_msg = format!("{}", err);
+                assert_eq!(error_msg, expected_error_msg)
+            }
+        }
+    }
+}
diff --git a/crates/core/tedge/src/cli/connect/mod.rs b/crates/core/tedge/src/cli/connect/mod.rs
index f89361c8..edd233ba 100644
--- a/crates/core/tedge/src/cli/connect/mod.rs
+++ b/crates/core/tedge/src/cli/connect/mod.rs
@@ -10,3 +10,4 @@ mod cli;
 mod command;
 mod common_mosquitto_config;
 mod error;
+mod jwt_token;