Introduce 'shovels enable_tls_peer_verification_for_all_{source,destination}_uris'

Author: michaelklishin

.config/nextest.toml

@@ -51,3 +51,18 @@ test-group = 'sequential'
 filter = 'test(deprecated_features)'
 priority = 10
 test-group = 'sequential'
+
+[[profile.default.overrides]]
+filter = 'binary(federation_upstream_uri_modification_tests)'
+priority = 25
+test-group = 'sequential'
+
+[[profile.default.overrides]]
+filter = 'binary(shovel_source_uri_modification_tests)'
+priority = 24
+test-group = 'sequential'
+
+[[profile.default.overrides]]
+filter = 'binary(shovel_destination_uri_modification_tests)'
+priority = 23
+test-group = 'sequential'

src/cli.rs

@@ -2959,7 +2959,7 @@ pub fn get_subcommands(pre_flight_settings: PreFlightSettings) -> [Command; 1] {
         )].map(|cmd| cmd.infer_long_args(pre_flight_settings.infer_long_options))
 }
 
-pub fn shovel_subcommands(pre_flight_settings: PreFlightSettings) -> [Command; 7] {
+pub fn shovel_subcommands(pre_flight_settings: PreFlightSettings) -> [Command; 9] {
     let idempotently_arg = Arg::new("idempotently")
         .long("idempotently")
         .value_parser(value_parser!(bool))
@@ -3138,6 +3138,76 @@ pub fn shovel_subcommands(pre_flight_settings: PreFlightSettings) -> [Command; 7
         .after_help(color_print::cformat!(
             r#"<bold>Doc guides</bold>:
 
+ * {}
+ * {}
+ * {}"#,
+            SHOVEL_GUIDE_URL,
+            TLS_GUIDE_URL,
+            "https://www.rabbitmq.com/docs/shovel#tls-connections"
+        ));
+
+    let enable_tls_peer_verification_source_cmd = Command::new("enable_tls_peer_verification_for_all_source_uris")
+        .about("Enables TLS peer verification for all shovel source URIs with provided [RabbitMQ node-local] certificate paths.")
+        .long_about("Enables TLS peer verification for all shovel source URIs by updating their 'verify' parameter and adding [RabbitMQ node-local] certificate and private key file paths.")
+        .arg(
+            Arg::new("node_local_ca_certificate_bundle_path")
+                .long("node-local-ca-certificate-bundle-path")
+                .help("Path to the CA certificate bundle file on the target RabbitMQ node(s)")
+                .required(true)
+                .value_name("PATH")
+        )
+        .arg(
+            Arg::new("node_local_client_certificate_file_path")
+                .long("node-local-client-certificate-file-path")
+                .help("Path to the client certificate file on the target RabbitMQ node(s)")
+                .required(true)
+                .value_name("PATH")
+        )
+        .arg(
+            Arg::new("node_local_client_private_key_file_path")
+                .long("node-local-client-private-key-file-path")
+                .help("Path to the client private key file on the target RabbitMQ node(s)")
+                .required(true)
+                .value_name("PATH")
+        )
+        .after_help(color_print::cformat!(
+            r#"<bold>Doc guides</bold>:
+
+ * {}
+ * {}
+ * {}"#,
+            SHOVEL_GUIDE_URL,
+            TLS_GUIDE_URL,
+            "https://www.rabbitmq.com/docs/shovel#tls-connections"
+        ));
+
+    let enable_tls_peer_verification_dest_cmd = Command::new("enable_tls_peer_verification_for_all_destination_uris")
+        .about("Enables TLS peer verification for all shovel destination URIs with provided [RabbitMQ node-local] certificate paths.")
+        .long_about("Enables TLS peer verification for all shovel destination URIs by updating their 'verify' parameter and adding [RabbitMQ node-local] certificate and private key file paths.")
+        .arg(
+            Arg::new("node_local_ca_certificate_bundle_path")
+                .long("node-local-ca-certificate-bundle-path")
+                .help("Path to the CA certificate bundle file on the target RabbitMQ node(s)")
+                .required(true)
+                .value_name("PATH")
+        )
+        .arg(
+            Arg::new("node_local_client_certificate_file_path")
+                .long("node-local-client-certificate-file-path")
+                .help("Path to the client certificate file on the target RabbitMQ node(s)")
+                .required(true)
+                .value_name("PATH")
+        )
+        .arg(
+            Arg::new("node_local_client_private_key_file_path")
+                .long("node-local-client-private-key-file-path")
+                .help("Path to the client private key file on the target RabbitMQ node(s)")
+                .required(true)
+                .value_name("PATH")
+        )
+        .after_help(color_print::cformat!(
+            r#"<bold>Doc guides</bold>:
+
  * {}
  * {}
  * {}"#,
@@ -3154,6 +3224,8 @@ pub fn shovel_subcommands(pre_flight_settings: PreFlightSettings) -> [Command; 7
         delete_cmd,
         disable_tls_peer_verification_cmd,
         disable_tls_peer_verification_dest_cmd,
+        enable_tls_peer_verification_source_cmd,
+        enable_tls_peer_verification_dest_cmd,
     ]
     .map(|cmd| cmd.infer_long_args(pre_flight_settings.infer_long_options))
 }

src/commands.rs

@@ -885,6 +885,118 @@ pub fn disable_tls_peer_verification_for_all_destination_uris(
     Ok(())
 }
 
+pub fn enable_tls_peer_verification_for_all_source_uris(
+    client: APIClient,
+    args: &ArgMatches,
+) -> Result<(), CommandRunError> {
+    let ca_cert_path = args
+        .get_one::<String>("node_local_ca_certificate_bundle_path")
+        .ok_or_else(|| CommandRunError::MissingArgumentValue {
+            property: "node_local_ca_certificate_bundle_path".to_string(),
+        })?;
+    let client_cert_path = args
+        .get_one::<String>("node_local_client_certificate_file_path")
+        .ok_or_else(|| CommandRunError::MissingArgumentValue {
+            property: "node_local_client_certificate_file_path".to_string(),
+        })?;
+    let client_key_path = args
+        .get_one::<String>("node_local_client_private_key_file_path")
+        .ok_or_else(|| CommandRunError::MissingArgumentValue {
+            property: "node_local_client_private_key_file_path".to_string(),
+        })?;
+
+    let all_params = client.list_runtime_parameters()?;
+    let shovel_params: Vec<_> = all_params
+        .into_iter()
+        .filter(|p| p.component == "shovel")
+        .collect();
+
+    for param in shovel_params {
+        let owned_params = match OwnedShovelParams::try_from(param.clone()) {
+            Ok(params) => params,
+            Err(_) => continue,
+        };
+
+        let original_source_uri = &owned_params.source_uri;
+        if original_source_uri.is_empty() {
+            continue;
+        }
+
+        let updated_source_uri = enable_tls_peer_verification(
+            original_source_uri,
+            ca_cert_path,
+            client_cert_path,
+            client_key_path,
+        )?;
+
+        if original_source_uri != &updated_source_uri {
+            let mut updated_params = owned_params;
+            updated_params.source_uri = updated_source_uri;
+
+            let param = RuntimeParameterDefinition::from(&updated_params);
+            client.upsert_runtime_parameter(&param)?;
+        }
+    }
+
+    Ok(())
+}
+
+pub fn enable_tls_peer_verification_for_all_destination_uris(
+    client: APIClient,
+    args: &ArgMatches,
+) -> Result<(), CommandRunError> {
+    let ca_cert_path = args
+        .get_one::<String>("node_local_ca_certificate_bundle_path")
+        .ok_or_else(|| CommandRunError::MissingArgumentValue {
+            property: "node_local_ca_certificate_bundle_path".to_string(),
+        })?;
+    let client_cert_path = args
+        .get_one::<String>("node_local_client_certificate_file_path")
+        .ok_or_else(|| CommandRunError::MissingArgumentValue {
+            property: "node_local_client_certificate_file_path".to_string(),
+        })?;
+    let client_key_path = args
+        .get_one::<String>("node_local_client_private_key_file_path")
+        .ok_or_else(|| CommandRunError::MissingArgumentValue {
+            property: "node_local_client_private_key_file_path".to_string(),
+        })?;
+
+    let all_params = client.list_runtime_parameters()?;
+    let shovel_params: Vec<_> = all_params
+        .into_iter()
+        .filter(|p| p.component == "shovel")
+        .collect();
+
+    for param in shovel_params {
+        let owned_params = match OwnedShovelParams::try_from(param.clone()) {
+            Ok(params) => params,
+            Err(_) => continue,
+        };
+
+        let original_destination_uri = &owned_params.destination_uri;
+        if original_destination_uri.is_empty() {
+            continue;
+        }
+
+        let updated_destination_uri = enable_tls_peer_verification(
+            original_destination_uri,
+            ca_cert_path,
+            client_cert_path,
+            client_key_path,
+        )?;
+
+        if original_destination_uri != &updated_destination_uri {
+            let mut updated_params = owned_params;
+            updated_params.destination_uri = updated_destination_uri;
+
+            let param = RuntimeParameterDefinition::from(&updated_params);
+            client.upsert_runtime_parameter(&param)?;
+        }
+    }
+
+    Ok(())
+}
+
 //
 // Feature flags
 //

src/main.rs

@@ -1086,6 +1086,20 @@ fn dispatch_common_subcommand(
             let result = commands::disable_tls_peer_verification_for_all_destination_uris(client);
             res_handler.no_output_on_success(result);
         }
+        ("shovels", "enable_tls_peer_verification_for_all_source_uris") => {
+            let result = commands::enable_tls_peer_verification_for_all_source_uris(
+                client,
+                second_level_args,
+            );
+            res_handler.no_output_on_success(result);
+        }
+        ("shovels", "enable_tls_peer_verification_for_all_destination_uris") => {
+            let result = commands::enable_tls_peer_verification_for_all_destination_uris(
+                client,
+                second_level_args,
+            );
+            res_handler.no_output_on_success(result);
+        }
         ("streams", "declare") => {
             let result = commands::declare_stream(client, &vhost, second_level_args);
             res_handler.no_output_on_success(result);

tests/bindings_tests.rs

@@ -202,7 +202,7 @@ fn test_bindings_list() -> Result<(), Box<dyn std::error::Error>> {
 
 #[test]
 fn test_bindings_delete_idempotently() -> Result<(), Box<dyn std::error::Error>> {
-    let vh = "bindings.delete.idempotently.1";
+    let vh = "rabbitmqadmin.bindings.test1";
     let source_ex = "test_source_exchange";
     let dest_queue = "test_dest_queue";
     let routing_key = "test.routing.key";

tests/combined_integration_tests.rs

@@ -20,7 +20,7 @@ use test_helpers::{run_fails, run_succeeds};
 
 #[test]
 fn combined_integration_test1() -> Result<(), Box<dyn std::error::Error>> {
-    let vh = "combined_integration_test1";
+    let vh = "rabbitmqadmin.combined_integration.test1";
     let config_path = path::absolute("./tests/fixtures/config_files/config_file1.conf")
         .expect("failed to compute an absolute version for a ./test/fixtures path");
 
@@ -45,7 +45,7 @@ fn combined_integration_test1() -> Result<(), Box<dyn std::error::Error>> {
 
 #[test]
 fn combined_integration_test2() -> Result<(), Box<dyn std::error::Error>> {
-    let vh = "combined_integration_test2";
+    let vh = "rabbitmqadmin.combined_integration.test2";
 
     // Uses a node alias that does not exist in the file
     let config_path = path::absolute("tests/fixtures/config_files/config_file1.conf")
@@ -77,7 +77,7 @@ fn combined_integration_test2() -> Result<(), Box<dyn std::error::Error>> {
 
 #[test]
 fn combined_integration_test3() -> Result<(), Box<dyn std::error::Error>> {
-    let vh = "combined_integration_test3";
+    let vh = "rabbitmqadmin.combined_integration.test3";
 
     // Uses a node alias that does not exist in the file
     let config_path = path::absolute("tests/fixtures/config_files/non_exis7ent_c0nfig_f1le.conf")
@@ -99,7 +99,7 @@ fn combined_integration_test3() -> Result<(), Box<dyn std::error::Error>> {
 fn combined_integration_test4() -> Result<(), Box<dyn std::error::Error>> {
     // This test uses administrative credentials to create a new user
     // and set up a topology using those new credentials
-    let vh = "combined_integration_test4";
+    let vh = "rabbitmqadmin.combined_integration.test4";
     let new_user = "user_from_combined_integration_test4";
     let new_pass = "p4$$w0rd_from_combined_integration_test4";
     let x = "fanout_combined_integration_test4";

tests/definitions_export_tests.rs

@@ -27,7 +27,7 @@ fn test_export_cluster_wide_definitions() -> Result<(), Box<dyn std::error::Erro
 
 #[test]
 fn test_export_vhost_definitions() -> Result<(), Box<dyn std::error::Error>> {
-    let vh = "test_export_vhost_definitions.1";
+    let vh = "rabbitmqadmin.definitions_export.test1";
     delete_vhost(vh).expect("failed to delete a virtual host");
     run_succeeds(["declare", "vhost", "--name", vh]);
 
@@ -49,7 +49,7 @@ fn test_export_vhost_definitions() -> Result<(), Box<dyn std::error::Error>> {
 #[test]
 fn test_export_cluster_wide_definitions_with_transformations_case1()
 -> Result<(), Box<dyn std::error::Error>> {
-    let vh = "test_export_cluster_definitions.transformations.1";
+    let vh = "rabbitmqadmin.definitions_export.test2";
     delete_vhost(vh).expect("failed to delete a virtual host");
     run_succeeds(["declare", "vhost", "--name", vh]);
 
@@ -98,7 +98,7 @@ fn test_export_cluster_wide_definitions_with_transformations_case1()
 #[test]
 fn test_export_vhost_definitions_with_transformations_case1()
 -> Result<(), Box<dyn std::error::Error>> {
-    let vh = "test_export_vhost_definitions.transformations.1";
+    let vh = "rabbitmqadmin.definitions_export.test3";
     delete_vhost(vh).expect("failed to delete a virtual host");
     run_succeeds(["declare", "vhost", "--name", vh]);
 

tests/definitions_import_tests.rs

@@ -35,7 +35,7 @@ fn test_import_cluster_definitions() -> Result<(), Box<dyn std::error::Error>> {
 
 #[test]
 fn test_import_vhost_definitions() -> Result<(), Box<dyn std::error::Error>> {
-    let vh = "test_import_vhost_definitions.1";
+    let vh = "rabbitmqadmin.definitions_import.test1";
 
     delete_vhost(vh).expect("failed to delete a virtual host");
     run_succeeds(["declare", "vhost", "--name", vh]);