feat(query): Support check password policy when login

src/meta/app/src/principal/user_info.rs

@@ -15,6 +15,8 @@
 use core::fmt;
 use std::convert::TryFrom;
 
+use chrono::DateTime;
+use chrono::Utc;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use enumflags2::bitflags;
@@ -41,6 +43,14 @@ pub struct UserInfo {
     pub quota: UserQuota,
 
     pub option: UserOption,
+
+    pub history_auth_infos: Vec<AuthInfo>,
+
+    pub password_fail_ons: Vec<DateTime<Utc>>,
+
+    pub password_update_on: Option<DateTime<Utc>>,
+
+    pub lockout_time: Option<DateTime<Utc>>,
 }
 
 impl UserInfo {
@@ -57,6 +67,10 @@ impl UserInfo {
             grants,
             quota,
             option,
+            history_auth_infos: Vec::new(),
+            password_fail_ons: Vec::new(),
+            password_update_on: None,
+            lockout_time: None,
         }
     }
 
@@ -78,11 +92,35 @@ impl UserInfo {
     pub fn update_auth_option(&mut self, auth: Option<AuthInfo>, option: Option<UserOption>) {
         if let Some(auth_info) = auth {
             self.auth_info = auth_info;
+
+            // update password change history
+            self.history_auth_infos.push(self.auth_info.clone());
+            // Maximum 24 password records retained
+            if self.history_auth_infos.len() > 24 {
+                self.history_auth_infos.remove(0);
+            }
+            self.password_update_on = Some(Utc::now());
         };
         if let Some(user_option) = option {
             self.option = user_option;
         };
     }
+
+    pub fn update_login_fail_history(&mut self) {
+        self.password_fail_ons.push(Utc::now());
+        if self.password_fail_ons.len() > 10 {
+            self.password_fail_ons.remove(0);
+        }
+    }
+
+    pub fn clear_login_fail_history(&mut self) {
+        self.password_fail_ons = Vec::new();
+        self.lockout_time = None;
+    }
+
+    pub fn update_lockout_time(&mut self, lockout_time: DateTime<Utc>) {
+        self.lockout_time = Some(lockout_time);
+    }
 }
 
 impl TryFrom<Vec<u8>> for UserInfo {

src/meta/proto-conv/src/user_from_to_protobuf_impl.rs

@@ -312,6 +312,24 @@ impl FromToProto for mt::principal::UserInfo {
             option: mt::principal::UserOption::from_pb(p.option.ok_or_else(|| Incompatible {
                 reason: "UserInfo.option cannot be None".to_string(),
             })?)?,
+            history_auth_infos: p
+                .history_auth_infos
+                .iter()
+                .map(|a| mt::principal::AuthInfo::from_pb(a.clone()))
+                .collect::<Result<Vec<mt::principal::AuthInfo>, Incompatible>>()?,
+            password_fail_ons: p
+                .password_fail_ons
+                .iter()
+                .map(|t| DateTime::<Utc>::from_pb(t.clone()))
+                .collect::<Result<Vec<DateTime<Utc>>, Incompatible>>()?,
+            password_update_on: match p.password_update_on {
+                Some(t) => Some(DateTime::<Utc>::from_pb(t)?),
+                None => None,
+            },
+            lockout_time: match p.lockout_time {
+                Some(t) => Some(DateTime::<Utc>::from_pb(t)?),
+                None => None,
+            },
         })
     }
 
@@ -325,6 +343,24 @@ impl FromToProto for mt::principal::UserInfo {
             grants: Some(mt::principal::UserGrantSet::to_pb(&self.grants)?),
             quota: Some(mt::principal::UserQuota::to_pb(&self.quota)?),
             option: Some(mt::principal::UserOption::to_pb(&self.option)?),
+            history_auth_infos: self
+                .history_auth_infos
+                .iter()
+                .map(mt::principal::AuthInfo::to_pb)
+                .collect::<Result<Vec<pb::AuthInfo>, Incompatible>>()?,
+            password_fail_ons: self
+                .password_fail_ons
+                .iter()
+                .map(|t| t.to_pb())
+                .collect::<Result<Vec<String>, Incompatible>>()?,
+            password_update_on: match self.password_update_on {
+                Some(t) => Some(t.to_pb()?),
+                None => None,
+            },
+            lockout_time: match self.lockout_time {
+                Some(t) => Some(t.to_pb()?),
+                None => None,
+            },
         })
     }
 }

src/meta/proto-conv/tests/it/user_proto_conv.rs

@@ -67,6 +67,10 @@ fn test_user_info() -> mt::principal::UserInfo {
             max_storage_in_bytes: 20480,
         },
         option,
+        history_auth_infos: vec![],
+        password_fail_ons: vec![],
+        password_update_on: None,
+        lockout_time: None,
     }
 }
 

src/meta/proto-conv/tests/it/v050_user_info.rs

