rocketmq_auth/authentication/strategy/
stateless_authentication_strategy.rs

1// Copyright 2023 The RocketMQ Rust Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Stateless authentication strategy without caching.
16//!
17//! This strategy performs authentication for every request without caching results.
18//! Each authentication is independent and does not rely on historical state.
19
20use std::collections::HashSet;
21use std::sync::Arc;
22
23use crate::authentication::context::default_authentication_context::DefaultAuthenticationContext;
24use crate::authentication::provider::AuthenticationProvider;
25use crate::authentication::strategy::abstract_authentication_strategy::AbstractAuthenticationStrategy;
26use crate::authentication::strategy::authenticate_with_provider;
27use crate::authentication::strategy::authentication_strategy::AuthenticationStrategy;
28use crate::authentication::strategy::AuthenticationFuture;
29use crate::authorization::context::authentication_context::AuthenticationContext;
30use crate::config::AuthConfig;
31use crate::AuthFailureKind;
32use crate::AuthOperation;
33use crate::AuthServiceError;
34use crate::AuthServiceResult;
35
36/// Stateless authentication strategy.
37///
38/// Performs authentication for every request without any caching or state retention.
39/// This ensures that each authentication check is independent and current.
40///
41/// - No state shared between requests
42/// - No caching of authentication results
43/// - Suitable for scenarios requiring fresh validation on every request
44pub struct StatelessAuthenticationStrategy<P>
45where
46    P: AuthenticationProvider<Context = DefaultAuthenticationContext>,
47{
48    auth_config: AuthConfig,
49    authentication_white_set: HashSet<String>,
50    authentication_provider: Option<Arc<P>>,
51}
52
53impl<P> StatelessAuthenticationStrategy<P>
54where
55    P: AuthenticationProvider<Context = DefaultAuthenticationContext> + Send + Sync + 'static,
56{
57    /// Create a new stateless authentication strategy.
58    pub fn new(auth_config: AuthConfig, provider: Option<Arc<P>>) -> Self {
59        let mut authentication_white_set = HashSet::new();
60
61        let whitelist = auth_config.authentication_whitelist.to_string();
62        if !whitelist.is_empty() {
63            for rpc_code in whitelist.split(',') {
64                let trimmed = rpc_code.trim();
65                if !trimmed.is_empty() {
66                    authentication_white_set.insert(trimmed.to_string());
67                }
68            }
69        }
70
71        Self {
72            auth_config,
73            authentication_white_set,
74            authentication_provider: provider,
75        }
76    }
77
78    /// Perform authentication without caching.
79    async fn do_authenticate_internal(&self, context: &dyn AuthenticationContext) -> AuthServiceResult<()> {
80        if !self.auth_config.authentication_enabled {
81            return Ok(());
82        }
83
84        let default_context = context
85            .as_any()
86            .downcast_ref::<DefaultAuthenticationContext>()
87            .ok_or_else(|| AuthServiceError::new(AuthOperation::Authenticate, AuthFailureKind::InvalidInput))?;
88
89        if let Some(rpc_code) = default_context.base.rpc_code() {
90            if self.is_whitelisted(rpc_code.as_str()) {
91                return Ok(());
92            }
93        }
94
95        let provider = match &self.authentication_provider {
96            Some(p) => p,
97            None => return Ok(()),
98        };
99
100        authenticate_with_provider(provider.as_ref(), default_context).await
101    }
102
103    pub fn provider(&self) -> Option<&P> {
104        self.authentication_provider.as_deref()
105    }
106}
107
108impl<P> AbstractAuthenticationStrategy for StatelessAuthenticationStrategy<P>
109where
110    P: AuthenticationProvider<Context = DefaultAuthenticationContext> + Send + Sync + 'static,
111{
112    fn authentication_white_set(&self) -> &HashSet<String> {
113        &self.authentication_white_set
114    }
115}
116
117impl<P> AuthenticationStrategy for StatelessAuthenticationStrategy<P>
118where
119    P: AuthenticationProvider<Context = DefaultAuthenticationContext> + Send + Sync + 'static,
120{
121    fn authenticate<'a>(&'a self, context: &'a dyn AuthenticationContext) -> AuthenticationFuture<'a> {
122        Box::pin(self.do_authenticate_internal(context))
123    }
124}
125
126#[cfg(test)]
127mod tests {
128    use std::any::Any;
129
130    use super::*;
131    use crate::authentication::provider::DefaultAuthenticationProvider;
132    use cheetah_string::CheetahString;
133
134    struct CountingAuthenticationProvider {
135        calls: Arc<std::sync::atomic::AtomicUsize>,
136    }
137
138    impl AuthenticationProvider for CountingAuthenticationProvider {
139        type Context = DefaultAuthenticationContext;
140
141        async fn initialize(
142            &mut self,
143            _config: AuthConfig,
144            _metadata_service: Option<Arc<dyn Any + Send + Sync>>,
145        ) -> AuthServiceResult<()> {
146            Ok(())
147        }
148
149        async fn authenticate(&self, _context: &Self::Context) -> AuthServiceResult<()> {
150            self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
151            Ok(())
152        }
153
154        fn new_context_from_metadata(
155            &self,
156            _metadata: &std::collections::HashMap<String, String>,
157            _request: Box<dyn Any + Send>,
158        ) -> Self::Context {
159            DefaultAuthenticationContext::new()
160        }
161
162        fn new_context_from_command(
163            &self,
164            _command: &rocketmq_protocol::protocol::remoting_command::RemotingCommand,
165        ) -> Self::Context {
166            DefaultAuthenticationContext::new()
167        }
168    }
169
170    #[test]
171    fn test_stateless_strategy_creation() {
172        let config = AuthConfig::default();
173        let provider = Arc::new(DefaultAuthenticationProvider::new());
174
175        let strategy = StatelessAuthenticationStrategy::new(config, Some(provider));
176
177        assert!(strategy.provider().is_some());
178        assert!(strategy.authentication_white_set().is_empty());
179    }
180
181    #[tokio::test]
182    async fn test_authentication_disabled() {
183        let config = AuthConfig {
184            authentication_enabled: false,
185            ..Default::default()
186        };
187
188        let strategy: StatelessAuthenticationStrategy<DefaultAuthenticationProvider> =
189            StatelessAuthenticationStrategy::new(config, None);
190
191        let context = DefaultAuthenticationContext::new();
192        let result = strategy.authenticate(&context).await;
193        assert!(result.is_ok());
194    }
195
196    #[test]
197    fn test_whitelist_parsing() {
198        let config = AuthConfig {
199            authentication_whitelist: "SEND_MESSAGE,PULL_MESSAGE".into(),
200            ..Default::default()
201        };
202
203        let strategy: StatelessAuthenticationStrategy<DefaultAuthenticationProvider> =
204            StatelessAuthenticationStrategy::new(config, None);
205
206        assert!(strategy.authentication_white_set().contains("SEND_MESSAGE"));
207        assert!(strategy.authentication_white_set().contains("PULL_MESSAGE"));
208    }
209
210    #[test]
211    fn test_whitelist_with_spaces() {
212        let config = AuthConfig {
213            authentication_whitelist: " SEND_MESSAGE , PULL_MESSAGE , QUERY_MESSAGE ".into(),
214            ..Default::default()
215        };
216
217        let strategy: StatelessAuthenticationStrategy<DefaultAuthenticationProvider> =
218            StatelessAuthenticationStrategy::new(config, None);
219
220        assert_eq!(strategy.authentication_white_set().len(), 3);
221        assert!(strategy.authentication_white_set().contains("SEND_MESSAGE"));
222        assert!(strategy.authentication_white_set().contains("PULL_MESSAGE"));
223        assert!(strategy.authentication_white_set().contains("QUERY_MESSAGE"));
224    }
225
226    #[tokio::test]
227    async fn test_authentication_enabled_without_provider() {
228        let config = AuthConfig {
229            authentication_enabled: true,
230            ..Default::default()
231        };
232
233        let strategy: StatelessAuthenticationStrategy<DefaultAuthenticationProvider> =
234            StatelessAuthenticationStrategy::new(config, None);
235
236        let context = DefaultAuthenticationContext::new();
237        let result = strategy.authenticate(&context).await;
238        assert!(result.is_ok());
239    }
240
241    #[tokio::test]
242    async fn test_whitelisted_rpc_code() {
243        let config = AuthConfig {
244            authentication_enabled: true,
245            authentication_whitelist: "100,200".into(),
246            ..Default::default()
247        };
248
249        let strategy: StatelessAuthenticationStrategy<DefaultAuthenticationProvider> =
250            StatelessAuthenticationStrategy::new(config, None);
251
252        let mut context = DefaultAuthenticationContext::new();
253        context.base.set_rpc_code(Some(CheetahString::from("100")));
254
255        let result = strategy.authenticate(&context).await;
256        assert!(result.is_ok());
257    }
258
259    #[tokio::test]
260    async fn test_no_caching_behavior() {
261        let config = AuthConfig {
262            authentication_enabled: false,
263            ..Default::default()
264        };
265
266        let strategy: StatelessAuthenticationStrategy<DefaultAuthenticationProvider> =
267            StatelessAuthenticationStrategy::new(config, None);
268
269        let context = DefaultAuthenticationContext::new();
270
271        let result1 = strategy.authenticate(&context).await;
272        assert!(result1.is_ok());
273
274        let result2 = strategy.authenticate(&context).await;
275        assert!(result2.is_ok());
276    }
277
278    #[tokio::test(flavor = "current_thread")]
279    async fn authenticates_inside_current_thread_runtime_without_block_in_place_panic() {
280        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
281        let provider = Arc::new(CountingAuthenticationProvider { calls: calls.clone() });
282        let strategy = StatelessAuthenticationStrategy::new(
283            AuthConfig {
284                authentication_enabled: true,
285                ..Default::default()
286            },
287            Some(provider),
288        );
289        let context = DefaultAuthenticationContext::new();
290        assert!(strategy.authenticate(&context).await.is_ok());
291
292        assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 1);
293    }
294}
rocketmq_auth/authentication/strategy/stateless_authentication_strategy.rs

rocketmq_auth/authentication/strategy/
stateless_authentication_strategy.rs
