From 002b9fc50c5b69e0e10c84e4db5ecea1b3941fad Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Tue, 18 Oct 2022 18:38:20 +0200 Subject: Add TLS support for Consul discovery + refactoring --- src/rpc/consul.rs | 129 ++++++++++++++++++++++++++++++++++-------------------- 1 file changed, 81 insertions(+), 48 deletions(-) (limited to 'src/rpc/consul.rs') diff --git a/src/rpc/consul.rs b/src/rpc/consul.rs index 15acbcef..05ed278a 100644 --- a/src/rpc/consul.rs +++ b/src/rpc/consul.rs @@ -1,14 +1,66 @@ use std::collections::HashMap; use std::net::{IpAddr, SocketAddr}; -use hyper::client::Client; -use hyper::StatusCode; -use hyper::{Body, Method, Request}; +use tokio::fs::File; +use tokio::io::AsyncReadExt; + +use err_derive::Error; use serde::{Deserialize, Serialize}; use netapp::NodeID; -use garage_util::error::Error; +use garage_util::config::ConsulDiscoveryConfig; + +async fn make_consul_client( + config: &ConsulDiscoveryConfig, +) -> Result { + match (&config.client_cert, &config.client_key) { + (Some(client_cert), Some(client_key)) => { + let mut client_cert_buf = vec![]; + File::open(client_cert) + .await? + .read_to_end(&mut client_cert_buf) + .await?; + + let mut client_key_buf = vec![]; + File::open(client_key) + .await? + .read_to_end(&mut client_key_buf) + .await?; + + let identity = reqwest::Identity::from_pem( + &[&client_cert_buf[..], &client_key_buf[..]].concat()[..], + )?; + + if config.tls_skip_verify { + Ok(reqwest::Client::builder() + .use_rustls_tls() + .danger_accept_invalid_certs(true) + .identity(identity) + .build()?) + } else if let Some(ca_cert) = &config.ca_cert { + let mut ca_cert_buf = vec![]; + File::open(ca_cert) + .await? + .read_to_end(&mut ca_cert_buf) + .await?; + + Ok(reqwest::Client::builder() + .use_rustls_tls() + .add_root_certificate(reqwest::Certificate::from_pem(&ca_cert_buf[..])?) + .identity(identity) + .build()?) + } else { + Ok(reqwest::Client::builder() + .use_rustls_tls() + .identity(identity) + .build()?) + } + } + (None, None) => Ok(reqwest::Client::new()), + _ => Err(ConsulError::InvalidTLSConfig), + } +} // ---- READING FROM CONSUL CATALOG ---- @@ -23,27 +75,16 @@ struct ConsulQueryEntry { } pub async fn get_consul_nodes( - consul_host: &str, - consul_service_name: &str, -) -> Result, Error> { + consul_config: &ConsulDiscoveryConfig, +) -> Result, ConsulError> { let url = format!( "http://{}/v1/catalog/service/{}", - consul_host, consul_service_name + consul_config.consul_host, consul_config.service_name ); - let req = Request::builder() - .uri(url) - .method(Method::GET) - .body(Body::default())?; - - let client = Client::new(); - - let resp = client.request(req).await?; - if resp.status() != StatusCode::OK { - return Err(Error::Message(format!("HTTP error {}", resp.status()))); - } - let body = hyper::body::to_bytes(resp.into_body()).await?; - let entries = serde_json::from_slice::>(body.as_ref())?; + let client = make_consul_client(consul_config).await?; + let http = client.get(&url).send().await?; + let entries: Vec = http.json().await?; let mut ret = vec![]; for ent in entries { @@ -96,15 +137,14 @@ struct ConsulPublishService { } pub async fn publish_consul_service( - consul_host: &str, - consul_service_name: &str, + consul_config: &ConsulDiscoveryConfig, node_id: NodeID, hostname: &str, rpc_public_addr: SocketAddr, -) -> Result<(), Error> { +) -> Result<(), ConsulError> { let node = format!("garage:{}", hex::encode(&node_id[..8])); - let advertisment = ConsulPublishEntry { + let advertisement = ConsulPublishEntry { node: node.clone(), address: rpc_public_addr.ip(), node_meta: [ @@ -116,36 +156,29 @@ pub async fn publish_consul_service( .collect(), service: ConsulPublishService { service_id: node.clone(), - service_name: consul_service_name.to_string(), + service_name: consul_config.service_name.clone(), tags: vec!["advertised-by-garage".into(), hostname.into()], address: rpc_public_addr.ip(), port: rpc_public_addr.port(), }, }; - let url = format!("http://{}/v1/catalog/register", consul_host); - let req_body = serde_json::to_string(&advertisment)?; - debug!("Request body for consul adv: {}", req_body); + let url = format!("http://{}/v1/catalog/register", consul_config.consul_host); - let req = Request::builder() - .uri(url) - .method(Method::PUT) - .body(Body::from(req_body))?; - - let client = Client::new(); - - let resp = client.request(req).await?; - debug!("Response of advertising to Consul: {:?}", resp); - let resp_code = resp.status(); - let resp_bytes = &hyper::body::to_bytes(resp.into_body()).await?; - debug!( - "{}", - std::str::from_utf8(resp_bytes).unwrap_or("") - ); - - if resp_code != StatusCode::OK { - return Err(Error::Message(format!("HTTP error {}", resp_code))); - } + let client = make_consul_client(consul_config).await?; + let http = client.put(&url).json(&advertisement).send().await?; + http.error_for_status()?; Ok(()) } + +/// Regroup all Garage errors +#[derive(Debug, Error)] +pub enum ConsulError { + #[error(display = "IO error: {}", _0)] + Io(#[error(source)] std::io::Error), + #[error(display = "HTTP error: {}", _0)] + Reqwest(#[error(source)] reqwest::Error), + #[error(display = "Invalid Consul TLS configuration")] + InvalidTLSConfig, +} -- cgit v1.2.3