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)
299            .map_err(|e| AllSourceError::ValidationError(format!("Failed to create token: {}", e)))?;
300
301        Ok(token)
302    }
303
304    /// Validate JWT token
305    pub fn validate_token(&self, token: &str) -> Result<Claims> {
306        let token_data = decode::<Claims>(token, &self.decoding_key, &self.validation)
307            .map_err(|e| AllSourceError::ValidationError(format!("Invalid token: {}", e)))?;
308
309        if token_data.claims.is_expired() {
310            return Err(AllSourceError::ValidationError("Token expired".to_string()));
311        }
312
313        Ok(token_data.claims)
314    }
315
316    /// Create API key
317    pub fn create_api_key(
318        &self,
319        name: String,
320        tenant_id: String,
321        role: Role,
322        expires_at: Option<chrono::DateTime<Utc>>,
323    ) -> (ApiKey, String) {
324        let (api_key, key) = ApiKey::new(name, tenant_id, role, expires_at);
325        self.api_keys.insert(api_key.id, api_key.clone());
326        (api_key, key)
327    }
328
329    /// Validate API key
330    pub fn validate_api_key(&self, key: &str) -> Result<Claims> {
331        for entry in self.api_keys.iter() {
332            let api_key = entry.value();
333            if api_key.verify(key) {
334                // Update last used timestamp
335                if let Some(mut key_mut) = self.api_keys.get_mut(&api_key.id) {
336                    key_mut.last_used = Some(Utc::now());
337                }
338
339                let claims = Claims::new(
340                    api_key.id.to_string(),
341                    api_key.tenant_id.clone(),
342                    api_key.role.clone(),
343                    Duration::hours(24),
344                );
345
346                return Ok(claims);
347            }
348        }
349
350        Err(AllSourceError::ValidationError(
351            "Invalid API key".to_string(),
352        ))
353    }
354
355    /// Get user by ID
356    pub fn get_user(&self, user_id: &Uuid) -> Option<User> {
357        self.users.get(user_id).map(|u| u.clone())
358    }
359
360    /// List all users (admin only)
361    pub fn list_users(&self) -> Vec<User> {
362        self.users.iter().map(|entry| entry.value().clone()).collect()
363    }
364
365    /// Delete user
366    pub fn delete_user(&self, user_id: &Uuid) -> Result<()> {
367        if let Some((_, user)) = self.users.remove(user_id) {
368            self.username_index.remove(&user.username);
369            Ok(())
370        } else {
371            Err(AllSourceError::ValidationError(
372                "User not found".to_string(),
373            ))
374        }
375    }
376
377    /// Revoke API key
378    pub fn revoke_api_key(&self, key_id: &Uuid) -> Result<()> {
379        if let Some(mut api_key) = self.api_keys.get_mut(key_id) {
380            api_key.active = false;
381            Ok(())
382        } else {
383            Err(AllSourceError::ValidationError(
384                "API key not found".to_string(),
385            ))
386        }
387    }
388
389    /// List API keys for a tenant
390    pub fn list_api_keys(&self, tenant_id: &str) -> Vec<ApiKey> {
391        self.api_keys
392            .iter()
393            .filter(|entry| entry.value().tenant_id == tenant_id)
394            .map(|entry| entry.value().clone())
395            .collect()
396    }
397}
398
399impl Default for AuthManager {
400    fn default() -> Self {
401        // Generate a random secret for development
402        // In production, this should come from configuration
403        use base64::{Engine as _, engine::general_purpose};
404        let secret = general_purpose::STANDARD.encode(rand::random::<[u8; 32]>());
405        Self::new(&secret)
406    }
407}
408
409/// Hash password using Argon2
410fn hash_password(password: &str) -> Result<String> {
411    let salt = SaltString::generate(&mut OsRng);
412    let argon2 = Argon2::default();
413
414    let hash = argon2
415        .hash_password(password.as_bytes(), &salt)
416        .map_err(|e| AllSourceError::ValidationError(format!("Password hashing failed: {}", e)))?;
417
418    Ok(hash.to_string())
419}
420
421/// Verify password against hash
422fn verify_password(password: &str, hash: &str) -> Result<bool> {
423    let parsed_hash = PasswordHash::new(hash)
424        .map_err(|e| AllSourceError::ValidationError(format!("Invalid password hash: {}", e)))?;
425
426    let argon2 = Argon2::default();
427    Ok(argon2.verify_password(password.as_bytes(), &parsed_hash).is_ok())
428}
429
430/// Generate API key
431fn generate_api_key() -> String {
432    use base64::{Engine as _, engine::general_purpose};
433    let random_bytes: [u8; 32] = rand::random();
434    format!("ask_{}", general_purpose::URL_SAFE_NO_PAD.encode(random_bytes))
435}
436
437/// Hash API key for storage
438fn hash_api_key(key: &str) -> String {
439    use std::collections::hash_map::DefaultHasher;
440    use std::hash::{Hash, Hasher};
441
442    let mut hasher = DefaultHasher::new();
443    key.hash(&mut hasher);
444    format!("{:x}", hasher.finish())
445}
446
447#[cfg(test)]
448mod tests {
449    use super::*;
450
451    #[test]
452    fn test_user_creation_and_verification() {
453        let user = User::new(
454            "testuser".to_string(),
455            "test@example.com".to_string(),
456            "password123",
457            Role::Developer,
458            "tenant1".to_string(),
459        )
460        .unwrap();
461
462        assert!(user.verify_password("password123").unwrap());
463        assert!(!user.verify_password("wrongpassword").unwrap());
464    }
465
466    #[test]
467    fn test_role_permissions() {
468        assert!(Role::Admin.has_permission(Permission::Admin));
469        assert!(Role::Developer.has_permission(Permission::Write));
470        assert!(!Role::ReadOnly.has_permission(Permission::Write));
471        assert!(Role::ReadOnly.has_permission(Permission::Read));
472    }
473
474    #[test]
475    fn test_auth_manager() {
476        let auth = AuthManager::new("test_secret");
477
478        // Register user
479        let user = auth
480            .register_user(
481                "alice".to_string(),
482                "alice@example.com".to_string(),
483                "password123",
484                Role::Developer,
485                "tenant1".to_string(),
486            )
487            .unwrap();
488
489        // Authenticate
490        let token = auth.authenticate("alice", "password123").unwrap();
491        assert!(!token.is_empty());
492
493        // Validate token
494        let claims = auth.validate_token(&token).unwrap();
495        assert_eq!(claims.tenant_id, "tenant1");
496        assert_eq!(claims.role, Role::Developer);
497    }
498
499    #[test]
500    fn test_api_key() {
501        let auth = AuthManager::new("test_secret");
502
503        let (api_key, key) = auth.create_api_key(
504            "test-key".to_string(),
505            "tenant1".to_string(),
506            Role::ServiceAccount,
507            None,
508        );
509
510        // Validate key
511        let claims = auth.validate_api_key(&key).unwrap();
512        assert_eq!(claims.tenant_id, "tenant1");
513        assert_eq!(claims.role, Role::ServiceAccount);
514
515        // Revoke key
516        auth.revoke_api_key(&api_key.id).unwrap();
517
518        // Should fail after revocation
519        assert!(auth.validate_api_key(&key).is_err());
520    }
521
522    #[test]
523    fn test_claims_expiration() {
524        let claims = Claims::new(
525            "user1".to_string(),
526            "tenant1".to_string(),
527            Role::Developer,
528            Duration::seconds(-1), // Expired 1 second ago
529        );
530
531        assert!(claims.is_expired());
532    }
533}
allsource_core/auth.rs

allsource_core/
auth.rs
