allsource_core/
auth.rs

1use crate::error::{AllSourceError, Result};
2use argon2::{
3    password_hash::{rand_core::OsRng, PasswordHash, PasswordHasher, PasswordVerifier, SaltString},
4    Argon2,
5};
6use chrono::{Duration, Utc};
7use dashmap::DashMap;
8use jsonwebtoken::{decode, encode, Algorithm, DecodingKey, EncodingKey, Header, Validation};
9use serde::{Deserialize, Serialize};
10use std::sync::Arc;
11use uuid::Uuid;
12
13/// User role for RBAC
14#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
15#[serde(rename_all = "lowercase")]
16pub enum Role {
17    Admin,          // Full system access
18    Developer,      // Read/write events, manage schemas
19    ReadOnly,       // Read-only access to events
20    ServiceAccount, // Programmatic access for services
21}
22
23impl Role {
24    /// Check if role has specific permission
25    pub fn has_permission(&self, permission: Permission) -> bool {
26        match self {
27            Role::Admin => true, // Admin has all permissions
28            Role::Developer => matches!(
29                permission,
30                Permission::Read
31                    | Permission::Write
32                    | Permission::Metrics
33                    | Permission::ManageSchemas
34                    | Permission::ManagePipelines
35            ),
36            Role::ReadOnly => matches!(permission, Permission::Read | Permission::Metrics),
37            Role::ServiceAccount => {
38                matches!(permission, Permission::Read | Permission::Write)
39            }
40        }
41    }
42}
43
44/// Permission types
45#[derive(Debug, Clone, Copy, PartialEq, Eq)]
46pub enum Permission {
47    Read,
48    Write,
49    Admin,
50    Metrics,
51    ManageSchemas,
52    ManagePipelines,
53    ManageTenants,
54}
55
56/// JWT Claims
57#[derive(Debug, Clone, Serialize, Deserialize)]
58pub struct Claims {
59    /// Subject (user ID or API key ID)
60    pub sub: String,
61    /// Tenant ID
62    pub tenant_id: String,
63    /// User role
64    pub role: Role,
65    /// Expiration time (UNIX timestamp)
66    pub exp: i64,
67    /// Issued at time (UNIX timestamp)
68    pub iat: i64,
69    /// Issuer
70    pub iss: String,
71}
72
73impl Claims {
74    /// Create new claims for a user
75    pub fn new(user_id: String, tenant_id: String, role: Role, expires_in: Duration) -> Self {
76        let now = Utc::now();
77        Self {
78            sub: user_id,
79            tenant_id,
80            role,
81            iat: now.timestamp(),
82            exp: (now + expires_in).timestamp(),
83            iss: "allsource".to_string(),
84        }
85    }
86
87    /// Check if claims are expired
88    pub fn is_expired(&self) -> bool {
89        Utc::now().timestamp() > self.exp
90    }
91
92    /// Check if user has permission
93    pub fn has_permission(&self, permission: Permission) -> bool {
94        self.role.has_permission(permission)
95    }
96}
97
98/// User account
99#[derive(Debug, Clone, Serialize, Deserialize)]
100pub struct User {
101    pub id: Uuid,
102    pub username: String,
103    pub email: String,
104    #[serde(skip_serializing)]
105    pub password_hash: String,
106    pub role: Role,
107    pub tenant_id: String,
108    pub created_at: chrono::DateTime<Utc>,
109    pub active: bool,
110}
111
112impl User {
113    /// Create a new user with hashed password
114    pub fn new(
115        username: String,
116        email: String,
117        password: &str,
118        role: Role,
119        tenant_id: String,
120    ) -> Result<Self> {
121        let password_hash = hash_password(password)?;
122
123        Ok(Self {
124            id: Uuid::new_v4(),
125            username,
126            email,
127            password_hash,
128            role,
129            tenant_id,
130            created_at: Utc::now(),
131            active: true,
132        })
133    }
134
135    /// Verify password
136    pub fn verify_password(&self, password: &str) -> Result<bool> {
137        verify_password(password, &self.password_hash)
138    }
139}
140
141/// API Key for programmatic access
142#[derive(Debug, Clone, Serialize, Deserialize)]
143pub struct ApiKey {
144    pub id: Uuid,
145    pub name: String,
146    pub tenant_id: String,
147    pub role: Role,
148    #[serde(skip_serializing)]
149    pub key_hash: String,
150    pub created_at: chrono::DateTime<Utc>,
151    pub expires_at: Option<chrono::DateTime<Utc>>,
152    pub active: bool,
153    pub last_used: Option<chrono::DateTime<Utc>>,
154}
155
156impl ApiKey {
157    /// Create new API key
158    pub fn new(
159        name: String,
160        tenant_id: String,
161        role: Role,
162        expires_at: Option<chrono::DateTime<Utc>>,
163    ) -> (Self, String) {
164        let key = generate_api_key();
165        let key_hash = hash_api_key(&key);
166
167        let api_key = Self {
168            id: Uuid::new_v4(),
169            name,
170            tenant_id,
171            role,
172            key_hash,
173            created_at: Utc::now(),
174            expires_at,
175            active: true,
176            last_used: None,
177        };
178
179        (api_key, key)
180    }
181
182    /// Verify API key
183    pub fn verify(&self, key: &str) -> bool {
184        if !self.active {
185            return false;
186        }
187
188        if let Some(expires_at) = self.expires_at {
189            if Utc::now() > expires_at {
190                return false;
191            }
192        }
193
194        hash_api_key(key) == self.key_hash
195    }
196
197    /// Check if expired
198    pub fn is_expired(&self) -> bool {
199        if let Some(expires_at) = self.expires_at {
200            Utc::now() > expires_at
201        } else {
202            false
203        }
204    }
205}
206
207/// Authentication manager
208pub struct AuthManager {
209    /// JWT encoding key
210    encoding_key: EncodingKey,
211    /// JWT decoding key
212    decoding_key: DecodingKey,
213    /// JWT validation rules
214    validation: Validation,
215    /// Users storage (in production, use database)
216    users: Arc<DashMap<Uuid, User>>,
217    /// API keys storage (in production, use database)
218    api_keys: Arc<DashMap<Uuid, ApiKey>>,
219    /// Username to user ID mapping
220    username_index: Arc<DashMap<String, Uuid>>,
221}
222
223impl AuthManager {
224    /// Create new authentication manager
225    pub fn new(jwt_secret: &str) -> Self {
226        let encoding_key = EncodingKey::from_secret(jwt_secret.as_bytes());
227        let decoding_key = DecodingKey::from_secret(jwt_secret.as_bytes());
228
229        let mut validation = Validation::new(Algorithm::HS256);
230        validation.set_issuer(&["allsource"]);
231
232        Self {
233            encoding_key,
234            decoding_key,
235            validation,
236            users: Arc::new(DashMap::new()),
237            api_keys: Arc::new(DashMap::new()),
238            username_index: Arc::new(DashMap::new()),
239        }
240    }
241
242    /// Register a new user
243    pub fn register_user(
244        &self,
245        username: String,
246        email: String,
247        password: &str,
248        role: Role,
249        tenant_id: String,
250    ) -> Result<User> {
251        // Check if username already exists
252        if self.username_index.contains_key(&username) {
253            return Err(AllSourceError::ValidationError(
254                "Username already exists".to_string(),
255            ));
256        }
257
258        let user = User::new(username.clone(), email, password, role, tenant_id)?;
259
260        self.users.insert(user.id, user.clone());
261        self.username_index.insert(username, user.id);
262
263        Ok(user)
264    }
265
266    /// Authenticate user with username and password
267    pub fn authenticate(&self, username: &str, password: &str) -> Result<String> {
268        let user_id = self
269            .username_index
270            .get(username)
271            .ok_or_else(|| AllSourceError::ValidationError("Invalid credentials".to_string()))?;
272
273        let user = self
274            .users
275            .get(&user_id)
276            .ok_or_else(|| AllSourceError::ValidationError("User not found".to_string()))?;
277
278        if !user.active {
279            return Err(AllSourceError::ValidationError(
280                "User account is inactive".to_string(),
281            ));
282        }
283
284        if !user.verify_password(password)? {
285            return Err(AllSourceError::ValidationError(
286                "Invalid credentials".to_string(),
287            ));
288        }
289
290        // Generate JWT token
291        let claims = Claims::new(
292            user.id.to_string(),
293            user.tenant_id.clone(),
294            user.role.clone(),
295            Duration::hours(24), // Token expires in 24 hours
296        );
297
298        let token = encode(&Header::default(), &claims, &self.encoding_key).map_err(|e| {
299            AllSourceError::ValidationError(format!("Failed to create token: {}", e))
300        })?;
301
302        Ok(token)
303    }
304
305    /// Validate JWT token
306    pub fn validate_token(&self, token: &str) -> Result<Claims> {
307        let token_data = decode::<Claims>(token, &self.decoding_key, &self.validation)
308            .map_err(|e| AllSourceError::ValidationError(format!("Invalid token: {}", e)))?;
309
310        if token_data.claims.is_expired() {
311            return Err(AllSourceError::ValidationError("Token expired".to_string()));
312        }
313
314        Ok(token_data.claims)
315    }
316
317    /// Create API key
318    pub fn create_api_key(
319        &self,
320        name: String,
321        tenant_id: String,
322        role: Role,
323        expires_at: Option<chrono::DateTime<Utc>>,
324    ) -> (ApiKey, String) {
325        let (api_key, key) = ApiKey::new(name, tenant_id, role, expires_at);
326        self.api_keys.insert(api_key.id, api_key.clone());
327        (api_key, key)
328    }
329
330    /// Validate API key
331    pub fn validate_api_key(&self, key: &str) -> Result<Claims> {
332        for entry in self.api_keys.iter() {
333            let api_key = entry.value();
334            if api_key.verify(key) {
335                // Update last used timestamp
336                if let Some(mut key_mut) = self.api_keys.get_mut(&api_key.id) {
337                    key_mut.last_used = Some(Utc::now());
338                }
339
340                let claims = Claims::new(
341                    api_key.id.to_string(),
342                    api_key.tenant_id.clone(),
343                    api_key.role.clone(),
344                    Duration::hours(24),
345                );
346
347                return Ok(claims);
348            }
349        }
350
351        Err(AllSourceError::ValidationError(
352            "Invalid API key".to_string(),
353        ))
354    }
355
356    /// Get user by ID
357    pub fn get_user(&self, user_id: &Uuid) -> Option<User> {
358        self.users.get(user_id).map(|u| u.clone())
359    }
360
361    /// List all users (admin only)
362    pub fn list_users(&self) -> Vec<User> {
363        self.users
364            .iter()
365            .map(|entry| entry.value().clone())
366            .collect()
367    }
368
369    /// Delete user
370    pub fn delete_user(&self, user_id: &Uuid) -> Result<()> {
371        if let Some((_, user)) = self.users.remove(user_id) {
372            self.username_index.remove(&user.username);
373            Ok(())
374        } else {
375            Err(AllSourceError::ValidationError(
376                "User not found".to_string(),
377            ))
378        }
379    }
380
381    /// Revoke API key
382    pub fn revoke_api_key(&self, key_id: &Uuid) -> Result<()> {
383        if let Some(mut api_key) = self.api_keys.get_mut(key_id) {
384            api_key.active = false;
385            Ok(())
386        } else {
387            Err(AllSourceError::ValidationError(
388                "API key not found".to_string(),
389            ))
390        }
391    }
392
393    /// List API keys for a tenant
394    pub fn list_api_keys(&self, tenant_id: &str) -> Vec<ApiKey> {
395        self.api_keys
396            .iter()
397            .filter(|entry| entry.value().tenant_id == tenant_id)
398            .map(|entry| entry.value().clone())
399            .collect()
400    }
401}
402
403impl Default for AuthManager {
404    fn default() -> Self {
405        // Generate a random secret for development
406        // In production, this should come from configuration
407        use base64::{engine::general_purpose, Engine as _};
408        let secret = general_purpose::STANDARD.encode(rand::random::<[u8; 32]>());
409        Self::new(&secret)
410    }
411}
412
413/// Hash password using Argon2
414fn hash_password(password: &str) -> Result<String> {
415    let salt = SaltString::generate(&mut OsRng);
416    let argon2 = Argon2::default();
417
418    let hash = argon2
419        .hash_password(password.as_bytes(), &salt)
420        .map_err(|e| AllSourceError::ValidationError(format!("Password hashing failed: {}", e)))?;
421
422    Ok(hash.to_string())
423}
424
425/// Verify password against hash
426fn verify_password(password: &str, hash: &str) -> Result<bool> {
427    let parsed_hash = PasswordHash::new(hash)
428        .map_err(|e| AllSourceError::ValidationError(format!("Invalid password hash: {}", e)))?;
429
430    let argon2 = Argon2::default();
431    Ok(argon2
432        .verify_password(password.as_bytes(), &parsed_hash)
433        .is_ok())
434}
435
436/// Generate API key
437fn generate_api_key() -> String {
438    use base64::{engine::general_purpose, Engine as _};
439    let random_bytes: [u8; 32] = rand::random();
440    format!(
441        "ask_{}",
442        general_purpose::URL_SAFE_NO_PAD.encode(random_bytes)
443    )
444}
445
446/// Hash API key for storage
447fn hash_api_key(key: &str) -> String {
448    use std::collections::hash_map::DefaultHasher;
449    use std::hash::{Hash, Hasher};
450
451    let mut hasher = DefaultHasher::new();
452    key.hash(&mut hasher);
453    format!("{:x}", hasher.finish())
454}
455
456#[cfg(test)]
457mod tests {
458    use super::*;
459
460    #[test]
461    fn test_user_creation_and_verification() {
462        let user = User::new(
463            "testuser".to_string(),
464            "test@example.com".to_string(),
465            "password123",
466            Role::Developer,
467            "tenant1".to_string(),
468        )
469        .unwrap();
470
471        assert!(user.verify_password("password123").unwrap());
472        assert!(!user.verify_password("wrongpassword").unwrap());
473    }
474
475    #[test]
476    fn test_role_permissions() {
477        assert!(Role::Admin.has_permission(Permission::Admin));
478        assert!(Role::Developer.has_permission(Permission::Write));
479        assert!(!Role::ReadOnly.has_permission(Permission::Write));
480        assert!(Role::ReadOnly.has_permission(Permission::Read));
481    }
482
483    #[test]
484    fn test_auth_manager() {
485        let auth = AuthManager::new("test_secret");
486
487        // Register user
488        let user = auth
489            .register_user(
490                "alice".to_string(),
491                "alice@example.com".to_string(),
492                "password123",
493                Role::Developer,
494                "tenant1".to_string(),
495            )
496            .unwrap();
497
498        // Authenticate
499        let token = auth.authenticate("alice", "password123").unwrap();
500        assert!(!token.is_empty());
501
502        // Validate token
503        let claims = auth.validate_token(&token).unwrap();
504        assert_eq!(claims.tenant_id, "tenant1");
505        assert_eq!(claims.role, Role::Developer);
506    }
507
508    #[test]
509    fn test_api_key() {
510        let auth = AuthManager::new("test_secret");
511
512        let (api_key, key) = auth.create_api_key(
513            "test-key".to_string(),
514            "tenant1".to_string(),
515            Role::ServiceAccount,
516            None,
517        );
518
519        // Validate key
520        let claims = auth.validate_api_key(&key).unwrap();
521        assert_eq!(claims.tenant_id, "tenant1");
522        assert_eq!(claims.role, Role::ServiceAccount);
523
524        // Revoke key
525        auth.revoke_api_key(&api_key.id).unwrap();
526
527        // Should fail after revocation
528        assert!(auth.validate_api_key(&key).is_err());
529    }
530
531    #[test]
532    fn test_claims_expiration() {
533        let claims = Claims::new(
534            "user1".to_string(),
535            "tenant1".to_string(),
536            Role::Developer,
537            Duration::seconds(-1), // Expired 1 second ago
538        );
539
540        assert!(claims.is_expired());
541    }
542}
allsource_core/auth.rs

allsource_core/
auth.rs
