nacos_sdk/api/plugin/auth/
auth_by_http.rs

1use arc_swap::ArcSwap;
2use rand::Rng;
3use std::ops::{Add, Deref};
4use std::sync::Arc;
5use tokio::time::{Duration, Instant};
6
7use crate::api::plugin::{AuthContext, AuthPlugin, LoginIdentityContext};
8
9use super::RequestResource;
10
11pub const USERNAME: &str = "username";
12
13pub const PASSWORD: &str = "password";
14
15pub(crate) const ACCESS_TOKEN: &str = "accessToken";
16
17#[allow(dead_code)]
18pub(crate) const TOKEN_TTL: &str = "tokenTtl";
19
20/// Http login AuthPlugin.
21pub struct HttpLoginAuthPlugin {
22    login_identity: ArcSwap<LoginIdentityContext>,
23    next_login_refresh: ArcSwap<Instant>,
24}
25
26impl Default for HttpLoginAuthPlugin {
27    fn default() -> Self {
28        Self {
29            login_identity: ArcSwap::from_pointee(LoginIdentityContext::default()),
30            next_login_refresh: ArcSwap::from_pointee(Instant::now()),
31        }
32    }
33}
34
35#[async_trait::async_trait]
36impl AuthPlugin for HttpLoginAuthPlugin {
37    async fn login(&self, server_list: Vec<String>, auth_context: AuthContext) {
38        let now_instant = Instant::now();
39        if now_instant.le(self.next_login_refresh.load().deref()) {
40            tracing::debug!("Http login return because now_instant lte next_login_refresh.");
41            return;
42        }
43
44        let username = auth_context.params.get(USERNAME).unwrap().to_owned();
45        let password = auth_context.params.get(PASSWORD).unwrap().to_owned();
46
47        let server_addr = {
48            // random one
49            server_list
50                .get(rand::thread_rng().gen_range(0..server_list.len()))
51                .unwrap()
52                .to_string()
53        };
54
55        let scheme = if cfg!(feature = "tls") {
56            "https"
57        } else {
58            "http"
59        };
60        let login_url = format!("{scheme}://{server_addr}/nacos/v1/auth/login");
61
62        tracing::debug!("Http login with username={username},password={password}");
63
64        let login_response = {
65            let resp = reqwest::Client::new()
66                .post(login_url)
67                .query(&[(USERNAME, username), (PASSWORD, password)])
68                .send()
69                .await;
70            tracing::debug!("Http login resp={resp:?}");
71
72            match resp {
73                Err(e) => {
74                    tracing::error!("Http login error, send response failed, err={e:?}");
75                    None
76                }
77                Ok(resp) => {
78                    let resp_text = resp.text().await.unwrap();
79                    let resp_obj = serde_json::from_str::<HttpLoginResponse>(&resp_text);
80                    match resp_obj {
81                        Err(e) => {
82                            tracing::error!("Http login error, resp_text={resp_text}, err={e:?}");
83                            None
84                        }
85                        Ok(resp_obj) => Some(resp_obj),
86                    }
87                }
88            }
89        };
90
91        if let Some(login_response) = login_response {
92            let delay_sec = login_response.token_ttl / 10;
93            let new_login_identity = Arc::new(
94                LoginIdentityContext::default()
95                    .add_context(ACCESS_TOKEN, login_response.access_token),
96            );
97            self.login_identity.store(new_login_identity);
98
99            self.next_login_refresh
100                .store(Arc::new(Instant::now().add(Duration::from_secs(delay_sec))));
101        }
102    }
103
104    fn get_login_identity(&self, _: RequestResource) -> LoginIdentityContext {
105        self.login_identity.load().deref().deref().to_owned()
106    }
107}
108
109#[derive(Default, serde::Deserialize)]
110#[serde(rename_all = "camelCase")]
111struct HttpLoginResponse {
112    access_token: String,
113    token_ttl: u64,
114}
115
116#[cfg(test)]
117mod tests {
118    use crate::api::plugin::{AuthContext, AuthPlugin, HttpLoginAuthPlugin, RequestResource};
119
120    #[tokio::test]
121    #[ignore]
122    async fn test_http_login_auth_plugin() {
123        tracing_subscriber::fmt()
124            .with_max_level(tracing::Level::DEBUG)
125            .init();
126
127        let http_auth_plugin = HttpLoginAuthPlugin::default();
128        let server_list = vec!["127.0.0.1:8848".to_string()];
129
130        let auth_context = AuthContext::default()
131            .add_param(crate::api::plugin::USERNAME, "nacos")
132            .add_param(crate::api::plugin::PASSWORD, "nacos");
133
134        http_auth_plugin
135            .login(server_list.clone(), auth_context.clone())
136            .await;
137        let login_identity_1 = http_auth_plugin.get_login_identity(RequestResource::default());
138        assert_eq!(login_identity_1.contexts.len(), 1);
139
140        tokio::time::sleep(tokio::time::Duration::from_millis(111)).await;
141
142        http_auth_plugin.login(server_list, auth_context).await;
143        let login_identity_2 = http_auth_plugin.get_login_identity(RequestResource::default());
144        assert_eq!(login_identity_1.contexts, login_identity_2.contexts)
145    }
146}
nacos_sdk/api/plugin/auth/auth_by_http.rs

nacos_sdk/api/plugin/auth/
auth_by_http.rs
