529 internode auth in swarm fix (#622)

* dnsrr in compose file

* base resolving

* configurable resolving

* update changelog

* better node request checker

* async sleep & no resolved|unresolved distinction

* fixes

* changes

* changes

* changes

* changes

* changes

* set inProgress

Author: Boneyan

CHANGELOG.md

@@ -7,7 +7,7 @@ Bob versions changelog
 
 
 #### Changed
-
+- Hostname resolving in background tasks (#529)
 
 #### Fixed
 

bob-access/Cargo.toml

@@ -20,3 +20,8 @@ serde_yaml = "0.8"
 serde_json = "1.0"
 futures = "0.3"
 unicase = "2.6"
+
+[dependencies.tokio]
+version = "1.21"
+default-features = false
+features = []

bob-access/src/authenticator/basic.rs

@@ -1,65 +1,135 @@
-use std::{collections::HashMap, net::IpAddr};
+use std::{
+    collections::HashMap,
+    net::SocketAddr,
+    sync::{RwLock, Arc},
+    time::Duration,
+};
 
-use crate::{credentials::{Credentials, CredentialsKind}, AuthenticationType, error::Error, permissions::Permissions};
+use crate::{credentials::{DeclaredCredentials, DCredentialsResolveGuard, RequestCredentials, CredentialsKind}, AuthenticationType, error::Error, permissions::Permissions};
 
 use super::{users_storage::UsersStorage, Authenticator};
 
 use sha2::{Digest, Sha512};
 
+use tokio::net::lookup_host;
+
+#[inline]
+async fn lookup(hostname: &str) -> Option<Vec<SocketAddr>> {
+    lookup_host(hostname).await
+    .ok()
+    .map(|addr| addr.collect())
+}
+
+type NodesCredentials = HashMap<String, DCredentialsResolveGuard>;
+
 #[derive(Debug, Default, Clone)]
 pub struct Basic<Storage: UsersStorage> {
     users_storage: Storage,
-    nodes: HashMap<IpAddr, Vec<Credentials>>,
+    nodes: Arc<RwLock<NodesCredentials>>,
+    resolve_sleep_period_ms: u64,
 }
 
 impl<Storage: UsersStorage> Basic<Storage> {
-    pub fn new(users_storage: Storage) -> Self {
+    pub fn new(users_storage: Storage, resolve_sleep_period_ms: u64) -> Self {
         Self {
             users_storage,
-            nodes: HashMap::new(),
+            nodes: Arc::new(RwLock::new(HashMap::new())),
+            resolve_sleep_period_ms,
         }
     }
 
+    fn node_creds_ok(creds: &HashMap<String, DeclaredCredentials>) -> bool {
+        creds.values()
+            .all(|cred| cred.validate_internode())
+    }
+
     pub fn set_nodes_credentials(
         &mut self,
-        nodes: HashMap<IpAddr, Vec<Credentials>>,
+        mut nodes: HashMap<String, DeclaredCredentials>,
     ) -> Result<(), Error> {
-        if nodes
-            .values()
-            .all(|creds|
-                creds
-                    .iter()
-                    .all(|cred|
-                        cred.ip().is_some() &&
-                        cred.kind().map(|k| k.is_internode()) == Some(true)))
-        {
-            self.nodes = nodes;
+        if Self::node_creds_ok(&nodes) {
+            let mut nodes_creds = self.nodes.write().expect("nodes credentials lock");
+            for (nodename, cred) in nodes.drain() {
+                let mut guard = DCredentialsResolveGuard::new(cred.clone(), self.resolve_sleep_period_ms);
+                if cred.ip().is_empty() && cred.hostname().is_some() {
+                    guard.set_in_progress();
+                    nodes_creds.insert(nodename, guard);
+                    self.spawn_resolver(cred);
+                } else {
+                    nodes_creds.insert(nodename, guard);
+                }
+            }
             Ok(())
         } else {
             let message = "nodes credentials missing ip or node name";
             Err(Error::CredentialsNotProvided(message.to_string()))
         }
     }
 
-    fn check_node_request(&self, node_name: &String, ip: Option<IpAddr>) -> Option<bool> {
-        if self.nodes.is_empty() {
-            warn!("nodes credentials not set");
+    fn spawn_resolver(&self, cred: DeclaredCredentials) {
+        tokio::spawn(Self::resolve_worker(self.nodes.clone(), cred, self.resolve_sleep_period_ms));
+    }
+
+    async fn resolve_worker(nodes: Arc<RwLock<NodesCredentials>>, cred: DeclaredCredentials, sleep_period_ms: u64) {
+        let hostname = cred.hostname().as_ref().expect("resolve worker without hostname");
+        let mut addr: Option<Vec<SocketAddr>> = lookup(hostname).await;
+        let mut cur_sleep_period_ms = 100;
+        while addr.is_none() || (addr.is_some() && addr.as_ref().unwrap().len() == 0) {
+            tokio::time::sleep(Duration::from_millis(cur_sleep_period_ms)).await;
+
+            addr = lookup(hostname).await;
+
+            cur_sleep_period_ms = sleep_period_ms.min(cur_sleep_period_ms * 2);
+        }
+
+        let addr = addr.expect("somehow addr is none");
+        if let CredentialsKind::InterNode(nodename) = cred.kind() {
+            let mut nodes = nodes.write().expect("nodes credentials lock");
+            if let Some(creds) = nodes.get_mut(nodename) {
+                creds.set_resolved(addr);
+            }
+        } else {
+            error!("resolved credentials are not internode");
+        }
+    }
+
+    fn process_auth_result(&self, nodename: &str, authenticated: bool) {
+        let mut nodes = self.nodes.write().expect("nodes credentials lock");
+        if let Some(node) = nodes.get_mut(nodename) {
+            if node.update_resolve_state(authenticated) {
+                node.set_in_progress();
+                self.spawn_resolver(node.creds().clone());
+            }
         }
-        self.nodes
-            .get(ip.as_ref()?)
-            .map(|creds| {
-                for cred in creds {
-                    if let Some(CredentialsKind::InterNode(other_name)) = cred.kind() {
-                        if node_name == other_name {
-                            return true;
-                        }
+    }
+
+    fn check_node_request(&self, node_name: &String, ip: Option<SocketAddr>) -> bool {
+        let mut authenticated = false;
+        let mut needs_update = false;
+        {
+            if ip.is_none() {
+                return false;
+            }
+            let ip = ip.unwrap().ip();
+            let nodes = self.nodes.read().expect("nodes credentials lock");
+            if let Some(guard) = nodes.get(node_name) {
+                let cred = guard.creds();
+                if let CredentialsKind::InterNode(other_name) = cred.kind() {
+                    debug_assert!(node_name == other_name);
+                    if cred.ip().iter().find(|cred_ip| cred_ip.ip() == ip).is_some() {
+                        authenticated = true;   
                     }
+                    needs_update = guard.needs_update(authenticated);
                 }
-                false
-            })
+            }
+        }
+        if needs_update {
+            self.process_auth_result(node_name, authenticated);
+        }
+        authenticated
     }
 
-    fn check_credentials_common(&self, credentials: Credentials) -> Result<Permissions, Error> {
+    fn check_credentials_common(&self, credentials: RequestCredentials) -> Result<Permissions, Error> {
         match credentials.kind() {
             Some(CredentialsKind::Basic { username, password }) => {
                 debug!(
@@ -101,11 +171,11 @@ impl<Storage> Authenticator for Basic<Storage>
 where
     Storage: UsersStorage,
 {
-    fn check_credentials_grpc(&self, credentials: Credentials) -> Result<Permissions, Error> {
+    fn check_credentials_grpc(&self, credentials: RequestCredentials) -> Result<Permissions, Error> {
         debug!("check {:?}", credentials);
         match credentials.kind() {
             Some(CredentialsKind::InterNode(node_name)) => {
-                if let Some(true) = self.check_node_request(node_name, credentials.ip()) {
+                if self.check_node_request(node_name, credentials.ip()) {
                     debug!("request from node: {:?}", credentials.ip());
                     Ok(Permissions::all())
                 } else {
@@ -116,7 +186,7 @@ where
         }
     }
 
-    fn check_credentials_rest(&self, credentials: Credentials) -> Result<Permissions, Error> {
+    fn check_credentials_rest(&self, credentials: RequestCredentials) -> Result<Permissions, Error> {
         debug!("check {:?}", credentials);
         self.check_credentials_common(credentials)
     }

bob-access/src/authenticator/mod.rs

@@ -2,13 +2,13 @@ pub mod basic;
 pub mod stub;
 mod users_storage;
 
-use crate::{credentials::Credentials, error::Error, permissions::Permissions};
+use crate::{credentials::RequestCredentials, error::Error, permissions::Permissions};
 
 pub use users_storage::{User, UsersMap};
 
 pub trait Authenticator: Clone + Send + Sync + 'static {
-    fn check_credentials_rest(&self, credentials: Credentials) -> Result<Permissions, Error>;
-    fn check_credentials_grpc(&self, credentials: Credentials) -> Result<Permissions, Error>;
+    fn check_credentials_rest(&self, credentials: RequestCredentials) -> Result<Permissions, Error>;
+    fn check_credentials_grpc(&self, credentials: RequestCredentials) -> Result<Permissions, Error>;
     fn credentials_type() -> AuthenticationType;
 }
 

bob-access/src/authenticator/stub.rs

@@ -1,4 +1,4 @@
-use crate::{credentials::Credentials, error::Error, permissions::Permissions};
+use crate::{credentials::RequestCredentials, error::Error, permissions::Permissions};
 
 use super::{Authenticator, AuthenticationType};
 
@@ -12,11 +12,11 @@ impl Stub {
 }
 
 impl Authenticator for Stub {
-    fn check_credentials_grpc(&self, _: Credentials) -> Result<Permissions, Error> {
+    fn check_credentials_grpc(&self, _: RequestCredentials) -> Result<Permissions, Error> {
         Ok(Permissions::all())
     }
 
-    fn check_credentials_rest(&self, _: Credentials) -> Result<Permissions, Error> {
+    fn check_credentials_rest(&self, _: RequestCredentials) -> Result<Permissions, Error> {
         Ok(Permissions::all())
     }