@@ -67,6 +67,10 @@ fn test_decode_v50_user_info() -> anyhow::Result<()> {
             .with_set_flag(databend_common_meta_app::principal::UserOptionFlag::TenantSetting)
             .with_default_role(Some("role1".into()))
             .with_network_policy(Some("mypolicy".to_string())),
+        history_auth_infos: vec![],
+        password_fail_ons: vec![],
+        password_update_on: None,
+        lockout_time: None,
     };
 
     common::test_pb_from_to(func_name!(), want())?;

src/meta/proto-conv/tests/it/v067_password_policy.rs

@@ -103,6 +103,10 @@ fn test_decode_v67_password_policy() -> anyhow::Result<()> {
             .with_default_role(Some("role1".into()))
             .with_network_policy(Some("mypolicy".to_string()))
             .with_password_policy(Some("testpasswordpolicy1".to_string())),
+        history_auth_infos: vec![],
+        password_fail_ons: vec![],
+        password_update_on: None,
+        lockout_time: None,
     };
 
     common::test_pb_from_to(func_name!(), want())?;

src/meta/protos/proto/user.proto

@@ -119,6 +119,10 @@ message UserInfo {
   UserGrantSet grants = 4;
   UserQuota quota = 5;
   UserOption option = 6;
+  repeated AuthInfo history_auth_infos = 7;
+  repeated string password_fail_ons = 8;
+  optional string password_update_on = 9;
+  optional string lockout_time = 10;
 }
 
 message UserIdentity {

src/query/service/src/interpreters/interpreter_user_create.rs

@@ -74,6 +74,10 @@ impl Interpreter for CreateUserInterpreter {
             grants: UserGrantSet::empty(),
             quota: UserQuota::no_limit(),
             option: plan.user_option,
+            history_auth_infos: Vec::new(),
+            password_fail_ons: Vec::new(),
+            password_update_on: None,
+            lockout_time: None,
         };
         user_mgr
             .add_user(&tenant, user_info, plan.if_not_exists)

src/query/service/src/servers/mysql/mysql_interactive_worker.rs

@@ -12,9 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use core::cmp::Ordering;
 use std::sync::Arc;
 use std::time::Instant;
 
+use chrono::Duration;
+use chrono::Utc;
 use databend_common_base::base::convert_byte_size;
 use databend_common_base::base::convert_number_size;
 use databend_common_base::base::tokio::io::AsyncWrite;
@@ -260,10 +263,72 @@ impl InteractiveWorkerBase {
         let identity = UserIdentity::new(&info.user_name, "%");
         let client_ip = info.user_client_address.split(':').collect::<Vec<_>>()[0];
         let user_info = UserApiProvider::instance()
-            .get_user_with_client_ip(&ctx.get_tenant(), identity, Some(client_ip))
+            .get_user_with_client_ip(&ctx.get_tenant(), identity.clone(), Some(client_ip))
             .await?;
 
+        // Lockout user can't login
+        if let Some(lockout_time) = user_info.lockout_time {
+            let now = Utc::now();
+            if let Ordering::Greater = lockout_time.cmp(&now) {
+                return Ok(false);
+            }
+        }
+
+        if let Some(name) = user_info.option.password_policy() {
+            if let Ok(password_policy) = UserApiProvider::instance()
+                .get_password_policy(&ctx.get_tenant(), name)
+                .await
+            {
+                // Check the number of password verification failures
+                if !user_info.password_fail_ons.is_empty() && password_policy.max_retries > 0 {
+                    let check_time = Utc::now()
+                        .checked_sub_signed(Duration::minutes(
+                            password_policy.lockout_time_mins as i64,
+                        ))
+                        .unwrap();
+
+                    let failed_retries = user_info
+                        .password_fail_ons
+                        .iter()
+                        .filter(|t| t.cmp(&&check_time) == Ordering::Greater)
+                        .count();
+
+                    // Too many failure retries, locked login
+                    if failed_retries > password_policy.max_retries as usize {
+                        let lockout_time = Utc::now()
+                            .checked_add_signed(Duration::minutes(
+                                password_policy.lockout_time_mins as i64,
+                            ))
+                            .unwrap();
+                        UserApiProvider::instance()
+                            .update_user_lockout_time(&ctx.get_tenant(), identity, lockout_time)
+                            .await?;
+
+                        return Ok(false);
+                    }
+                }
+
+                if password_policy.max_age_days > 0 {
+                    if let Some(password_update_on) = user_info.password_update_on {
+                        let max_change_time = password_update_on
+                            .checked_add_signed(Duration::days(password_policy.max_age_days as i64))
+                            .unwrap();
+
+                        let now = Utc::now();
+                        // Password has not been changed for more than max age days, cannot login
+                        if let Ordering::Less = max_change_time.cmp(&now) {
+                            return Ok(false);
+                        }
+                    }
+                }
+            }
+        }
+
         let authed = user_info.auth_info.auth_mysql(&info.user_password, salt)?;
+        // store login result
+        UserApiProvider::instance()
+            .update_user_login_result(&ctx.get_tenant(), identity, authed)
+            .await?;
         if authed {
             self.session.set_authed_user(user_info, None).await?;
         }

src/query/service/tests/it/storages/system.rs

@@ -393,6 +393,10 @@ async fn test_users_table() -> Result<()> {
                 grants: UserGrantSet::empty(),
                 quota: UserQuota::no_limit(),
                 option: UserOption::default(),
+                history_auth_infos: vec![],
+                password_fail_ons: vec![],
+                password_update_on: None,
+                lockout_time: None,
             },
             false,
         )
@@ -409,6 +413,10 @@ async fn test_users_table() -> Result<()> {
                 grants: UserGrantSet::empty(),
                 quota: UserQuota::no_limit(),
                 option: UserOption::default().with_default_role(Some("role1".to_string())),
+                history_auth_infos: vec![],
+                password_fail_ons: vec![],
+                password_update_on: None,
+                lockout_time: None,
             },
             false,
         )

