mecha10_core/
auth.rs

1//! Authorization and Access Control
2//!
3//! Provides role-based access control (RBAC) for pub/sub topics.
4//! Enables controlling which nodes can publish/subscribe to specific topics.
5//!
6//! # Architecture
7//!
8//! - Simple role-based permissions
9//! - Wildcard pattern matching for topics
10//! - Policies stored in Redis
11//! - Optional enforcement (can be disabled for development)
12//!
13//! # Example
14//!
15//! ```rust
16//! use mecha10_core::prelude::*;
17//! use mecha10_core::auth::{AuthExt, Policy, Permission};
18//!
19//! # async fn example(ctx: &Context) -> Result<()> {
20//! // Define a policy
21//! let policy = Policy::new("camera_driver")
22//!     .allow(Permission::Publish, "/sensors/camera/*")
23//!     .allow(Permission::Subscribe, "/control/camera/*");
24//!
25//! ctx.register_policy(&policy).await?;
26//!
27//! // Assign role to node
28//! ctx.assign_role("camera-1", "camera_driver").await?;
29//!
30//! // Check permission (automatically done in pub/sub)
31//! let allowed = ctx.check_permission(
32//!     "camera-1",
33//!     Permission::Publish,
34//!     "/sensors/camera/rgb"
35//! ).await?;
36//!
37//! # Ok(())
38//! # }
39//! ```
40
41use crate::context::Context;
42use crate::enum_str_method;
43use crate::error::{Mecha10Error, Result};
44use serde::{Deserialize, Serialize};
45
46/// Permission type
47#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
48#[serde(rename_all = "lowercase")]
49pub enum Permission {
50    /// Permission to publish to a topic
51    Publish,
52    /// Permission to subscribe to a topic
53    Subscribe,
54}
55
56enum_str_method!(
57    #[allow(dead_code)]
58    fn as_str(Permission) -> &str {
59        Publish => "publish",
60        Subscribe => "subscribe",
61    }
62);
63
64/// Access control rule
65#[derive(Debug, Clone, Serialize, Deserialize)]
66pub struct Rule {
67    /// Permission type
68    pub permission: Permission,
69
70    /// Topic pattern (supports wildcards: /sensors/*)
71    pub topic_pattern: String,
72}
73
74impl Rule {
75    /// Create a new rule
76    pub fn new(permission: Permission, topic_pattern: impl Into<String>) -> Self {
77        Self {
78            permission,
79            topic_pattern: topic_pattern.into(),
80        }
81    }
82
83    /// Check if a topic matches this rule's pattern
84    pub fn matches(&self, topic: &str) -> bool {
85        matches_pattern(&self.topic_pattern, topic)
86    }
87}
88
89/// Policy defining permissions for a role
90#[derive(Debug, Clone, Serialize, Deserialize)]
91pub struct Policy {
92    /// Role name
93    pub role: String,
94
95    /// Access control rules
96    pub rules: Vec<Rule>,
97
98    /// Policy description
99    #[serde(default)]
100    pub description: String,
101}
102
103impl Policy {
104    /// Create a new policy for a role
105    pub fn new(role: impl Into<String>) -> Self {
106        Self {
107            role: role.into(),
108            rules: Vec::new(),
109            description: String::new(),
110        }
111    }
112
113    /// Set description
114    pub fn with_description(mut self, description: impl Into<String>) -> Self {
115        self.description = description.into();
116        self
117    }
118
119    /// Add a permission rule
120    pub fn allow(mut self, permission: Permission, topic_pattern: impl Into<String>) -> Self {
121        self.rules.push(Rule::new(permission, topic_pattern));
122        self
123    }
124
125    /// Add publish permission
126    pub fn allow_publish(self, topic_pattern: impl Into<String>) -> Self {
127        self.allow(Permission::Publish, topic_pattern)
128    }
129
130    /// Add subscribe permission
131    pub fn allow_subscribe(self, topic_pattern: impl Into<String>) -> Self {
132        self.allow(Permission::Subscribe, topic_pattern)
133    }
134
135    /// Check if this policy allows a specific action
136    pub fn allows(&self, permission: Permission, topic: &str) -> bool {
137        self.rules
138            .iter()
139            .any(|rule| rule.permission == permission && rule.matches(topic))
140    }
141
142    /// Get Redis key for this policy
143    fn redis_key(&self) -> String {
144        format!("mecha10:auth:policies:{}", self.role)
145    }
146
147    /// Get Redis key for policies list
148    fn redis_list_key() -> String {
149        "mecha10:auth:policies:list".to_string()
150    }
151
152    /// Get Redis key for node role assignment
153    fn redis_role_key(node_id: &str) -> String {
154        format!("mecha10:auth:roles:{}", node_id)
155    }
156}
157
158/// Authorization extension trait for Context
159pub trait AuthExt {
160    /// Register a policy
161    ///
162    /// # Arguments
163    ///
164    /// * `policy` - Policy to register
165    ///
166    /// # Example
167    ///
168    /// ```rust
169    /// use mecha10_core::prelude::*;
170    /// use mecha10_core::auth::{AuthExt, Policy, Permission};
171    ///
172    /// # async fn example(ctx: &Context) -> Result<()> {
173    /// let policy = Policy::new("camera_driver")
174    ///     .allow_publish("/sensors/camera/*")
175    ///     .allow_subscribe("/control/camera/*");
176    ///
177    /// ctx.register_policy(&policy).await?;
178    /// # Ok(())
179    /// # }
180    /// ```
181    fn register_policy(&self, policy: &Policy) -> impl std::future::Future<Output = Result<()>> + Send;
182
183    /// Get a policy by role name
184    ///
185    /// # Arguments
186    ///
187    /// * `role` - Role name
188    fn get_policy(&self, role: &str) -> impl std::future::Future<Output = Result<Option<Policy>>> + Send;
189
190    /// List all registered policies
191    fn list_policies(&self) -> impl std::future::Future<Output = Result<Vec<Policy>>> + Send;
192
193    /// Assign a role to a node
194    ///
195    /// # Arguments
196    ///
197    /// * `node_id` - Node identifier
198    /// * `role` - Role to assign
199    fn assign_role(&self, node_id: &str, role: &str) -> impl std::future::Future<Output = Result<()>> + Send;
200
201    /// Get the role assigned to a node
202    ///
203    /// # Arguments
204    ///
205    /// * `node_id` - Node identifier
206    fn get_role(&self, node_id: &str) -> impl std::future::Future<Output = Result<Option<String>>> + Send;
207
208    /// Check if a node has permission to perform an action on a topic
209    ///
210    /// # Arguments
211    ///
212    /// * `node_id` - Node identifier
213    /// * `permission` - Permission type (Publish/Subscribe)
214    /// * `topic` - Topic path
215    ///
216    /// # Returns
217    ///
218    /// `true` if permission is granted, `false` otherwise
219    fn check_permission(
220        &self,
221        node_id: &str,
222        permission: Permission,
223        topic: &str,
224    ) -> impl std::future::Future<Output = Result<bool>> + Send;
225
226    /// Delete a policy
227    ///
228    /// # Arguments
229    ///
230    /// * `role` - Role name
231    fn delete_policy(&self, role: &str) -> impl std::future::Future<Output = Result<()>> + Send;
232
233    /// Remove role assignment from a node
234    ///
235    /// # Arguments
236    ///
237    /// * `node_id` - Node identifier
238    fn unassign_role(&self, node_id: &str) -> impl std::future::Future<Output = Result<()>> + Send;
239}
240
241impl AuthExt for Context {
242    async fn register_policy(&self, policy: &Policy) -> Result<()> {
243        #[cfg(feature = "messaging")]
244        {
245            use redis::AsyncCommands;
246
247            let redis_url = Context::get_redis_url()?;
248            let client = redis::Client::open(redis_url.as_str()).map_err(|e| Mecha10Error::MessagingError {
249                message: format!("Failed to connect to Redis: {}", e),
250                suggestion: "Ensure Redis is running".to_string(),
251            })?;
252
253            let mut conn =
254                client
255                    .get_multiplexed_async_connection()
256                    .await
257                    .map_err(|e| Mecha10Error::MessagingError {
258                        message: format!("Failed to get Redis connection: {}", e),
259                        suggestion: "Ensure Redis is running".to_string(),
260                    })?;
261
262            // Serialize policy
263            let policy_json = serde_json::to_string(&policy)
264                .map_err(|e| Mecha10Error::Other(format!("Failed to serialize policy: {}", e)))?;
265
266            // Store policy
267            let key = policy.redis_key();
268            conn.set::<_, _, ()>(&key, &policy_json)
269                .await
270                .map_err(|e| Mecha10Error::MessagingError {
271                    message: format!("Failed to register policy: {}", e),
272                    suggestion: "Check Redis connection".to_string(),
273                })?;
274
275            // Add to policies list
276            let list_key = Policy::redis_list_key();
277            conn.sadd::<_, _, ()>(&list_key, &policy.role)
278                .await
279                .map_err(|e| Mecha10Error::MessagingError {
280                    message: format!("Failed to update policy list: {}", e),
281                    suggestion: "Check Redis connection".to_string(),
282                })?;
283
284            Ok(())
285        }
286
287        #[cfg(not(feature = "messaging"))]
288        {
289            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
290        }
291    }
292
293    async fn get_policy(&self, role: &str) -> Result<Option<Policy>> {
294        #[cfg(feature = "messaging")]
295        {
296            use redis::AsyncCommands;
297
298            let redis_url = Context::get_redis_url()?;
299            let client = redis::Client::open(redis_url.as_str()).map_err(|e| Mecha10Error::MessagingError {
300                message: format!("Failed to connect to Redis: {}", e),
301                suggestion: "Ensure Redis is running".to_string(),
302            })?;
303
304            let mut conn =
305                client
306                    .get_multiplexed_async_connection()
307                    .await
308                    .map_err(|e| Mecha10Error::MessagingError {
309                        message: format!("Failed to get Redis connection: {}", e),
310                        suggestion: "Ensure Redis is running".to_string(),
311                    })?;
312
313            let key = format!("mecha10:auth:policies:{}", role);
314            let json: Option<String> = conn.get(&key).await.map_err(|e| Mecha10Error::MessagingError {
315                message: format!("Failed to get policy: {}", e),
316                suggestion: "Check Redis connection".to_string(),
317            })?;
318
319            if let Some(json) = json {
320                let policy = serde_json::from_str::<Policy>(&json)
321                    .map_err(|e| Mecha10Error::Other(format!("Failed to parse policy: {}", e)))?;
322                Ok(Some(policy))
323            } else {
324                Ok(None)
325            }
326        }
327
328        #[cfg(not(feature = "messaging"))]
329        {
330            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
331        }
332    }
333
334    async fn list_policies(&self) -> Result<Vec<Policy>> {
335        #[cfg(feature = "messaging")]
336        {
337            crate::redis_list_util::collect_from_redis_set(&Policy::redis_list_key(), |role| async move {
338                self.get_policy(&role).await
339            })
340            .await
341        }
342
343        #[cfg(not(feature = "messaging"))]
344        {
345            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
346        }
347    }
348
349    async fn assign_role(&self, node_id: &str, role: &str) -> Result<()> {
350        #[cfg(feature = "messaging")]
351        {
352            use redis::AsyncCommands;
353
354            let redis_url = Context::get_redis_url()?;
355            let client = redis::Client::open(redis_url.as_str()).map_err(|e| Mecha10Error::MessagingError {
356                message: format!("Failed to connect to Redis: {}", e),
357                suggestion: "Ensure Redis is running".to_string(),
358            })?;
359
360            let mut conn =
361                client
362                    .get_multiplexed_async_connection()
363                    .await
364                    .map_err(|e| Mecha10Error::MessagingError {
365                        message: format!("Failed to get Redis connection: {}", e),
366                        suggestion: "Ensure Redis is running".to_string(),
367                    })?;
368
369            let key = Policy::redis_role_key(node_id);
370            conn.set::<_, _, ()>(&key, role)
371                .await
372                .map_err(|e| Mecha10Error::MessagingError {
373                    message: format!("Failed to assign role: {}", e),
374                    suggestion: "Check Redis connection".to_string(),
375                })?;
376
377            Ok(())
378        }
379
380        #[cfg(not(feature = "messaging"))]
381        {
382            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
383        }
384    }
385
386    async fn get_role(&self, node_id: &str) -> Result<Option<String>> {
387        #[cfg(feature = "messaging")]
388        {
389            use redis::AsyncCommands;
390
391            let redis_url = Context::get_redis_url()?;
392            let client = redis::Client::open(redis_url.as_str()).map_err(|e| Mecha10Error::MessagingError {
393                message: format!("Failed to connect to Redis: {}", e),
394                suggestion: "Ensure Redis is running".to_string(),
395            })?;
396
397            let mut conn =
398                client
399                    .get_multiplexed_async_connection()
400                    .await
401                    .map_err(|e| Mecha10Error::MessagingError {
402                        message: format!("Failed to get Redis connection: {}", e),
403                        suggestion: "Ensure Redis is running".to_string(),
404                    })?;
405
406            let key = Policy::redis_role_key(node_id);
407            let role: Option<String> = conn.get(&key).await.map_err(|e| Mecha10Error::MessagingError {
408                message: format!("Failed to get role: {}", e),
409                suggestion: "Check Redis connection".to_string(),
410            })?;
411
412            Ok(role)
413        }
414
415        #[cfg(not(feature = "messaging"))]
416        {
417            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
418        }
419    }
420
421    async fn check_permission(&self, node_id: &str, permission: Permission, topic: &str) -> Result<bool> {
422        // Get node's role
423        let role = match self.get_role(node_id).await? {
424            Some(role) => role,
425            None => return Ok(false), // No role = no permissions
426        };
427
428        // Get policy for role
429        let policy = match self.get_policy(&role).await? {
430            Some(policy) => policy,
431            None => return Ok(false), // No policy = no permissions
432        };
433
434        // Check if policy allows this action
435        Ok(policy.allows(permission, topic))
436    }
437
438    async fn delete_policy(&self, role: &str) -> Result<()> {
439        #[cfg(feature = "messaging")]
440        {
441            use redis::AsyncCommands;
442
443            let redis_url = Context::get_redis_url()?;
444            let client = redis::Client::open(redis_url.as_str()).map_err(|e| Mecha10Error::MessagingError {
445                message: format!("Failed to connect to Redis: {}", e),
446                suggestion: "Ensure Redis is running".to_string(),
447            })?;
448
449            let mut conn =
450                client
451                    .get_multiplexed_async_connection()
452                    .await
453                    .map_err(|e| Mecha10Error::MessagingError {
454                        message: format!("Failed to get Redis connection: {}", e),
455                        suggestion: "Ensure Redis is running".to_string(),
456                    })?;
457
458            // Delete policy
459            let key = format!("mecha10:auth:policies:{}", role);
460            conn.del::<_, ()>(&key)
461                .await
462                .map_err(|e| Mecha10Error::MessagingError {
463                    message: format!("Failed to delete policy: {}", e),
464                    suggestion: "Check Redis connection".to_string(),
465                })?;
466
467            // Remove from list
468            let list_key = Policy::redis_list_key();
469            conn.srem::<_, _, ()>(&list_key, role)
470                .await
471                .map_err(|e| Mecha10Error::MessagingError {
472                    message: format!("Failed to update policy list: {}", e),
473                    suggestion: "Check Redis connection".to_string(),
474                })?;
475
476            Ok(())
477        }
478
479        #[cfg(not(feature = "messaging"))]
480        {
481            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
482        }
483    }
484
485    async fn unassign_role(&self, node_id: &str) -> Result<()> {
486        #[cfg(feature = "messaging")]
487        {
488            use redis::AsyncCommands;
489
490            let redis_url = Context::get_redis_url()?;
491            let client = redis::Client::open(redis_url.as_str()).map_err(|e| Mecha10Error::MessagingError {
492                message: format!("Failed to connect to Redis: {}", e),
493                suggestion: "Ensure Redis is running".to_string(),
494            })?;
495
496            let mut conn =
497                client
498                    .get_multiplexed_async_connection()
499                    .await
500                    .map_err(|e| Mecha10Error::MessagingError {
501                        message: format!("Failed to get Redis connection: {}", e),
502                        suggestion: "Ensure Redis is running".to_string(),
503                    })?;
504
505            let key = Policy::redis_role_key(node_id);
506            conn.del::<_, ()>(&key)
507                .await
508                .map_err(|e| Mecha10Error::MessagingError {
509                    message: format!("Failed to unassign role: {}", e),
510                    suggestion: "Check Redis connection".to_string(),
511                })?;
512
513            Ok(())
514        }
515
516        #[cfg(not(feature = "messaging"))]
517        {
518            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
519        }
520    }
521}
522
523/// Check if a topic matches a pattern
524///
525/// Uses the same glob matching implementation as aggregated topics,
526/// supporting full glob syntax including *, **, ?, [abc], {a,b}, etc.
527///
528/// # Examples
529///
530/// ```
531/// # use mecha10_core::auth::matches_pattern;
532/// assert!(matches_pattern("/sensors/*", "/sensors/camera"));
533/// assert!(matches_pattern("/sensors/*/rgb", "/sensors/camera/rgb"));
534/// assert!(!matches_pattern("/sensors/*", "/control/motor"));
535///
536/// // Advanced patterns
537/// assert!(matches_pattern("/sensors/**", "/sensors/camera/left/rgb"));
538/// assert!(matches_pattern("/sensors/camera?", "/sensors/camera1"));
539/// assert!(matches_pattern("/sensors/{left,right}", "/sensors/left"));
540/// ```
541pub fn matches_pattern(pattern: &str, topic: &str) -> bool {
542    // Use the centralized glob matching from aggregated module
543    crate::aggregated::pattern::matches_pattern(topic, pattern)
544}
mecha10_core/auth.rs

mecha10_core/
auth.rs
