Introduce 'federation enable_tls_peer_verification_for_all_upstreams'

Author: michaelklishin

CHANGELOG.md

@@ -4,6 +4,26 @@
 
 ### Enhancements
 
+* `federation enable_tls_peer_verification_for_all_upstreams` is a new command that enables TLS peer verification
+  for all federation upstreams:
+
+  ```shell
+  # Note that the certificate and private key paths below refer
+  # to the files deployed to the target RabbitMQ node(s), not to the
+  # local files.
+  #
+  # As such, these arguments are command-specific and should not be confused
+  # with the global `--tls-ca-cert-file`, `--tls-cert-file`, and `--tls-key-file`
+  # arguments that are used by `rabbitmqadmin` itself to connect to the target node
+  # over the HTTP API.
+  rabbitmqadmin federation enable_tls_peer_verification_for_all_upstreams \
+      --node-local-ca-certificate-bundle-path /path/to/node/local/ca_bundle.pem \
+      --node-local-client-certificate-file-path /path/to/node/local/client_certificate.pem \
+      --node-local-client-private-key-file-path /path/to/node/local/client_private_key.pem
+  ```
+
+   See [TLS guide](https://www.rabbitmq.com/docs/ssl#peer-verification) and [Federation guide](https://www.rabbitmq.com/docs/federation#tls-connections) to learn more.
+
  * `shovel disable_tls_peer_verification_for_all_source_uris` is a new command that disables TLS peer verification
    for all shovel source URIs.
 

src/cli.rs

@@ -3083,7 +3083,7 @@ pub fn shovel_subcommands(pre_flight_settings: PreFlightSettings) -> [Command; 7
     .map(|cmd| cmd.infer_long_args(pre_flight_settings.infer_long_options))
 }
 
-fn federation_subcommands(pre_flight_settings: PreFlightSettings) -> [Command; 7] {
+fn federation_subcommands(pre_flight_settings: PreFlightSettings) -> [Command; 8] {
     let list_all_upstreams = Command::new("list_all_upstreams")
         .long_about("Lists federation upstreams in all virtual hosts")
         .after_help(color_print::cformat!(
@@ -3426,6 +3426,41 @@ fn federation_subcommands(pre_flight_settings: PreFlightSettings) -> [Command; 7
         .after_help(color_print::cformat!(
             r#"<bold>Doc guides</bold>:
 
+ * {}
+ * {}
+ * {}"#,
+            FEDERATION_GUIDE_URL,
+            TLS_GUIDE_URL,
+            "https://www.rabbitmq.com/docs/federation#tls-connections"
+        ));
+
+    let enable_tls_peer_verification_cmd = Command::new("enable_tls_peer_verification_for_all_upstreams")
+        .about("Enables TLS peer verification for all federation upstreams with provided [RabbitMQ node-local] certificate paths.")
+        .long_about("Enables TLS peer verification for all federation upstreams 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>:
+
  * {}
  * {}
  * {}"#,
@@ -3442,6 +3477,7 @@ fn federation_subcommands(pre_flight_settings: PreFlightSettings) -> [Command; 7
         delete_upstream,
         list_all_links,
         disable_tls_peer_verification_cmd,
+        enable_tls_peer_verification_cmd,
     ]
     .map(|cmd| cmd.infer_long_args(pre_flight_settings.infer_long_options))
 }

src/commands.rs

@@ -733,6 +733,80 @@ pub fn disable_tls_peer_verification_for_all_federation_upstreams(
     Ok(())
 }
 
+pub fn enable_tls_peer_verification_for_all_federation_upstreams(
+    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 upstreams = client.list_federation_upstreams()?;
+
+    for upstream in upstreams {
+        let original_uri = &upstream.uri;
+        let updated_uri = enable_tls_peer_verification(
+            original_uri,
+            ca_cert_path,
+            client_cert_path,
+            client_key_path,
+        )?;
+
+        if original_uri != &updated_uri {
+            let upstream_params = FederationUpstreamParams {
+                name: &upstream.name,
+                vhost: &upstream.vhost,
+                uri: &updated_uri,
+                prefetch_count: upstream
+                    .prefetch_count
+                    .unwrap_or(DEFAULT_FEDERATION_PREFETCH),
+                reconnect_delay: upstream.reconnect_delay.unwrap_or(5),
+                ack_mode: upstream.ack_mode,
+                trust_user_id: upstream.trust_user_id.unwrap_or_default(),
+                bind_using_nowait: upstream.bind_using_nowait,
+                channel_use_mode: upstream.channel_use_mode,
+                queue_federation: if upstream.queue.is_some() {
+                    Some(QueueFederationParams {
+                        queue: upstream.queue.as_deref(),
+                        consumer_tag: upstream.consumer_tag.as_deref(),
+                    })
+                } else {
+                    None
+                },
+                exchange_federation: if upstream.exchange.is_some() {
+                    Some(ExchangeFederationParams {
+                        exchange: upstream.exchange.as_deref(),
+                        max_hops: upstream.max_hops,
+                        queue_type: upstream.queue_type.unwrap_or(QueueType::Classic),
+                        ttl: upstream.expires,
+                        message_ttl: upstream.message_ttl,
+                        resource_cleanup_mode: upstream.resource_cleanup_mode,
+                    })
+                } else {
+                    None
+                },
+            };
+
+            let param = RuntimeParameterDefinition::from(upstream_params);
+            client.upsert_runtime_parameter(&param)?;
+        }
+    }
+
+    Ok(())
+}
+
 pub fn disable_tls_peer_verification_for_all_source_uris(
     client: APIClient,
 ) -> Result<(), CommandRunError> {
@@ -1986,3 +2060,26 @@ fn disable_tls_peer_verification(uri: &str) -> Result<String, CommandRunError> {
             message: format!("Failed to reconstruct (modify) a URI: {}", e),
         })
 }
+
+fn enable_tls_peer_verification(
+    uri: &str,
+    ca_cert_path: &str,
+    client_cert_path: &str,
+    client_key_path: &str,
+) -> Result<String, CommandRunError> {
+    use rabbitmq_http_client::uris::UriBuilder;
+
+    let ub = UriBuilder::new(uri)
+        .map_err(|e| CommandRunError::FailureDuringExecution {
+            message: format!("Could not parse a value as a URI '{}': {}", uri, e),
+        })?
+        .with_tls_peer_verification(TlsPeerVerificationMode::Enabled)
+        .with_ca_cert_file(ca_cert_path)
+        .with_client_cert_file(client_cert_path)
+        .with_client_key_file(client_key_path);
+
+    ub.build()
+        .map_err(|e| CommandRunError::FailureDuringExecution {
+            message: format!("Failed to reconstruct (modify) a URI: {}", e),
+        })
+}

src/main.rs

@@ -673,6 +673,13 @@ fn dispatch_common_subcommand(
                 commands::disable_tls_peer_verification_for_all_federation_upstreams(client);
             res_handler.no_output_on_success(result);
         }
+        ("federation", "enable_tls_peer_verification_for_all_upstreams") => {
+            let result = commands::enable_tls_peer_verification_for_all_federation_upstreams(
+                client,
+                second_level_args,
+            );
+            res_handler.no_output_on_success(result);
+        }
         ("get", "messages") => {
             let result = commands::get_messages(client, &vhost, second_level_args);
             res_handler.tabular_result(result)