src/query/sql/src/planner/binder/ddl/account.rs

@@ -12,6 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use core::cmp::Ordering;
+
+use chrono::Duration;
+use chrono::Utc;
 use databend_common_ast::ast::AccountMgrLevel;
 use databend_common_ast::ast::AccountMgrSource;
 use databend_common_ast::ast::AlterUserStmt;
@@ -23,6 +27,7 @@ use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_meta_app::principal::AuthInfo;
 use databend_common_meta_app::principal::GrantObject;
+use databend_common_meta_app::principal::UserInfo;
 use databend_common_meta_app::principal::UserOption;
 use databend_common_meta_app::principal::UserPrivilegeSet;
 use databend_common_users::UserApiProvider;
@@ -164,7 +169,8 @@ impl Binder {
         for option in user_options {
             option.apply(&mut user_option);
         }
-        self.verify_password(&user_option, auth_option).await?;
+        self.verify_password(&user_option, auth_option, None, None)
+            .await?;
 
         let plan = CreateUserPlan {
             user: user.clone(),
@@ -201,11 +207,17 @@ impl Binder {
 
         // None means no change to make
         let new_auth_info = if let Some(auth_option) = &auth_option {
-            // verify the password if changed
-            self.verify_password(&user_option, auth_option).await?;
             let auth_info = user_info
                 .auth_info
                 .alter2(&auth_option.auth_type, &auth_option.password)?;
+            // verify the password if changed
+            self.verify_password(
+                &user_option,
+                auth_option,
+                Some(&user_info),
+                Some(&auth_info),
+            )
+            .await?;
             if user_info.auth_info == auth_info {
                 None
             } else {
@@ -235,13 +247,64 @@ impl Binder {
         &mut self,
         user_option: &UserOption,
         auth_option: &AuthOption,
+        user_info: Option<&UserInfo>,
+        auth_info: Option<&AuthInfo>,
     ) -> Result<()> {
         if let (Some(name), Some(password)) = (user_option.password_policy(), &auth_option.password)
         {
             if let Ok(password_policy) = UserApiProvider::instance()
                 .get_password_policy(&self.ctx.get_tenant(), name)
                 .await
             {
+                // For changing password, there are two modification conditions that need to be met.
+                // 1. Verify that the current time must be more than the minimum allowed days
+                //    since last password changed.
+                // 2. The password cannot be repeated with history passwords.
+                if let (Some(user_info), Some(auth_info)) = (user_info, auth_info) {
+                    if password_policy.min_age_days > 0 {
+                        if let Some(password_update_on) = user_info.password_update_on {
+                            let allow_change_time = password_update_on
+                                .checked_add_signed(Duration::days(
+                                    password_policy.min_age_days as i64,
+                                ))
+                                .unwrap();
+
+                            let now = Utc::now();
+                            if let Ordering::Greater = allow_change_time.cmp(&now) {
+                                return Err(ErrorCode::InvalidPassword(format!(
+                                    "The time since the last change is too short, the password cannot be changed again before {}",
+                                    allow_change_time
+                                )));
+                            }
+                        }
+                    }
+
+                    if password_policy.history > 0 {
+                        let auth_type = auth_info.get_type();
+                        let password = auth_info.get_password();
+
+                        for (i, history_auth_info) in
+                            user_info.history_auth_infos.iter().rev().enumerate()
+                        {
+                            if i > password_policy.history as usize {
+                                break;
+                            }
+
+                            let history_auth_type = history_auth_info.get_type();
+                            let history_password = history_auth_info.get_password();
+
+                            // Using hash value of plain password to check may have false positives
+                            if auth_type == history_auth_type && password == history_password {
+                                return Err(ErrorCode::InvalidPassword(format!(
+                                    "The newly changed password cannot be repeated with the last {} passwords.",
+                                    password_policy.history
+                                )));
+                            }
+                        }
+                    }
+                }
+
+                // Verify the password complexity meets the requirements of the password policy
                 let analyzed = analyzer::analyze(password);
 
                 let mut invalids = Vec::new();