Async

Cargo.toml

@@ -14,7 +14,7 @@ acme-lib = { git = 'https://github.com/DBCDK/acme-lib', branch = 'dbc-fork' }
 regex = "1"
 lazy_static = "1"
 walkdir = "2"
-trust-dns-resolver = "0"
+trust-dns-resolver = { version = "0", features = ["tokio-runtime"] }
 env_logger = "0"
 prometheus_exporter_base = { version = "=1.4.0", features = ["hyper_server"] }
 tokio = { version = "1", features = [ "full" ] }

nixos/vault-test.nix

@@ -215,6 +215,7 @@ nixos-lib.runTest (
               dnsutils
               dig
             ];
+            environment.RUST_BACKTRACE = "full";
             wantedBy = [ "multi-user.target" ];
             wants = [ "vault-provision.service" ];
             after = [ "vault-provision.service" ];

src/common.rs

@@ -156,7 +156,7 @@ impl CertSpec {
 }
 
 pub trait Persistable {
-    fn persist(&self, cert: Certificate) -> Result<(), PersistError>;
+    async fn persist(&self, cert: Certificate) -> Result<(), PersistError>;
 }
 
 #[derive(Debug, Clone, Serialize)]
@@ -183,11 +183,11 @@ pub enum PersistSpec {
 }
 
 impl Persistable for CertSpec {
-    fn persist(&self, cert: Certificate) -> Result<(), PersistError> {
+    async fn persist(&self, cert: Certificate) -> Result<(), PersistError> {
         match &self.persist_spec {
             PersistSpec::KUBERNETES(spec) => Ok(kube::persist(&spec, &cert)?),
             PersistSpec::FILE(spec) => Ok(file::persist(&spec, &cert)?),
-            PersistSpec::VAULT(spec) => Ok(vault::persist(&spec, cert)?),
+            PersistSpec::VAULT(spec) => Ok(vault::persist(&spec, cert).await?),
             //PersistSpec::FILE(_spec) => { unimplemented!() },
             PersistSpec::DONTPERSIST => { Ok(()) }
         }
@@ -325,8 +325,8 @@ pub trait ValidityVerifier {
 
 pub trait CertSpecable: IssueSource {
     fn to_cert_spec(&self, config: &ConfigContainer) -> Result<CertSpec, SpecError>;
-    fn touch(&self, config: &ConfigContainer) -> Result<(), TouchError>;
-    fn should_retry(&self, config: &ConfigContainer) -> bool;
+    async fn touch(&self, config: &ConfigContainer) -> Result<(), TouchError>;
+    async fn should_retry(&self, config: &ConfigContainer) -> bool;
 }
 
 pub trait IssueSource {

src/dns/mod.rs

@@ -8,7 +8,7 @@ use std::convert::From;
 use crate::log;
 use crate::common::{CertSpec, DNSName, SpecError};
 use crate::config::Zone;
-use self::trust_dns_resolver::Resolver;
+use self::trust_dns_resolver::TokioAsyncResolver;
 use self::trust_dns_resolver::error::{ResolveError,ResolveErrorKind};
 use std::string::String;
 
@@ -77,9 +77,9 @@ pub fn delete(config: &FaytheConfig, spec: &CertSpec) -> Result<(), DNSError> {
     Ok(())
 }
 
-pub fn query(resolver: &Resolver, host: &DNSName, proof: &String) -> Result<(), DNSError> {
+pub async fn query(resolver: &TokioAsyncResolver, host: &DNSName, proof: &String) -> Result<(), DNSError> {
     let challenge_host = challenge_host(host, None);
-    match resolver.txt_lookup(&challenge_host) {
+    match resolver.txt_lookup(&challenge_host).await {
         Ok(res) => {
             let trim_chars: &[_] = &['"', '\n'];
             res.iter().find(|record_set|

src/file.rs

@@ -151,7 +151,7 @@ impl CertSpecable for FileSpec {
         })
     }
 
-    fn touch(&self, config: &ConfigContainer) -> Result<(), TouchError> {
+    async fn touch(&self, config: &ConfigContainer) -> Result<(), TouchError> {
         let monitor_config = config.get_file_monitor_config()?;
         let names = default_file_names(&self);
         let sub_dir = absolute_dir_path(&monitor_config, names.sub_directory.as_ref());
@@ -164,7 +164,7 @@ impl CertSpecable for FileSpec {
         Ok(())
     }
 
-    fn should_retry(&self, config: &ConfigContainer) -> bool {
+    async fn should_retry(&self, config: &ConfigContainer) -> bool {
         use std::time::Duration;
 
         match || -> Result<(), TouchError> {
@@ -212,6 +212,7 @@ impl FileNames {
     }
 }
 
+#[derive(Debug)]
 pub enum FileError {
     IO
 }

src/issuer.rs

@@ -1,11 +1,11 @@
 
-use std::thread;
 use std::time::Duration;
 
 use crate::{dns, FaytheConfig, common};
 use crate::log;
 
-use std::sync::mpsc::{Receiver,TryRecvError};
+use tokio::sync::mpsc::Receiver;
+use tokio::sync::mpsc::error::TryRecvError;
 use std::collections::{VecDeque, HashSet, HashMap};
 
 use acme_lib::{ClientConfig, Directory, DirectoryUrl, create_rsa_key};
@@ -17,8 +17,11 @@ use std::prelude::v1::Vec;
 
 use serde_json::json;
 use std::convert::TryFrom;
-use trust_dns_resolver::Resolver;
-use std::sync::RwLock;
+use trust_dns_resolver::{AsyncResolver, TokioAsyncResolver};
+use trust_dns_resolver::config::{ResolverConfig, NameServerConfigGroup, ResolverOpts};
+use trust_dns_resolver::error::ResolveErrorKind;
+use std::net::IpAddr;
+use tokio::sync::RwLock;
 use std::fmt::Debug;
 use crate::dns::DNSError;
 
@@ -27,10 +30,11 @@ use crate::metrics::MetricsType;
 
 use chrono::Utc;
 
-pub fn process(faythe_config: FaytheConfig, rx: Receiver<CertSpec>) {
+pub async fn process(faythe_config: FaytheConfig, mut rx: Receiver<CertSpec>) {
 
     let mut queue: VecDeque<IssueOrder> = VecDeque::new();
-    RESOLVERS.with(|r| r.write().unwrap().inner = init_resolvers(&faythe_config));
+    let resolvers = init_resolvers(&faythe_config).await.unwrap();
+    RESOLVERS.write().await.inner = resolvers;
 
     log::info("processing-started");
     loop {
@@ -53,23 +57,23 @@ pub fn process(faythe_config: FaytheConfig, rx: Receiver<CertSpec>) {
             Err(_) => {}
         }
 
-        let queue_check = check_queue(&mut queue);
+        let queue_check = check_queue(&mut queue).await;
         if queue_check.is_err() {
             log::info("check queue err");
             log::info(&format!("{:?}", queue_check));
         }
-        thread::sleep(Duration::from_millis(5000));
+        tokio::time::sleep(Duration::from_secs(5)).await;
     }
 }
 
-fn check_queue(queue: &mut VecDeque<IssueOrder>) -> Result<(), IssuerError> {
+async fn check_queue(queue: &mut VecDeque<IssueOrder>) -> Result<(), IssuerError> {
     match queue.pop_front() {
         Some(mut order) => {
-            match validate_challenge(&order) {
+            match validate_challenge(&order).await {
                 Ok(_) => {
                     order.inner.refresh()?;
                     if order.inner.is_validated() {
-                        let result = order.issue();
+                        let result = order.issue().await;
                         match result {
                             Ok(_) => metrics::new_event(&order.spec.name, MetricsType::Success),
                             Err(_) => metrics::new_event(&order.spec.name, MetricsType::Failure),
@@ -103,28 +107,28 @@ fn check_queue(queue: &mut VecDeque<IssueOrder>) -> Result<(), IssuerError> {
     }
 }
 
-fn validate_challenge(order: &IssueOrder) -> Result<(), IssuerError> {
-
+async fn validate_challenge(order: &IssueOrder) -> Result<(), IssuerError> {
     for a in &order.authorizations {
         let domain = DNSName::try_from(&String::from(a.domain_name()))?;
         let challenge = a.dns_challenge();
         let proof = challenge.dns_proof();
         let log_data = json!({ "domain": &domain, "proof": &proof });
 
-        RESOLVERS.with(|r| -> Result<(), DNSError> {
+        {
+            let resolvers = RESOLVERS.read().await;
+
             // TODO: Proper retry logic
             log::info("Validating internally after 20s");
 
             log::data("Validating auth_dns_servers internally", &log_data);
             for d in &order.auth_dns_servers {
-                dns::query(r.read().unwrap().get(&d).unwrap(), &domain, &proof)?;
+                dns::query(resolvers.get(&d).unwrap(), &domain, &proof).await?;
             }
             log::data("Validating val_dns_servers internally", &log_data);
             for d in &order.val_dns_servers {
-                dns::query(r.read().unwrap().get(&d).unwrap(), &domain, &proof)?;
+                dns::query(resolvers.get(&d).unwrap(), &domain, &proof).await?;
             }
-            Ok(())
-        })?;
+        }
         log::data("Asking LE to validate", &log_data);
         challenge.validate(5000)?;
     }
@@ -187,7 +191,7 @@ struct IssueOrder {
 }
 
 impl IssueOrder {
-    fn issue(&self) -> Result<(), IssuerError> {
+    async fn issue(&self) -> Result<(), IssuerError> {
         log::data("Issuing", &self.spec);
 
         let pkey_pri = create_rsa_key(2048);
@@ -200,7 +204,7 @@ impl IssueOrder {
             ord_csr.finalize_pkey(pkey_pri, 5000)?;
         let cert = ord_cert.download_and_save_cert()?;
 
-        Ok(self.spec.persist(cert)?)
+        Ok(self.spec.persist(cert).await?)
     }
 }
 
@@ -250,66 +254,61 @@ impl std::convert::From<std::io::Error> for ResolverError<'_> {
     }
 }
 
-thread_local! {
-    static RESOLVERS: RwLock<Resolvers> = RwLock::new(Resolvers{
+lazy_static! {
+    static ref RESOLVERS: RwLock<Resolvers> = RwLock::new(Resolvers{
         inner: HashMap::with_capacity(0)
     });
 }
 
 struct Resolvers {
-    inner: HashMap<String, Resolver>
+    inner: HashMap<String, TokioAsyncResolver>
 }
 
 impl Resolvers {
-    fn get(&self, server: &String) -> Option<&Resolver> {
+    fn get(&self, server: &String) -> Option<&TokioAsyncResolver> {
         self.inner.get(server)
     }
 }
 
-fn init_resolvers<'l>(config: &FaytheConfig) -> HashMap<String, Resolver> {
-    use trust_dns_resolver::config::{ResolverConfig, NameServerConfigGroup, ResolverOpts};
-    use trust_dns_resolver::error::ResolveErrorKind;
-    use std::net::IpAddr;
-
+async fn init_resolvers<'l>(config: &FaytheConfig) -> Result<HashMap<String, TokioAsyncResolver>, ResolverError> {
     let mut resolvers = HashMap::new();
 
-    let create_resolvers = |server: &String, resolvers: &mut HashMap<String, Resolver>| {
-
-        match || -> Result<(), ResolverError> {
-            let system_resolver = Resolver::from_system_conf().or(Err(ResolverError::SystemResolveConf))?;
-
-            //try-parse what's in the config file as an ip-address, if that fails, assume it's a hostname that can be looked up
-            let ip: IpAddr = match server.parse() {
-                Ok(ip) => Ok(ip),
-                Err(_) => {
-                    match system_resolver.lookup_ip(server) {
-                        Ok(res) => res.iter().next().ok_or(ResolverError::NoIpsForResolversFound(server)), // grabbing the first A record only for now
-                        Err(err) => {
-                            Err(match err.kind() {
-                                ResolveErrorKind::NoRecordsFound { .. } => ResolverError::NoIpsForResolversFound(server),
-                                _ => ResolverError::Other
-                            })
-                        }
-                    }
-                }
-            }?;
+    for z in &config.zones {
+        let server = &z.1.auth_dns_server;
+        resolvers.insert(server.clone(), create_resolvers(&server).await?);
+    }
+    for s in &config.val_dns_servers {
+        resolvers.insert(s.to_string(), create_resolvers(&s).await?);
+    }
+    Ok(resolvers)
+}
 
-            let mut conf = ResolverConfig::new();
-            for c in &*NameServerConfigGroup::from_ips_clear(&[ip.to_owned()], 53, true) {
-                conf.add_name_server(c.to_owned());
+async fn create_resolvers<'a>(server: &'a String) -> Result<TokioAsyncResolver, ResolverError<'a>> {
+
+    let system_resolver = AsyncResolver::tokio_from_system_conf().or(Err(ResolverError::SystemResolveConf))?;
+
+    //try-parse what's in the config file as an ip-address, if that fails, assume it's a hostname that can be looked up
+    let ip: IpAddr = match server.parse() {
+        Ok(ip) => Ok(ip),
+        Err(_) => {
+            match system_resolver.lookup_ip(server).await {
+                Ok(res) => res.iter().next().ok_or(ResolverError::NoIpsForResolversFound(server)), // grabbing the first A record only for now
+                Err(err) => {
+                    Err(match err.kind() {
+                        ResolveErrorKind::NoRecordsFound { .. } => ResolverError::NoIpsForResolversFound(server),
+                        _ => ResolverError::Other
+                    })
+                }
             }
-            let mut opts = ResolverOpts::default();
-            // Never believe NXDOMAIN for more than 1 minute
-            opts.negative_max_ttl = Some(Duration::new(60,0));
-            resolvers.insert(server.to_owned(), Resolver::new(conf, opts).unwrap());
-            Ok(())
-        }() {
-            Err(e) => { log::error(format!("failed to init resolver for server: {}", &server).as_str(), &e); }
-            _ => {}
-        };
-    };
+        }
+    }?;
 
-    config.zones.iter().for_each(|(_, z)| create_resolvers(&z.auth_dns_server, &mut resolvers));
-    config.val_dns_servers.iter().for_each(|s| create_resolvers(&s, &mut resolvers));
-    resolvers
+    let mut conf = ResolverConfig::new();
+    for c in &*NameServerConfigGroup::from_ips_clear(&[ip.to_owned()], 53, true) {
+        conf.add_name_server(c.to_owned());
+    }
+    let mut opts = ResolverOpts::default();
+    // Never believe NXDOMAIN for more than 1 minute
+    opts.negative_max_ttl = Some(Duration::new(60,0));
+    Ok(AsyncResolver::tokio(conf, opts))
 }

src/kube.rs

@@ -41,7 +41,7 @@ pub struct Secret {
 
 const TIME_FORMAT: &str = "%Y-%m-%dT%H:%M:%S%z"; // 2019-10-09T11:50:22+0200
 
-pub fn get_secrets(config: &KubeMonitorConfig) -> Result<HashMap<CertName, Secret>, KubeError> {
+pub async fn get_secrets(config: &KubeMonitorConfig) -> Result<HashMap<CertName, Secret>, KubeError> {
 
     let v = kubectl(&["get", "secrets",
         "-l", config.secret_hostlabel.as_str(),
@@ -70,7 +70,7 @@ pub fn get_secrets(config: &KubeMonitorConfig) -> Result<HashMap<CertName, Secre
     Ok(secrets)
 }
 
-pub fn get_ingresses(config: &KubeMonitorConfig) -> Result<Vec<Ingress>, KubeError> {
+pub async fn get_ingresses(config: &KubeMonitorConfig) -> Result<Vec<Ingress>, KubeError> {
     let v = kubectl(&["get", "ingresses", "--all-namespaces"])?;
 
     let mut ingresses :Vec<Ingress> = Vec::new();
@@ -259,15 +259,15 @@ impl CertSpecable for Ingress {
         })
     }
 
-    fn touch(&self, config: &ConfigContainer) -> Result<(), TouchError> {
+    async fn touch(&self, config: &ConfigContainer) -> Result<(), TouchError> {
         let monitor_config = config.get_kube_monitor_config()?;
         match &monitor_config.touch_annotation {
             Some(a) => Ok(self.annotate(&a, &Utc::now().format(TIME_FORMAT).to_string())?),
             None => Ok(())
         }
     }
 
-    fn should_retry(&self, config: &ConfigContainer) -> bool {
+    async fn should_retry(&self, config: &ConfigContainer) -> bool {
         Utc::now() > self.touched + chrono::Duration::milliseconds(config.faythe_config.issue_grace as i64)
     }
 }