vecstore/
audit.rs

1//! Audit Logging
2//!
3//! Provides comprehensive audit trails for compliance and security monitoring.
4
5use serde::{Deserialize, Serialize};
6use std::collections::VecDeque;
7use std::fs::{File, OpenOptions};
8use std::io::Write as IoWrite;
9use std::path::PathBuf;
10use std::sync::{Arc, Mutex};
11use std::time::SystemTime;
12
13/// Audit event types
14#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
15pub enum AuditEventType {
16    /// Vector insertion
17    Insert,
18    /// Vector update
19    Update,
20    /// Vector deletion
21    Delete,
22    /// Query operation
23    Query,
24    /// Batch operation
25    Batch,
26    /// Index creation
27    IndexCreate,
28    /// Index deletion
29    IndexDelete,
30    /// Authentication
31    Auth,
32    /// Authorization
33    Authz,
34    /// Configuration change
35    ConfigChange,
36    /// Backup operation
37    Backup,
38    /// Restore operation
39    Restore,
40    /// Export operation
41    Export,
42    /// Import operation
43    Import,
44}
45
46/// Audit severity levels
47#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
48pub enum AuditSeverity {
49    Debug,
50    Info,
51    Warning,
52    Error,
53    Critical,
54}
55
56/// Audit outcome
57#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
58pub enum AuditOutcome {
59    Success,
60    Failure,
61    Denied,
62}
63
64/// Audit entry metadata
65#[derive(Debug, Clone, Serialize, Deserialize)]
66pub struct AuditMetadata {
67    /// User identifier
68    pub user_id: Option<String>,
69
70    /// IP address
71    pub ip_address: Option<String>,
72
73    /// Session ID
74    pub session_id: Option<String>,
75
76    /// Request ID for tracing
77    pub request_id: Option<String>,
78
79    /// Additional custom fields
80    #[serde(flatten)]
81    pub custom: std::collections::HashMap<String, serde_json::Value>,
82}
83
84impl Default for AuditMetadata {
85    fn default() -> Self {
86        Self {
87            user_id: None,
88            ip_address: None,
89            session_id: None,
90            request_id: None,
91            custom: std::collections::HashMap::new(),
92        }
93    }
94}
95
96/// Audit log entry
97#[derive(Debug, Clone, Serialize, Deserialize)]
98pub struct AuditEntry {
99    /// Unique entry ID
100    pub id: String,
101
102    /// Timestamp
103    pub timestamp: SystemTime,
104
105    /// Event type
106    pub event_type: AuditEventType,
107
108    /// Severity level
109    pub severity: AuditSeverity,
110
111    /// Outcome
112    pub outcome: AuditOutcome,
113
114    /// Resource affected (e.g., vector ID, index name)
115    pub resource: Option<String>,
116
117    /// Action description
118    pub action: String,
119
120    /// Additional details
121    pub details: Option<String>,
122
123    /// Metadata (user, IP, etc.)
124    pub metadata: AuditMetadata,
125
126    /// Duration in milliseconds (for operations)
127    pub duration_ms: Option<u64>,
128}
129
130impl AuditEntry {
131    /// Create a new audit entry
132    pub fn new(event_type: AuditEventType, action: impl Into<String>) -> Self {
133        let timestamp = SystemTime::now();
134        let micros = timestamp
135            .duration_since(SystemTime::UNIX_EPOCH)
136            .unwrap()
137            .as_micros();
138
139        // Generate a simple unique ID based on timestamp and random component
140        let random_component = (micros % 1000000) as u32;
141        let id = format!("audit-{}-{}", micros, random_component);
142
143        Self {
144            id,
145            timestamp,
146            event_type,
147            severity: AuditSeverity::Info,
148            outcome: AuditOutcome::Success,
149            resource: None,
150            action: action.into(),
151            details: None,
152            metadata: AuditMetadata::default(),
153            duration_ms: None,
154        }
155    }
156
157    /// Set severity
158    pub fn with_severity(mut self, severity: AuditSeverity) -> Self {
159        self.severity = severity;
160        self
161    }
162
163    /// Set outcome
164    pub fn with_outcome(mut self, outcome: AuditOutcome) -> Self {
165        self.outcome = outcome;
166        self
167    }
168
169    /// Set resource
170    pub fn with_resource(mut self, resource: impl Into<String>) -> Self {
171        self.resource = Some(resource.into());
172        self
173    }
174
175    /// Set details
176    pub fn with_details(mut self, details: impl Into<String>) -> Self {
177        self.details = Some(details.into());
178        self
179    }
180
181    /// Set user ID
182    pub fn with_user(mut self, user_id: impl Into<String>) -> Self {
183        self.metadata.user_id = Some(user_id.into());
184        self
185    }
186
187    /// Set IP address
188    pub fn with_ip(mut self, ip: impl Into<String>) -> Self {
189        self.metadata.ip_address = Some(ip.into());
190        self
191    }
192
193    /// Set duration
194    pub fn with_duration(mut self, duration_ms: u64) -> Self {
195        self.duration_ms = Some(duration_ms);
196        self
197    }
198}
199
200/// Audit backend trait
201pub trait AuditBackend: Send + Sync {
202    /// Write an audit entry
203    fn write(&mut self, entry: &AuditEntry) -> Result<(), String>;
204
205    /// Flush any buffered entries
206    fn flush(&mut self) -> Result<(), String>;
207}
208
209/// In-memory audit backend
210pub struct MemoryBackend {
211    entries: Arc<Mutex<VecDeque<AuditEntry>>>,
212    max_size: usize,
213}
214
215impl MemoryBackend {
216    pub fn new(max_size: usize) -> Self {
217        Self {
218            entries: Arc::new(Mutex::new(VecDeque::with_capacity(max_size))),
219            max_size,
220        }
221    }
222
223    pub fn get_entries(&self) -> Vec<AuditEntry> {
224        self.entries.lock().unwrap().iter().cloned().collect()
225    }
226
227    pub fn clear(&self) {
228        self.entries.lock().unwrap().clear();
229    }
230}
231
232impl AuditBackend for MemoryBackend {
233    fn write(&mut self, entry: &AuditEntry) -> Result<(), String> {
234        let mut entries = self.entries.lock().unwrap();
235        if entries.len() >= self.max_size {
236            entries.pop_front();
237        }
238        entries.push_back(entry.clone());
239        Ok(())
240    }
241
242    fn flush(&mut self) -> Result<(), String> {
243        Ok(())
244    }
245}
246
247/// File-based audit backend
248pub struct FileBackend {
249    file: Arc<Mutex<File>>,
250    buffer: Arc<Mutex<Vec<String>>>,
251    buffer_size: usize,
252}
253
254impl FileBackend {
255    pub fn new(path: PathBuf) -> Result<Self, String> {
256        let file = OpenOptions::new()
257            .create(true)
258            .append(true)
259            .open(path)
260            .map_err(|e| format!("Failed to open audit log file: {}", e))?;
261
262        Ok(Self {
263            file: Arc::new(Mutex::new(file)),
264            buffer: Arc::new(Mutex::new(Vec::new())),
265            buffer_size: 100,
266        })
267    }
268
269    pub fn with_buffer_size(mut self, size: usize) -> Self {
270        self.buffer_size = size;
271        self
272    }
273}
274
275impl AuditBackend for FileBackend {
276    fn write(&mut self, entry: &AuditEntry) -> Result<(), String> {
277        let json = serde_json::to_string(entry)
278            .map_err(|e| format!("Failed to serialize audit entry: {}", e))?;
279
280        let mut buffer = self.buffer.lock().unwrap();
281        buffer.push(json);
282
283        if buffer.len() >= self.buffer_size {
284            drop(buffer);
285            self.flush()?;
286        }
287
288        Ok(())
289    }
290
291    fn flush(&mut self) -> Result<(), String> {
292        let mut buffer = self.buffer.lock().unwrap();
293        if buffer.is_empty() {
294            return Ok(());
295        }
296
297        let mut file = self.file.lock().unwrap();
298
299        for line in buffer.iter() {
300            writeln!(file, "{}", line).map_err(|e| format!("Failed to write audit log: {}", e))?;
301        }
302
303        file.flush()
304            .map_err(|e| format!("Failed to flush audit log: {}", e))?;
305
306        buffer.clear();
307        Ok(())
308    }
309}
310
311/// Stdout audit backend (for development)
312pub struct StdoutBackend;
313
314impl AuditBackend for StdoutBackend {
315    fn write(&mut self, entry: &AuditEntry) -> Result<(), String> {
316        let json = serde_json::to_string(entry)
317            .map_err(|e| format!("Failed to serialize audit entry: {}", e))?;
318        println!("{}", json);
319        Ok(())
320    }
321
322    fn flush(&mut self) -> Result<(), String> {
323        Ok(())
324    }
325}
326
327/// Audit logger configuration
328#[derive(Debug, Clone)]
329pub struct AuditConfig {
330    /// Enable audit logging
331    pub enabled: bool,
332
333    /// Minimum severity level to log
334    pub min_severity: AuditSeverity,
335
336    /// Event types to log
337    pub event_types: Vec<AuditEventType>,
338
339    /// Include stack traces for errors
340    pub include_stack_traces: bool,
341}
342
343impl Default for AuditConfig {
344    fn default() -> Self {
345        Self {
346            enabled: true,
347            min_severity: AuditSeverity::Info,
348            event_types: vec![
349                AuditEventType::Insert,
350                AuditEventType::Update,
351                AuditEventType::Delete,
352                AuditEventType::Query,
353                AuditEventType::Auth,
354                AuditEventType::Authz,
355                AuditEventType::ConfigChange,
356            ],
357            include_stack_traces: false,
358        }
359    }
360}
361
362/// Audit logger
363pub struct AuditLogger {
364    config: AuditConfig,
365    backends: Arc<Mutex<Vec<Box<dyn AuditBackend>>>>,
366}
367
368impl AuditLogger {
369    /// Create a new audit logger
370    pub fn new(config: AuditConfig) -> Self {
371        Self {
372            config,
373            backends: Arc::new(Mutex::new(Vec::new())),
374        }
375    }
376
377    /// Create with default configuration
378    pub fn default() -> Self {
379        Self::new(AuditConfig::default())
380    }
381
382    /// Add a backend
383    pub fn add_backend(&self, backend: Box<dyn AuditBackend>) {
384        self.backends.lock().unwrap().push(backend);
385    }
386
387    /// Log an audit entry
388    pub fn log(&self, entry: AuditEntry) -> Result<(), String> {
389        if !self.config.enabled {
390            return Ok(());
391        }
392
393        // Check severity filter
394        if entry.severity < self.config.min_severity {
395            return Ok(());
396        }
397
398        // Check event type filter
399        if !self.config.event_types.is_empty()
400            && !self.config.event_types.contains(&entry.event_type)
401        {
402            return Ok(());
403        }
404
405        // Write to all backends
406        let mut backends = self.backends.lock().unwrap();
407        for backend in backends.iter_mut() {
408            backend.write(&entry)?;
409        }
410
411        Ok(())
412    }
413
414    /// Flush all backends
415    pub fn flush(&self) -> Result<(), String> {
416        let mut backends = self.backends.lock().unwrap();
417        for backend in backends.iter_mut() {
418            backend.flush()?;
419        }
420        Ok(())
421    }
422
423    /// Log an insert operation
424    pub fn log_insert(&self, resource: &str, user_id: Option<&str>) -> Result<(), String> {
425        let mut entry = AuditEntry::new(AuditEventType::Insert, "insert vector");
426        entry = entry.with_resource(resource);
427        if let Some(user) = user_id {
428            entry = entry.with_user(user);
429        }
430        self.log(entry)
431    }
432
433    /// Log a query operation
434    pub fn log_query(
435        &self,
436        query_type: &str,
437        duration_ms: u64,
438        user_id: Option<&str>,
439    ) -> Result<(), String> {
440        let mut entry = AuditEntry::new(AuditEventType::Query, format!("query: {}", query_type));
441        entry = entry.with_duration(duration_ms);
442        if let Some(user) = user_id {
443            entry = entry.with_user(user);
444        }
445        self.log(entry)
446    }
447
448    /// Log a delete operation
449    pub fn log_delete(&self, resource: &str, user_id: Option<&str>) -> Result<(), String> {
450        let mut entry = AuditEntry::new(AuditEventType::Delete, "delete vector");
451        entry = entry
452            .with_resource(resource)
453            .with_severity(AuditSeverity::Warning);
454        if let Some(user) = user_id {
455            entry = entry.with_user(user);
456        }
457        self.log(entry)
458    }
459
460    /// Log an authentication event
461    pub fn log_auth(&self, user_id: &str, ip: &str, outcome: AuditOutcome) -> Result<(), String> {
462        let entry = AuditEntry::new(AuditEventType::Auth, "authentication attempt")
463            .with_user(user_id)
464            .with_ip(ip)
465            .with_outcome(outcome)
466            .with_severity(if outcome == AuditOutcome::Success {
467                AuditSeverity::Info
468            } else {
469                AuditSeverity::Warning
470            });
471        self.log(entry)
472    }
473
474    /// Log an authorization event
475    pub fn log_authz(
476        &self,
477        user_id: &str,
478        resource: &str,
479        action: &str,
480        outcome: AuditOutcome,
481    ) -> Result<(), String> {
482        let entry = AuditEntry::new(AuditEventType::Authz, format!("authz: {}", action))
483            .with_user(user_id)
484            .with_resource(resource)
485            .with_outcome(outcome)
486            .with_severity(if outcome == AuditOutcome::Denied {
487                AuditSeverity::Warning
488            } else {
489                AuditSeverity::Info
490            });
491        self.log(entry)
492    }
493}
494
495impl Drop for AuditLogger {
496    fn drop(&mut self) {
497        let _ = self.flush();
498    }
499}
500
501#[cfg(test)]
502mod tests {
503    use super::*;
504    use std::sync::Arc;
505
506    #[test]
507    fn test_audit_entry_creation() {
508        let entry = AuditEntry::new(AuditEventType::Insert, "test action");
509        assert_eq!(entry.event_type, AuditEventType::Insert);
510        assert_eq!(entry.action, "test action");
511        assert_eq!(entry.severity, AuditSeverity::Info);
512        assert_eq!(entry.outcome, AuditOutcome::Success);
513    }
514
515    #[test]
516    fn test_audit_entry_builder() {
517        let entry = AuditEntry::new(AuditEventType::Query, "test query")
518            .with_severity(AuditSeverity::Warning)
519            .with_outcome(AuditOutcome::Failure)
520            .with_resource("vector_123")
521            .with_user("user_456")
522            .with_ip("192.168.1.1")
523            .with_duration(100);
524
525        assert_eq!(entry.severity, AuditSeverity::Warning);
526        assert_eq!(entry.outcome, AuditOutcome::Failure);
527        assert_eq!(entry.resource, Some("vector_123".to_string()));
528        assert_eq!(entry.metadata.user_id, Some("user_456".to_string()));
529        assert_eq!(entry.metadata.ip_address, Some("192.168.1.1".to_string()));
530        assert_eq!(entry.duration_ms, Some(100));
531    }
532
533    #[test]
534    fn test_memory_backend() {
535        let mut backend = MemoryBackend::new(10);
536
537        let entry = AuditEntry::new(AuditEventType::Insert, "test");
538        backend.write(&entry).unwrap();
539
540        let entries = backend.get_entries();
541        assert_eq!(entries.len(), 1);
542        assert_eq!(entries[0].action, "test");
543    }
544
545    #[test]
546    fn test_memory_backend_overflow() {
547        let mut backend = MemoryBackend::new(3);
548
549        for i in 0..5 {
550            let entry = AuditEntry::new(AuditEventType::Insert, format!("action_{}", i));
551            backend.write(&entry).unwrap();
552        }
553
554        let entries = backend.get_entries();
555        assert_eq!(entries.len(), 3);
556        // Should have the last 3 entries
557        assert_eq!(entries[0].action, "action_2");
558        assert_eq!(entries[2].action, "action_4");
559    }
560
561    #[test]
562    fn test_audit_logger() {
563        let logger = AuditLogger::default();
564        let backend = Box::new(MemoryBackend::new(100));
565        let backend_ref = unsafe {
566            let ptr = &*backend as *const MemoryBackend;
567            &*ptr
568        };
569        logger.add_backend(backend);
570
571        let entry = AuditEntry::new(AuditEventType::Query, "test query");
572        logger.log(entry).unwrap();
573
574        let entries = backend_ref.get_entries();
575        assert_eq!(entries.len(), 1);
576    }
577
578    #[test]
579    fn test_severity_filtering() {
580        let mut config = AuditConfig::default();
581        config.min_severity = AuditSeverity::Warning;
582
583        let logger = AuditLogger::new(config);
584        let backend = Box::new(MemoryBackend::new(100));
585        let backend_ref = unsafe {
586            let ptr = &*backend as *const MemoryBackend;
587            &*ptr
588        };
589        logger.add_backend(backend);
590
591        // Info level - should be filtered
592        let entry1 =
593            AuditEntry::new(AuditEventType::Query, "info entry").with_severity(AuditSeverity::Info);
594        logger.log(entry1).unwrap();
595
596        // Warning level - should be logged
597        let entry2 = AuditEntry::new(AuditEventType::Query, "warning entry")
598            .with_severity(AuditSeverity::Warning);
599        logger.log(entry2).unwrap();
600
601        let entries = backend_ref.get_entries();
602        assert_eq!(entries.len(), 1);
603        assert_eq!(entries[0].action, "warning entry");
604    }
605
606    #[test]
607    fn test_event_type_filtering() {
608        let mut config = AuditConfig::default();
609        config.event_types = vec![AuditEventType::Insert, AuditEventType::Delete];
610
611        let logger = AuditLogger::new(config);
612        let backend = Box::new(MemoryBackend::new(100));
613        let backend_ref = unsafe {
614            let ptr = &*backend as *const MemoryBackend;
615            &*ptr
616        };
617        logger.add_backend(backend);
618
619        // Insert - should be logged
620        logger.log_insert("vec_1", Some("user_1")).unwrap();
621
622        // Query - should be filtered
623        logger.log_query("knn", 100, Some("user_1")).unwrap();
624
625        // Delete - should be logged
626        logger.log_delete("vec_2", Some("user_1")).unwrap();
627
628        let entries = backend_ref.get_entries();
629        assert_eq!(entries.len(), 2);
630    }
631
632    #[test]
633    fn test_helper_methods() {
634        let logger = AuditLogger::default();
635        let backend = Box::new(MemoryBackend::new(100));
636        let backend_ref = unsafe {
637            let ptr = &*backend as *const MemoryBackend;
638            &*ptr
639        };
640        logger.add_backend(backend);
641
642        logger.log_insert("vec_1", Some("user_1")).unwrap();
643        logger.log_query("knn", 50, Some("user_2")).unwrap();
644        logger.log_delete("vec_3", Some("user_3")).unwrap();
645        logger
646            .log_auth("user_4", "192.168.1.1", AuditOutcome::Success)
647            .unwrap();
648        logger
649            .log_authz("user_5", "vec_5", "read", AuditOutcome::Denied)
650            .unwrap();
651
652        let entries = backend_ref.get_entries();
653        assert_eq!(entries.len(), 5);
654    }
655
656    #[test]
657    fn test_disabled_logger() {
658        let mut config = AuditConfig::default();
659        config.enabled = false;
660
661        let logger = AuditLogger::new(config);
662        let backend = Box::new(MemoryBackend::new(100));
663        let backend_ref = unsafe {
664            let ptr = &*backend as *const MemoryBackend;
665            &*ptr
666        };
667        logger.add_backend(backend);
668
669        logger.log_insert("vec_1", Some("user_1")).unwrap();
670
671        let entries = backend_ref.get_entries();
672        assert_eq!(entries.len(), 0);
673    }
674}
vecstore/audit.rs

vecstore/
audit.rs
