1use std::future::Future;
2use std::pin::Pin;
3use std::sync::Arc;
4use std::task::{Context, Poll};
5
6use tower::{Layer, Service};
7
8use camel_api::security_policy::{
9 AuthorizationDecision, SecurityPolicy, store_principal_properties,
10};
11use camel_api::{CamelError, Exchange};
12
13#[derive(Clone)]
14pub struct SecurityPolicyLayer {
15 policy: Arc<dyn SecurityPolicy>,
16}
17
18impl SecurityPolicyLayer {
19 pub fn new(policy: Arc<dyn SecurityPolicy>) -> Self {
20 Self { policy }
21 }
22}
23
24impl<S> Layer<S> for SecurityPolicyLayer {
25 type Service = SecurityPolicyService<S>;
26
27 fn layer(&self, inner: S) -> Self::Service {
28 SecurityPolicyService {
29 inner,
30 policy: Arc::clone(&self.policy),
31 }
32 }
33}
34
35pub struct SecurityPolicyService<S> {
36 inner: S,
37 policy: Arc<dyn SecurityPolicy>,
38}
39
40impl<S: Clone> Clone for SecurityPolicyService<S> {
41 fn clone(&self) -> Self {
42 Self {
43 inner: self.inner.clone(),
44 policy: Arc::clone(&self.policy),
45 }
46 }
47}
48
49impl<S> Service<Exchange> for SecurityPolicyService<S>
50where
51 S: Service<Exchange, Response = Exchange, Error = CamelError> + Clone + Send + 'static,
52 S::Future: Send,
53{
54 type Response = Exchange;
55 type Error = CamelError;
56 type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
57
58 fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
59 self.inner.poll_ready(cx)
60 }
61
62 fn call(&mut self, mut exchange: Exchange) -> Self::Future {
63 let policy = Arc::clone(&self.policy);
64 let clone = self.inner.clone();
65 let mut inner = std::mem::replace(&mut self.inner, clone);
66
67 Box::pin(async move {
68 match policy.evaluate(&mut exchange).await {
69 Ok(AuthorizationDecision::Granted { principal }) => {
70 store_principal_properties(&mut exchange, &principal);
71 inner.call(exchange).await
72 }
73 Ok(AuthorizationDecision::Denied {
74 reason,
75 required,
76 actual,
77 }) => {
78 let msg = format!(
79 "Access denied: {reason}. Required: {required:?}, actual: {actual:?}"
80 );
81 Err(CamelError::Unauthorized(msg))
82 }
83 Err(e) => Err(e),
84 _ => Err(CamelError::Unauthorized(
86 "access denied by security policy".to_string(),
87 )),
88 }
89 })
90 }
91}
92
93#[cfg(test)]
94mod tests {
95 use super::*;
96 use async_trait::async_trait;
97 use camel_api::security_policy::{
98 PRINCIPAL_AUDIENCE_KEY, PRINCIPAL_CLAIMS_KEY, PRINCIPAL_ISSUER_KEY, PRINCIPAL_KEY,
99 PRINCIPAL_ROLES_KEY, PRINCIPAL_SCOPES_KEY, PRINCIPAL_SUBJECT_KEY, Principal,
100 };
101 use camel_api::{BoxProcessor, BoxProcessorExt, Message};
102 use std::sync::atomic::{AtomicU32, Ordering};
103 use tower::ServiceExt;
104
105 fn make_exchange() -> Exchange {
106 Exchange::new(Message::new("test"))
107 }
108
109 fn ok_processor() -> BoxProcessor {
110 BoxProcessor::from_fn(|ex| Box::pin(async move { Ok(ex) }))
111 }
112
113 fn test_principal() -> Principal {
114 Principal {
115 subject: "user1".into(),
116 issuer: "test-issuer".into(),
117 audience: vec!["api".into()],
118 scopes: vec!["read".into()],
119 roles: vec!["admin".into()],
120 claims: serde_json::json!({"sub": "user1"}),
121 }
122 }
123
124 struct GrantPolicy;
125 #[async_trait]
126 impl SecurityPolicy for GrantPolicy {
127 async fn evaluate(
128 &self,
129 _exchange: &mut Exchange,
130 ) -> Result<AuthorizationDecision, CamelError> {
131 Ok(AuthorizationDecision::Granted {
132 principal: test_principal(),
133 })
134 }
135 }
136
137 struct DenyPolicy;
138 #[async_trait]
139 impl SecurityPolicy for DenyPolicy {
140 async fn evaluate(
141 &self,
142 _exchange: &mut Exchange,
143 ) -> Result<AuthorizationDecision, CamelError> {
144 Ok(AuthorizationDecision::Denied {
145 reason: "missing role".into(),
146 required: vec!["admin".into()],
147 actual: vec!["user".into()],
148 })
149 }
150 }
151
152 struct FailPolicy;
153 #[async_trait]
154 impl SecurityPolicy for FailPolicy {
155 async fn evaluate(
156 &self,
157 _exchange: &mut Exchange,
158 ) -> Result<AuthorizationDecision, CamelError> {
159 Err(CamelError::Unauthenticated("invalid token".into()))
160 }
161 }
162
163 #[tokio::test]
164 async fn test_granted_stores_properties() {
165 let layer = SecurityPolicyLayer::new(Arc::new(GrantPolicy));
166 let mut svc = layer.layer(ok_processor());
167 let result = svc.ready().await.unwrap().call(make_exchange()).await;
168 assert!(result.is_ok());
169 let ex = result.unwrap();
170 assert_eq!(
171 ex.property(PRINCIPAL_SUBJECT_KEY),
172 Some(&serde_json::Value::String("user1".into()))
173 );
174 assert_eq!(
175 ex.property(PRINCIPAL_ISSUER_KEY),
176 Some(&serde_json::Value::String("test-issuer".into()))
177 );
178 assert!(ex.property(PRINCIPAL_KEY).is_some());
179 }
180
181 #[tokio::test]
182 async fn test_denied_returns_unauthorized_error() {
183 let layer = SecurityPolicyLayer::new(Arc::new(DenyPolicy));
184 let mut svc = layer.layer(ok_processor());
185 let result = svc.ready().await.unwrap().call(make_exchange()).await;
186 assert!(result.is_err());
187 match result.unwrap_err() {
188 CamelError::Unauthorized(msg) => assert!(msg.contains("missing role")),
189 other => panic!("expected Unauthorized, got: {other:?}"),
190 }
191 }
192
193 #[tokio::test]
194 async fn test_denied_error_contains_required_actual() {
195 let layer = SecurityPolicyLayer::new(Arc::new(DenyPolicy));
196 let mut svc = layer.layer(ok_processor());
197 let result = svc.ready().await.unwrap().call(make_exchange()).await;
198 let msg = match result.unwrap_err() {
199 CamelError::Unauthorized(msg) => msg,
200 other => panic!("expected Unauthorized, got: {other:?}"),
201 };
202 assert!(msg.contains("admin"));
203 assert!(msg.contains("user"));
204 }
205
206 #[tokio::test]
207 async fn test_evaluate_error_propagates() {
208 let layer = SecurityPolicyLayer::new(Arc::new(FailPolicy));
209 let mut svc = layer.layer(ok_processor());
210 let result = svc.ready().await.unwrap().call(make_exchange()).await;
211 match result.unwrap_err() {
212 CamelError::Unauthenticated(msg) => assert!(msg.contains("invalid token")),
213 other => panic!("expected Unauthenticated, got: {other:?}"),
214 }
215 }
216
217 #[tokio::test]
218 async fn test_multiple_calls_share_policy() {
219 let count = Arc::new(AtomicU32::new(0));
220 struct CountingPolicy {
221 count: Arc<AtomicU32>,
222 }
223 #[async_trait]
224 impl SecurityPolicy for CountingPolicy {
225 async fn evaluate(
226 &self,
227 _exchange: &mut Exchange,
228 ) -> Result<AuthorizationDecision, CamelError> {
229 self.count.fetch_add(1, Ordering::SeqCst);
230 Ok(AuthorizationDecision::Granted {
231 principal: Principal {
232 subject: "user1".into(),
233 issuer: "test".into(),
234 audience: vec![],
235 scopes: vec![],
236 roles: vec![],
237 claims: serde_json::Value::Null,
238 },
239 })
240 }
241 }
242 let policy = Arc::new(CountingPolicy {
243 count: Arc::clone(&count),
244 });
245 let layer = SecurityPolicyLayer::new(Arc::clone(&policy) as Arc<dyn SecurityPolicy>);
246 let mut svc = layer.layer(ok_processor());
247 for _ in 0..3 {
248 let result = svc.ready().await.unwrap().call(make_exchange()).await;
249 assert!(result.is_ok());
250 }
251 assert_eq!(count.load(Ordering::SeqCst), 3);
252 }
253
254 #[tokio::test]
255 async fn test_granted_all_property_json_formats() {
256 let layer = SecurityPolicyLayer::new(Arc::new(GrantPolicy));
257 let mut svc = layer.layer(ok_processor());
258 let result = svc.ready().await.unwrap().call(make_exchange()).await;
259 let ex = result.unwrap();
260
261 let roles: Vec<String> =
262 serde_json::from_str(ex.property(PRINCIPAL_ROLES_KEY).unwrap().as_str().unwrap())
263 .unwrap();
264 assert_eq!(roles, vec!["admin"]);
265
266 let scopes: Vec<String> =
267 serde_json::from_str(ex.property(PRINCIPAL_SCOPES_KEY).unwrap().as_str().unwrap())
268 .unwrap();
269 assert_eq!(scopes, vec!["read"]);
270
271 let audience: Vec<String> = serde_json::from_str(
272 ex.property(PRINCIPAL_AUDIENCE_KEY)
273 .unwrap()
274 .as_str()
275 .unwrap(),
276 )
277 .unwrap();
278 assert_eq!(audience, vec!["api"]);
279
280 let claims: serde_json::Value =
281 serde_json::from_str(ex.property(PRINCIPAL_CLAIMS_KEY).unwrap().as_str().unwrap())
282 .unwrap();
283 assert_eq!(claims["sub"], "user1");
284 }
285
286 #[tokio::test]
287 async fn test_granted_empty_principal_fields() {
288 struct EmptyPrincipalPolicy;
289 #[async_trait]
290 impl SecurityPolicy for EmptyPrincipalPolicy {
291 async fn evaluate(
292 &self,
293 _exchange: &mut Exchange,
294 ) -> Result<AuthorizationDecision, CamelError> {
295 Ok(AuthorizationDecision::Granted {
296 principal: Principal {
297 subject: "minimal".into(),
298 issuer: String::new(),
299 audience: vec![],
300 scopes: vec![],
301 roles: vec![],
302 claims: serde_json::Value::Null,
303 },
304 })
305 }
306 }
307 let layer = SecurityPolicyLayer::new(Arc::new(EmptyPrincipalPolicy));
308 let mut svc = layer.layer(ok_processor());
309 let result = svc.ready().await.unwrap().call(make_exchange()).await;
310 let ex = result.unwrap();
311
312 assert_eq!(
313 ex.property(PRINCIPAL_SUBJECT_KEY),
314 Some(&serde_json::Value::String("minimal".into()))
315 );
316 assert_eq!(
317 ex.property(PRINCIPAL_ISSUER_KEY),
318 Some(&serde_json::Value::String(String::new()))
319 );
320 let roles: Vec<String> =
321 serde_json::from_str(ex.property(PRINCIPAL_ROLES_KEY).unwrap().as_str().unwrap())
322 .unwrap();
323 assert!(roles.is_empty());
324 }
325
326 #[tokio::test]
327 async fn test_layer_clone_produces_working_service() {
328 let layer = SecurityPolicyLayer::new(Arc::new(GrantPolicy));
329 let mut svc1 = layer.layer(ok_processor());
330 let svc2 = svc1.clone();
331
332 let r1 = svc1.ready().await.unwrap().call(make_exchange()).await;
333 let mut svc2 = svc2;
334 let r2 = svc2.ready().await.unwrap().call(make_exchange()).await;
335 assert!(r1.is_ok());
336 assert!(r2.is_ok());
337 }
338
339 #[tokio::test]
340 async fn test_granted_preserves_original_exchange_properties() {
341 struct GrantPolicy;
342 #[async_trait]
343 impl SecurityPolicy for GrantPolicy {
344 async fn evaluate(
345 &self,
346 _exchange: &mut Exchange,
347 ) -> Result<AuthorizationDecision, CamelError> {
348 Ok(AuthorizationDecision::Granted {
349 principal: Principal {
350 subject: "u".into(),
351 issuer: "i".into(),
352 audience: vec![],
353 scopes: vec![],
354 roles: vec![],
355 claims: serde_json::Value::Null,
356 },
357 })
358 }
359 }
360 let layer = SecurityPolicyLayer::new(Arc::new(GrantPolicy));
361 let mut svc = layer.layer(ok_processor());
362 let mut ex = make_exchange();
363 ex.set_property("custom.key", "custom-value");
364 let result = svc.ready().await.unwrap().call(ex).await;
365 let ex = result.unwrap();
366 assert_eq!(
367 ex.property("custom.key"),
368 Some(&serde_json::Value::String("custom-value".into()))
369 );
370 assert!(ex.property(PRINCIPAL_SUBJECT_KEY).is_some());
371 }
372}