First impl of LDAP login

3 files changed, 193 insertions, 22 deletions

diff --git a/src/bayou.rs b/src/bayou.rs
index 56203eb..c9ae67f 100644
--- a/src/bayou.rs
+++ b/src/bayou.rs
@@ -1,6 +1,7 @@
 use std::time::{Duration, Instant};
 
 use anyhow::{anyhow, bail, Result};
+use log::debug;
 use rand::prelude::*;
 use serde::{Deserialize, Serialize};
 use tokio::io::AsyncReadExt;
@@ -76,14 +77,14 @@ impl<S: BayouState> Bayou<S> {
     pub async fn sync(&mut self) -> Result<()> {
         // 1. List checkpoints
         let checkpoints = self.list_checkpoints().await?;
-        eprintln!("(sync) listed checkpoints: {:?}", checkpoints);
+        debug!("(sync) listed checkpoints: {:?}", checkpoints);
 
         // 2. Load last checkpoint if different from currently used one
         let checkpoint = if let Some((ts, key)) = checkpoints.last() {
             if *ts == self.checkpoint.0 {
                 (*ts, None)
             } else {
-                eprintln!("(sync) loading checkpoint: {}", key);
+                debug!("(sync) loading checkpoint: {}", key);
 
                 let mut gor = GetObjectRequest::default();
                 gor.bucket = self.bucket.clone();
@@ -94,7 +95,7 @@ impl<S: BayouState> Bayou<S> {
                 let mut buf = Vec::with_capacity(obj_res.content_length.unwrap_or(128) as usize);
                 obj_body.into_async_read().read_to_end(&mut buf).await?;
 
-                eprintln!("(sync) checkpoint body length: {}", buf.len());
+                debug!("(sync) checkpoint body length: {}", buf.len());
 
                 let ck = open_deserialize::<S>(&buf, &self.key)?;
                 (*ts, Some(ck))
@@ -108,7 +109,7 @@ impl<S: BayouState> Bayou<S> {
         }
 
         if let Some(ck) = checkpoint.1 {
-            eprintln!(
+            debug!(
                 "(sync) updating checkpoint to loaded state at {:?}",
                 checkpoint.0
             );
@@ -123,7 +124,7 @@ impl<S: BayouState> Bayou<S> {
 
         // 3. List all operations starting from checkpoint
         let ts_ser = self.checkpoint.0.serialize();
-        eprintln!("(sync) looking up operations starting at {}", ts_ser);
+        debug!("(sync) looking up operations starting at {}", ts_ser);
         let ops_map = self
             .k2v
             .read_batch(&[BatchReadOp {
@@ -155,7 +156,7 @@ impl<S: BayouState> Bayou<S> {
             match &val.value[0] {
                 K2vValue::Value(v) => {
                     let op = open_deserialize::<S::Op>(&v, &self.key)?;
-                    eprintln!("(sync) operation {}: {} {:?}", tsstr, base64::encode(v), op);
+                    debug!("(sync) operation {}: {} {:?}", tsstr, base64::encode(v), op);
                     ops.push((ts, op));
                 }
                 K2vValue::Tombstone => {
@@ -164,7 +165,7 @@ impl<S: BayouState> Bayou<S> {
             }
         }
         ops.sort_by_key(|(ts, _)| *ts);
-        eprintln!("(sync) {} operations", ops.len());
+        debug!("(sync) {} operations", ops.len());
 
         if ops.len() < self.history.len() {
             bail!("Some operations have disappeared from storage!");
@@ -239,7 +240,7 @@ impl<S: BayouState> Bayou<S> {
     pub async fn push(&mut self, op: S::Op) -> Result<()> {
         self.check_recent_sync().await?;
 
-        eprintln!("(push) add operation: {:?}", op);
+        debug!("(push) add operation: {:?}", op);
 
         let ts = Timestamp::after(
             self.history
@@ -302,18 +303,18 @@ impl<S: BayouState> Bayou<S> {
         {
             Some(i) => i,
             None => {
-                eprintln!("(cp) Oldest operation is too recent to trigger checkpoint");
+                debug!("(cp) Oldest operation is too recent to trigger checkpoint");
                 return Ok(());
             }
         };
 
         if i_cp < CHECKPOINT_MIN_OPS {
-            eprintln!("(cp) Not enough old operations to trigger checkpoint");
+            debug!("(cp) Not enough old operations to trigger checkpoint");
             return Ok(());
         }
 
         let ts_cp = self.history[i_cp].0;
-        eprintln!(
+        debug!(
             "(cp) we could checkpoint at time {} (index {} in history)",
             ts_cp.serialize(),
             i_cp
@@ -321,13 +322,13 @@ impl<S: BayouState> Bayou<S> {
 
         // Check existing checkpoints: if last one is too recent, don't checkpoint again.
         let existing_checkpoints = self.list_checkpoints().await?;
-        eprintln!("(cp) listed checkpoints: {:?}", existing_checkpoints);
+        debug!("(cp) listed checkpoints: {:?}", existing_checkpoints);
 
         if let Some(last_cp) = existing_checkpoints.last() {
             if (ts_cp.msec as i128 - last_cp.0.msec as i128)
                 < CHECKPOINT_INTERVAL.as_millis() as i128
             {
-                eprintln!(
+                debug!(
                     "(cp) last checkpoint is too recent: {}, not checkpointing",
                     last_cp.0.serialize()
                 );
@@ -335,7 +336,7 @@ impl<S: BayouState> Bayou<S> {
             }
         }
 
-        eprintln!("(cp) saving checkpoint at {}", ts_cp.serialize());
+        debug!("(cp) saving checkpoint at {}", ts_cp.serialize());
 
         // Calculate state at time of checkpoint
         let mut last_known_state = (0, &self.checkpoint.1);
@@ -351,7 +352,7 @@ impl<S: BayouState> Bayou<S> {
 
         // Serialize and save checkpoint
         let cryptoblob = seal_serialize(&state_cp, &self.key)?;
-        eprintln!("(cp) checkpoint body length: {}", cryptoblob.len());
+        debug!("(cp) checkpoint body length: {}", cryptoblob.len());
 
         let mut por = PutObjectRequest::default();
         por.bucket = self.bucket.clone();
@@ -366,7 +367,7 @@ impl<S: BayouState> Bayou<S> {
 
             // Delete blobs
             for (_ts, key) in existing_checkpoints[..last_to_keep].iter() {
-                eprintln!("(cp) drop old checkpoint {}", key);
+                debug!("(cp) drop old checkpoint {}", key);
                 let mut dor = DeleteObjectRequest::default();
                 dor.bucket = self.bucket.clone();
                 dor.key = key.to_string();
diff --git a/src/config.rs b/src/config.rs
index ab40824..b77288b 100644
--- a/src/config.rs
+++ b/src/config.rs
@@ -41,8 +41,16 @@ pub struct LoginStaticUser {
 pub struct LoginLdapConfig {
     pub ldap_server: String,
 
-    pub search_dn: String,
+    #[serde(default)]
+    pub pre_bind_on_login: bool,
+    pub bind_dn: Option<String>,
+    pub bind_password: Option<String>,
+
+    pub search_base: String,
     pub username_attr: String,
+    #[serde(default = "default_mail_attr")]
+    pub mail_attr: String,
+
     pub aws_access_key_id_attr: String,
     pub aws_secret_access_key_attr: String,
     pub user_secret_attr: String,
@@ -62,3 +70,7 @@ pub fn read_config(config_file: PathBuf) -> Result<Config> {
 
     Ok(toml::from_str(&config)?)
 }
+
+fn default_mail_attr() -> String {
+    "mail".into()
+}
diff --git a/src/login/ldap_provider.rs b/src/login/ldap_provider.rs
index 54ddbd5..c9d23a0 100644
--- a/src/login/ldap_provider.rs
+++ b/src/login/ldap_provider.rs
@@ -1,23 +1,181 @@
 use anyhow::Result;
 use async_trait::async_trait;
+use ldap3::{LdapConnAsync, Scope, SearchEntry};
+use log::debug;
 use rusoto_signature::Region;
 
 use crate::config::*;
 use crate::login::*;
 
 pub struct LdapLoginProvider {
-    // TODO
+    k2v_region: Region,
+    s3_region: Region,
+    ldap_server: String,
+
+    pre_bind_on_login: bool,
+    bind_dn_and_pw: Option<(String, String)>,
+
+    search_base: String,
+    attrs_to_retrieve: Vec<String>,
+    username_attr: String,
+    mail_attr: String,
+
+    aws_access_key_id_attr: String,
+    aws_secret_access_key_attr: String,
+    user_secret_attr: String,
+    alternate_user_secrets_attr: Option<String>,
+
+    bucket_source: BucketSource,
+}
+
+enum BucketSource {
+    Constant(String),
+    Attr(String),
 }
 
 impl LdapLoginProvider {
-    pub fn new(_config: LoginLdapConfig, _k2v_region: Region, _s3_region: Region) -> Result<Self> {
-        unimplemented!()
+    pub fn new(config: LoginLdapConfig, k2v_region: Region, s3_region: Region) -> Result<Self> {
+        let bind_dn_and_pw = match (config.bind_dn, config.bind_password) {
+            (Some(dn), Some(pw)) => Some((dn, pw)),
+            (None, None) => None,
+            _ => bail!(
+                "If either of `bind_dn` or `bind_password` is set, the other must be set as well."
+            ),
+        };
+
+        let bucket_source = match (config.bucket, config.bucket_attr) {
+            (Some(b), None) => BucketSource::Constant(b),
+            (None, Some(a)) => BucketSource::Attr(a),
+            _ => bail!("Must set `bucket` or `bucket_attr`, but not both"),
+        };
+
+        if config.pre_bind_on_login && bind_dn_and_pw.is_none() {
+            bail!("Cannot use `pre_bind_on_login` without setting `bind_dn` and `bind_password`");
+        }
+
+        let mut attrs_to_retrieve = vec![
+            config.username_attr.clone(),
+            config.mail_attr.clone(),
+            config.aws_access_key_id_attr.clone(),
+            config.aws_secret_access_key_attr.clone(),
+            config.user_secret_attr.clone(),
+        ];
+        if let Some(a) = &config.alternate_user_secrets_attr {
+            attrs_to_retrieve.push(a.clone());
+        }
+        if let BucketSource::Attr(a) = &bucket_source {
+            attrs_to_retrieve.push(a.clone());
+        }
+
+        Ok(Self {
+            k2v_region,
+            s3_region,
+            ldap_server: config.ldap_server,
+            pre_bind_on_login: config.pre_bind_on_login,
+            bind_dn_and_pw,
+            search_base: config.search_base,
+            attrs_to_retrieve,
+            username_attr: config.username_attr,
+            mail_attr: config.mail_attr,
+            aws_access_key_id_attr: config.aws_access_key_id_attr,
+            aws_secret_access_key_attr: config.aws_secret_access_key_attr,
+            user_secret_attr: config.user_secret_attr,
+            alternate_user_secrets_attr: config.alternate_user_secrets_attr,
+            bucket_source,
+        })
     }
 }
 
 #[async_trait]
 impl LoginProvider for LdapLoginProvider {
-    async fn login(&self, _username: &str, _password: &str) -> Result<Credentials> {
-        unimplemented!()
+    async fn login(&self, username: &str, password: &str) -> Result<Credentials> {
+        let (conn, mut ldap) = LdapConnAsync::new(&self.ldap_server).await?;
+        ldap3::drive!(conn);
+
+        if self.pre_bind_on_login {
+            let (dn, pw) = self.bind_dn_and_pw.as_ref().unwrap();
+            ldap.simple_bind(dn, pw).await?.success()?;
+        }
+
+        let username_is_ok = username
+            .chars()
+            .all(|c| c.is_alphanumeric() || "-+_.@".contains(c));
+        if !username_is_ok {
+            bail!("Invalid username, must contain only a-z A-Z 0-9 - + _ . @");
+        }
+
+        let (matches, _res) = ldap
+            .search(
+                &self.search_base,
+                Scope::Subtree,
+                &format!(
+                    "(&(objectClass=inetOrgPerson)({}={}))",
+                    self.username_attr, username
+                ),
+                &self.attrs_to_retrieve,
+            )
+            .await?
+            .success()?;
+
+        if matches.is_empty() {
+            bail!("Invalid username");
+        }
+        if matches.len() > 1 {
+            bail!("Invalid username (multiple matching accounts)");
+        }
+        let user = SearchEntry::construct(matches.into_iter().next().unwrap());
+        debug!(
+            "Found matching LDAP user for username {}: {}",
+            username, user.dn
+        );
+
+        // Try to login against LDAP server with provided password
+        // to check user's password
+        ldap.simple_bind(&user.dn, password)
+            .await?
+            .success()
+            .context("Invalid password")?;
+        debug!("Ldap login with user name {} successfull", username);
+
+        let get_attr = |attr: &str| -> Result<String> {
+            Ok(user
+                .attrs
+                .get(attr)
+                .ok_or(anyhow!("Missing attr: {}", attr))?
+                .iter()
+                .next()
+                .ok_or(anyhow!("No value for attr: {}", attr))?
+                .clone())
+        };
+        let aws_access_key_id = get_attr(&self.aws_access_key_id_attr)?;
+        let aws_secret_access_key = get_attr(&self.aws_secret_access_key_attr)?;
+        let bucket = match &self.bucket_source {
+            BucketSource::Constant(b) => b.clone(),
+            BucketSource::Attr(a) => get_attr(a)?,
+        };
+
+        let storage = StorageCredentials {
+            k2v_region: self.k2v_region.clone(),
+            s3_region: self.s3_region.clone(),
+            aws_access_key_id,
+            aws_secret_access_key,
+            bucket,
+        };
+
+        let user_secret = get_attr(&self.user_secret_attr)?;
+        let alternate_user_secrets = match &self.alternate_user_secrets_attr {
+            None => vec![],
+            Some(a) => user.attrs.get(a).cloned().unwrap_or_default(),
+        };
+        let user_secrets = UserSecrets {
+            user_secret,
+            alternate_user_secrets,
+        };
+
+        drop(ldap);
+
+        let keys = CryptoKeys::open(&storage, &user_secrets, password).await?;
+
+        Ok(Credentials { storage, keys })
     }
 }