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    // duplicate: false positive — matches `HealthStatus::with_message` in
114    // messages/system.rs on AST shape only (a plain `impl Into<String>` field setter);
115    // unrelated types, nothing to share.
116    /// Set description
117    pub fn with_description(mut self, description: impl Into<String>) -> Self {
118        self.description = description.into();
119        self
120    }
121
122    // duplicate: false positive — matches `LlmNode::add_message` (packages/ai-llm/src/llm_node.rs)
123    // on AST shape only; unrelated domains, nothing to share.
124    /// Add a permission rule
125    pub fn allow(mut self, permission: Permission, topic_pattern: impl Into<String>) -> Self {
126        self.rules.push(Rule::new(permission, topic_pattern));
127        self
128    }
129
130    // duplicate: false positive — `allow_publish`/`allow_subscribe` already delegate their
131    // shared shape to `allow`; only the `Permission` variant differs.
132    /// Add publish permission
133    pub fn allow_publish(self, topic_pattern: impl Into<String>) -> Self {
134        self.allow(Permission::Publish, topic_pattern)
135    }
136
137    /// Add subscribe permission
138    pub fn allow_subscribe(self, topic_pattern: impl Into<String>) -> Self {
139        self.allow(Permission::Subscribe, topic_pattern)
140    }
141
142    /// Check if this policy allows a specific action
143    pub fn allows(&self, permission: Permission, topic: &str) -> bool {
144        self.rules
145            .iter()
146            .any(|rule| rule.permission == permission && rule.matches(topic))
147    }
148
149    /// Get Redis key for this policy
150    fn redis_key(&self) -> String {
151        format!("mecha10:auth:policies:{}", self.role)
152    }
153
154    /// Get Redis key for policies list
155    fn redis_list_key() -> String {
156        "mecha10:auth:policies:list".to_string()
157    }
158
159    /// Get Redis key for node role assignment
160    fn redis_role_key(node_id: &str) -> String {
161        format!("mecha10:auth:roles:{}", node_id)
162    }
163}
164
165/// Authorization extension trait for Context
166pub trait AuthExt {
167    /// Register a policy
168    ///
169    /// # Arguments
170    ///
171    /// * `policy` - Policy to register
172    ///
173    /// # Example
174    ///
175    /// ```rust
176    /// use mecha10_core::prelude::*;
177    /// use mecha10_core::auth::{AuthExt, Policy, Permission};
178    ///
179    /// # async fn example(ctx: &Context) -> Result<()> {
180    /// let policy = Policy::new("camera_driver")
181    ///     .allow_publish("/sensors/camera/*")
182    ///     .allow_subscribe("/control/camera/*");
183    ///
184    /// ctx.register_policy(&policy).await?;
185    /// # Ok(())
186    /// # }
187    /// ```
188    fn register_policy(&self, policy: &Policy) -> impl std::future::Future<Output = Result<()>> + Send;
189
190    /// Get a policy by role name
191    ///
192    /// # Arguments
193    ///
194    /// * `role` - Role name
195    fn get_policy(&self, role: &str) -> impl std::future::Future<Output = Result<Option<Policy>>> + Send;
196
197    /// List all registered policies
198    fn list_policies(&self) -> impl std::future::Future<Output = Result<Vec<Policy>>> + Send;
199
200    /// Assign a role to a node
201    ///
202    /// # Arguments
203    ///
204    /// * `node_id` - Node identifier
205    /// * `role` - Role to assign
206    fn assign_role(&self, node_id: &str, role: &str) -> impl std::future::Future<Output = Result<()>> + Send;
207
208    /// Get the role assigned to a node
209    ///
210    /// # Arguments
211    ///
212    /// * `node_id` - Node identifier
213    fn get_role(&self, node_id: &str) -> impl std::future::Future<Output = Result<Option<String>>> + Send;
214
215    /// Check if a node has permission to perform an action on a topic
216    ///
217    /// # Arguments
218    ///
219    /// * `node_id` - Node identifier
220    /// * `permission` - Permission type (Publish/Subscribe)
221    /// * `topic` - Topic path
222    ///
223    /// # Returns
224    ///
225    /// `true` if permission is granted, `false` otherwise
226    fn check_permission(
227        &self,
228        node_id: &str,
229        permission: Permission,
230        topic: &str,
231    ) -> impl std::future::Future<Output = Result<bool>> + Send;
232
233    /// Delete a policy
234    ///
235    /// # Arguments
236    ///
237    /// * `role` - Role name
238    fn delete_policy(&self, role: &str) -> impl std::future::Future<Output = Result<()>> + Send;
239
240    /// Remove role assignment from a node
241    ///
242    /// # Arguments
243    ///
244    /// * `node_id` - Node identifier
245    fn unassign_role(&self, node_id: &str) -> impl std::future::Future<Output = Result<()>> + Send;
246}
247
248impl AuthExt for Context {
249    async fn register_policy(&self, policy: &Policy) -> Result<()> {
250        #[cfg(feature = "messaging")]
251        {
252            use redis::AsyncCommands;
253
254            let mut conn = crate::redis_list_util::get_redis_connection().await?;
255
256            let policy_json = serde_json::to_string(&policy)
257                .map_err(|e| Mecha10Error::Other(format!("Failed to serialize policy: {}", e)))?;
258
259            let key = policy.redis_key();
260            conn.set::<_, _, ()>(&key, &policy_json)
261                .await
262                .map_err(|e| Mecha10Error::MessagingError {
263                    message: format!("Failed to register policy: {}", e),
264                    suggestion: "Check Redis connection".to_string(),
265                })?;
266
267            let list_key = Policy::redis_list_key();
268            conn.sadd::<_, _, ()>(&list_key, &policy.role)
269                .await
270                .map_err(|e| Mecha10Error::MessagingError {
271                    message: format!("Failed to update policy list: {}", e),
272                    suggestion: "Check Redis connection".to_string(),
273                })?;
274
275            Ok(())
276        }
277
278        #[cfg(not(feature = "messaging"))]
279        {
280            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
281        }
282    }
283
284    async fn get_policy(&self, role: &str) -> Result<Option<Policy>> {
285        #[cfg(feature = "messaging")]
286        {
287            use redis::AsyncCommands;
288
289            let mut conn = crate::redis_list_util::get_redis_connection().await?;
290
291            let key = format!("mecha10:auth:policies:{}", role);
292            let json: Option<String> = conn.get(&key).await.map_err(|e| Mecha10Error::MessagingError {
293                message: format!("Failed to get policy: {}", e),
294                suggestion: "Check Redis connection".to_string(),
295            })?;
296
297            if let Some(json) = json {
298                let policy = serde_json::from_str::<Policy>(&json)
299                    .map_err(|e| Mecha10Error::Other(format!("Failed to parse policy: {}", e)))?;
300                Ok(Some(policy))
301            } else {
302                Ok(None)
303            }
304        }
305
306        #[cfg(not(feature = "messaging"))]
307        {
308            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
309        }
310    }
311
312    // duplicate: false positive — matches `SchemaRegistryExt::list_schemas` in
313    // schema/registry_impl.rs on AST shape only; both already delegate the actual list-and-fetch
314    // logic to the shared `redis_list_util::collect_from_redis_set` helper, so what's left is
315    // the standard `#[cfg(feature = "messaging")]` gate repeated across this crate's trait impls.
316    async fn list_policies(&self) -> Result<Vec<Policy>> {
317        #[cfg(feature = "messaging")]
318        {
319            crate::redis_list_util::collect_from_redis_set(&Policy::redis_list_key(), |role| async move {
320                self.get_policy(&role).await
321            })
322            .await
323        }
324
325        #[cfg(not(feature = "messaging"))]
326        {
327            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
328        }
329    }
330
331    async fn assign_role(&self, node_id: &str, role: &str) -> Result<()> {
332        #[cfg(feature = "messaging")]
333        {
334            use redis::AsyncCommands;
335
336            let mut conn = crate::redis_list_util::get_redis_connection().await?;
337
338            let key = Policy::redis_role_key(node_id);
339            conn.set::<_, _, ()>(&key, role)
340                .await
341                .map_err(|e| Mecha10Error::MessagingError {
342                    message: format!("Failed to assign role: {}", e),
343                    suggestion: "Check Redis connection".to_string(),
344                })?;
345
346            Ok(())
347        }
348
349        #[cfg(not(feature = "messaging"))]
350        {
351            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
352        }
353    }
354
355    async fn get_role(&self, node_id: &str) -> Result<Option<String>> {
356        #[cfg(feature = "messaging")]
357        {
358            let key = Policy::redis_role_key(node_id);
359            crate::redis_list_util::get_optional_from_redis(&key, "Failed to get role").await
360        }
361
362        #[cfg(not(feature = "messaging"))]
363        {
364            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
365        }
366    }
367
368    async fn check_permission(&self, node_id: &str, permission: Permission, topic: &str) -> Result<bool> {
369        let role = match self.get_role(node_id).await? {
370            Some(role) => role,
371            None => return Ok(false),
372        };
373
374        let policy = match self.get_policy(&role).await? {
375            Some(policy) => policy,
376            None => return Ok(false),
377        };
378
379        Ok(policy.allows(permission, topic))
380    }
381
382    async fn delete_policy(&self, role: &str) -> Result<()> {
383        #[cfg(feature = "messaging")]
384        {
385            use redis::AsyncCommands;
386
387            let mut conn = crate::redis_list_util::get_redis_connection().await?;
388
389            let key = format!("mecha10:auth:policies:{}", role);
390            conn.del::<_, ()>(&key)
391                .await
392                .map_err(|e| Mecha10Error::MessagingError {
393                    message: format!("Failed to delete policy: {}", e),
394                    suggestion: "Check Redis connection".to_string(),
395                })?;
396
397            let list_key = Policy::redis_list_key();
398            conn.srem::<_, _, ()>(&list_key, role)
399                .await
400                .map_err(|e| Mecha10Error::MessagingError {
401                    message: format!("Failed to update policy list: {}", e),
402                    suggestion: "Check Redis connection".to_string(),
403                })?;
404
405            Ok(())
406        }
407
408        #[cfg(not(feature = "messaging"))]
409        {
410            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
411        }
412    }
413
414    async fn unassign_role(&self, node_id: &str) -> Result<()> {
415        #[cfg(feature = "messaging")]
416        {
417            use redis::AsyncCommands;
418
419            let mut conn = crate::redis_list_util::get_redis_connection().await?;
420
421            let key = Policy::redis_role_key(node_id);
422            conn.del::<_, ()>(&key)
423                .await
424                .map_err(|e| Mecha10Error::MessagingError {
425                    message: format!("Failed to unassign role: {}", e),
426                    suggestion: "Check Redis connection".to_string(),
427                })?;
428
429            Ok(())
430        }
431
432        #[cfg(not(feature = "messaging"))]
433        {
434            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
435        }
436    }
437}
438
439/// Check if a topic matches a pattern
440///
441/// Uses the same glob matching implementation as aggregated topics,
442/// supporting full glob syntax including *, **, ?, [abc], {a,b}, etc.
443///
444/// # Examples
445///
446/// ```
447/// # use mecha10_core::auth::matches_pattern;
448/// assert!(matches_pattern("/sensors/*", "/sensors/camera"));
449/// assert!(matches_pattern("/sensors/*/rgb", "/sensors/camera/rgb"));
450/// assert!(!matches_pattern("/sensors/*", "/control/motor"));
451///
452/// // Advanced patterns
453/// assert!(matches_pattern("/sensors/**", "/sensors/camera/left/rgb"));
454/// assert!(matches_pattern("/sensors/camera?", "/sensors/camera1"));
455/// assert!(matches_pattern("/sensors/{left,right}", "/sensors/left"));
456/// ```
457pub fn matches_pattern(pattern: &str, topic: &str) -> bool {
458    crate::aggregated::pattern::matches_pattern(topic, pattern)
459}
mecha10_core/auth.rs

mecha10_core/
auth.rs
