feat: 用户权限信息更新后，支持刷新已登陆用户session的缓存权限信息，让更新的权限能即时生效; #188

src/common/macros.rs

@@ -41,7 +41,7 @@ macro_rules! user_namespace_privilege {
     ($req:expr) => {{
         if let Some(session) = $req
             .extensions()
-            .get::<Arc<$crate::common::model::UserSession>>()
+            .get::<std::sync::Arc<$crate::common::model::UserSession>>()
         {
             session
                 .namespace_privilege
@@ -57,8 +57,8 @@ macro_rules! user_namespace_privilege {
 #[macro_export]
 macro_rules! user_no_namespace_permission {
     ($param:expr) => {{
-        return actix_web::HttpResponse::Ok().json(crate::common::model::ApiResult::<()>::error(
-            crate::common::error_code::NO_NAMESPACE_PERMISSION.to_string(),
+        return actix_web::HttpResponse::Ok().json($crate::common::model::ApiResult::<()>::error(
+            $crate::common::error_code::NO_NAMESPACE_PERMISSION.to_string(),
             Some(format!("user no such namespace permission: {:?}", $param)),
         ));
     }};

src/common/model/mod.rs

@@ -93,6 +93,8 @@ pub struct UserSession {
     pub roles: Vec<Arc<String>>,
     pub namespace_privilege: Option<PrivilegeGroup<Arc<String>>>,
     pub extend_infos: HashMap<String, String>,
+    /// 时间戳，单位秒
+    pub refresh_time: u32,
 }
 
 #[derive(Debug, Default, Clone, Deserialize, Serialize)]

src/console/config_api.rs

@@ -34,7 +34,17 @@ pub async fn query_config_list(
     request: web::Query<OpsConfigQueryListRequest>,
     config_addr: web::Data<Addr<ConfigActor>>,
 ) -> impl Responder {
-    let cmd = ConfigCmd::QueryPageInfo(Box::new(request.0.to_param(&req).unwrap()));
+    let param = request.0.to_param(&req).unwrap();
+    if !param
+        .namespace_privilege
+        .check_option_value_permission(&param.tenant, true)
+    {
+        return HttpResponse::Unauthorized().body(format!(
+            "user no such namespace permission: {:?}",
+            &param.tenant
+        ));
+    }
+    let cmd = ConfigCmd::QueryPageInfo(Box::new(param));
     match config_addr.send(cmd).await {
         Ok(res) => {
             let r: ConfigResult = res.unwrap();

src/console/login_api.rs

@@ -16,6 +16,7 @@ use crate::{
         crypto_utils,
         model::{ApiResult, UserSession},
     },
+    now_second_i32,
     raft::cache::{
         model::{CacheKey, CacheType, CacheValue},
         CacheLimiterReq, CacheManagerReq, CacheManagerResult,
@@ -114,6 +115,7 @@ pub async fn login(
                     roles: user.roles.unwrap_or_default(),
                     extend_infos: user.extend_info.unwrap_or_default(),
                     namespace_privilege: user.namespace_privilege,
+                    refresh_time: now_second_i32() as u32,
                 });
                 let cache_req = CacheManagerReq::Set {
                     key: CacheKey::new(CacheType::UserSession, token.clone()),

src/console/middle/login_middle.rs

@@ -2,7 +2,6 @@ use std::collections::HashMap;
 use std::future::{ready, Ready};
 use std::sync::Arc;
 
-use actix::Addr;
 use actix_http::{HttpMessage, StatusCode};
 use actix_web::{
     body::EitherBody,
@@ -14,9 +13,12 @@ use regex::Regex;
 
 use crate::common::appdata::AppShareData;
 use crate::common::model::{ApiResultOld, UserSession};
+use crate::now_second_i32;
 use crate::raft::cache::model::{CacheKey, CacheType, CacheValue};
-use crate::raft::cache::{CacheManager, CacheManagerReq, CacheManagerResult};
+use crate::raft::cache::{CacheManagerReq, CacheManagerResult, CacheUserChangeReq};
+use crate::user::model::UserDto;
 use crate::user::permission::UserRole;
+use crate::user::{UserManagerReq, UserManagerResult};
 
 lazy_static::lazy_static! {
     pub static ref IGNORE_CHECK_LOGIN: Vec<&'static str> = vec![
@@ -89,7 +91,7 @@ where
             "".to_owned()
         };
         let token = Arc::new(token);
-        let cache_manager = self.app_share_data.cache_manager.clone();
+        let app_share_data = self.app_share_data.clone();
         //request.parts()
         //let (http_request, _pl) = request.parts();
         //let http_request = http_request.to_owned();
@@ -104,11 +106,8 @@ where
             if is_check_path {
                 is_login = if token.is_empty() {
                     false
-                } else if let Ok(Some(session)) = get_user_session(
-                    &cache_manager,
-                    CacheManagerReq::Get(CacheKey::new(CacheType::UserSession, token.clone())),
-                )
-                .await
+                } else if let Ok(Some(session)) =
+                    get_user_session(&app_share_data, token.clone()).await
                 {
                     user_has_permission =
                         UserRole::match_url_by_roles(&session.roles, path, method);
@@ -184,11 +183,45 @@ where
 }
 
 async fn get_user_session(
-    cache_manager: &Addr<CacheManager>,
-    req: CacheManagerReq,
+    app_share_data: &AppShareData,
+    token: Arc<String>,
 ) -> anyhow::Result<Option<Arc<UserSession>>> {
-    match cache_manager.send(req).await?? {
+    let cache_key = CacheKey::new(CacheType::UserSession, token);
+    let req = CacheManagerReq::Get(cache_key.clone());
+    match app_share_data.cache_manager.send(req).await?? {
+        CacheManagerResult::ChangedValue(CacheValue::UserSession(session)) => {
+            match app_share_data
+                .user_manager
+                .send(UserManagerReq::Query {
+                    name: session.username.clone(),
+                })
+                .await??
+            {
+                UserManagerResult::QueryUser(Some(user)) => {
+                    let new_session = build_user_session(user);
+                    app_share_data
+                        .cache_manager
+                        .do_send(CacheUserChangeReq::UpdateUserSession {
+                            key: cache_key,
+                            session: new_session.clone(),
+                        });
+                    Ok(Some(new_session))
+                }
+                _ => Ok(None),
+            }
+        }
         CacheManagerResult::Value(CacheValue::UserSession(session)) => Ok(Some(session)),
         _ => Ok(None),
     }
 }
+
+fn build_user_session(user: UserDto) -> Arc<UserSession> {
+    Arc::new(UserSession {
+        username: user.username,
+        nickname: user.nickname,
+        roles: user.roles.unwrap_or_default(),
+        namespace_privilege: user.namespace_privilege,
+        extend_infos: user.extend_info.unwrap_or_default(),
+        refresh_time: now_second_i32() as u32,
+    })
+}

src/console/naming_api.rs

@@ -1,6 +1,6 @@
 #![allow(unused_imports)]
 
-use actix_web::{http::header, web, HttpRequest, HttpResponse, Responder};
+use actix_web::{http::header, web, HttpMessage, HttpRequest, HttpResponse, Responder};
 
 use actix::prelude::Addr;
 
@@ -9,31 +9,42 @@ use super::model::{
     PageResult,
 };
 use crate::naming::core::{NamingActor, NamingCmd, NamingResult};
+use crate::user_namespace_privilege;
 
 pub async fn query_ops_instances_list(
+    req: HttpRequest,
     param: web::Query<QueryAllInstanceListParam>,
     naming_addr: web::Data<Addr<NamingActor>>,
 ) -> impl Responder {
     match param.0.to_service_key() {
-        Ok(key) => match naming_addr.send(NamingCmd::QueryAllInstanceList(key)).await {
-            Ok(res) => match res as anyhow::Result<NamingResult> {
-                Ok(result) => match result {
-                    NamingResult::InstanceList(list) => {
-                        let resp = OpsNamingQueryListResponse {
-                            count: list.len() as u64,
-                            list,
-                        };
-                        let v = serde_json::to_string(&resp).unwrap();
-                        HttpResponse::Ok()
-                            .insert_header(header::ContentType(mime::APPLICATION_JSON))
-                            .body(v)
-                    }
-                    _ => HttpResponse::InternalServerError().body("error result"),
+        Ok(key) => {
+            let namespace_privilege = user_namespace_privilege!(req);
+            if !namespace_privilege.check_permission(&key.namespace_id) {
+                return HttpResponse::Unauthorized().body(format!(
+                    "user no such namespace permission: {}",
+                    &key.namespace_id
+                ));
+            }
+            match naming_addr.send(NamingCmd::QueryAllInstanceList(key)).await {
+                Ok(res) => match res as anyhow::Result<NamingResult> {
+                    Ok(result) => match result {
+                        NamingResult::InstanceList(list) => {
+                            let resp = OpsNamingQueryListResponse {
+                                count: list.len() as u64,
+                                list,
+                            };
+                            let v = serde_json::to_string(&resp).unwrap();
+                            HttpResponse::Ok()
+                                .insert_header(header::ContentType(mime::APPLICATION_JSON))
+                                .body(v)
+                        }
+                        _ => HttpResponse::InternalServerError().body("error result"),
+                    },
+                    Err(err) => HttpResponse::InternalServerError().body(err.to_string()),
                 },
                 Err(err) => HttpResponse::InternalServerError().body(err.to_string()),
-            },
-            Err(err) => HttpResponse::InternalServerError().body(err.to_string()),
-        },
+            }
+        }
         Err(err) => HttpResponse::InternalServerError().body(err.to_string()),
     }
 }

src/raft/cache/mod.rs

@@ -1,23 +1,22 @@
 // raft缓存数据
 
-use std::time::Duration;
-use std::{convert::TryInto, sync::Arc};
-
 use actix::prelude::*;
 use bean_factory::{bean, Inject};
 use inner_mem_cache::{MemCache, MemCacheMode};
 use ratelimiter_rs::RateLimiter;
 use serde::{Deserialize, Serialize};
-
-use crate::common::constant::CACHE_TREE_NAME;
-use crate::{common::limiter_utils::LimiterData, now_millis_i64, now_second_i32};
+use std::collections::HashMap;
+use std::time::Duration;
+use std::{convert::TryInto, sync::Arc};
 
 use self::model::{CacheItemDo, CacheKey, CacheValue};
-
 use super::db::{
     route::TableRoute,
     table::{TableManager, TableManagerQueryReq, TableManagerReq, TableManagerResult},
 };
+use crate::common::constant::CACHE_TREE_NAME;
+use crate::common::model::UserSession;
+use crate::{common::limiter_utils::LimiterData, now_millis_i64, now_second_i32};
 
 pub mod api;
 pub mod model;
@@ -29,6 +28,7 @@ pub struct CacheManager {
     //default_timeout: i32,
     raft_table_route: Option<Arc<TableRoute>>,
     table_manager: Option<Addr<TableManager>>,
+    user_privilege_change_time: HashMap<Arc<String>, u32>,
 }
 
 impl Default for CacheManager {
@@ -46,6 +46,7 @@ impl CacheManager {
             //default_timeout: 1200,
             raft_table_route: None,
             table_manager: None,
+            user_privilege_change_time: HashMap::new(),
         }
     }
 
@@ -106,6 +107,19 @@ impl CacheManager {
             table_manager.do_send(req);
         }
     }
+
+    fn user_privilege_has_changed(&self, v: &CacheValue) -> bool {
+        if let CacheValue::UserSession(session) = &v {
+            if session.refresh_time > 0 {
+                if let Some(change_time) = self.user_privilege_change_time.get(&session.username) {
+                    if *change_time > session.refresh_time {
+                        return true;
+                    }
+                }
+            }
+        }
+        false
+    }
 }
 
 impl Actor for CacheManager {
@@ -187,10 +201,27 @@ pub enum CacheLimiterReq {
     },
 }
 
+#[derive(Message, Clone, Debug)]
+#[rtype(result = "anyhow::Result<CacheManagerResult>")]
+pub enum CacheUserChangeReq {
+    UserPrivilegeChange {
+        username: Arc<String>,
+        change_time: u32,
+    },
+    RemoveUser {
+        username: Arc<String>,
+    },
+    UpdateUserSession {
+        key: CacheKey,
+        session: Arc<UserSession>,
+    },
+}
+
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub enum CacheManagerResult {
     None,
     Value(CacheValue),
+    ChangedValue(CacheValue),
     Limiter(bool),
 }
 
@@ -260,7 +291,13 @@ impl Handler<CacheManagerReq> for CacheManager {
         .map(
             |inner_ctx: anyhow::Result<CacheManagerInnerCtx>, act, _| match inner_ctx? {
                 CacheManagerInnerCtx::Get(key) => match act.cache.get(&key) {
-                    Ok(v) => Ok(CacheManagerResult::Value(v)),
+                    Ok(v) => {
+                        if act.user_privilege_has_changed(&v) {
+                            Ok(CacheManagerResult::ChangedValue(v))
+                        } else {
+                            Ok(CacheManagerResult::Value(v))
+                        }
+                    }
                     Err(_) => Ok(CacheManagerResult::None),
                 },
                 CacheManagerInnerCtx::Remove(key) => {
@@ -339,3 +376,29 @@ impl Handler<CacheLimiterReq> for CacheManager {
         Ok(CacheManagerResult::Limiter(r))
     }
 }
+
+impl Handler<CacheUserChangeReq> for CacheManager {
+    type Result = anyhow::Result<CacheManagerResult>;
+
+    fn handle(&mut self, msg: CacheUserChangeReq, _ctx: &mut Self::Context) -> Self::Result {
+        match msg {
+            CacheUserChangeReq::UserPrivilegeChange {
+                username: user_name,
+                change_time,
+            } => {
+                self.user_privilege_change_time
+                    .insert(user_name, change_time);
+            }
+            CacheUserChangeReq::RemoveUser {
+                username: user_name,
+            } => {
+                self.user_privilege_change_time.remove(&user_name);
+            }
+            CacheUserChangeReq::UpdateUserSession { key, session } => {
+                let ttl = self.cache.time_to_live(&key);
+                self.cache.set(key, CacheValue::UserSession(session), ttl);
+            }
+        }
+        Ok(CacheManagerResult::None)
+    }
+}