Skip to main content

oxirs_stream/
enterprise_audit.rs

1//! # Enterprise Audit Logging System
2//!
3//! Comprehensive audit logging for compliance, security monitoring, and operational visibility.
4//! Supports structured logging, log aggregation, retention policies, and compliance reporting.
5
6use anyhow::Result;
7use chrono::{DateTime, Utc};
8use serde::{Deserialize, Serialize};
9use std::collections::HashMap;
10use std::fmt;
11use std::path::PathBuf;
12use std::sync::Arc;
13use tokio::sync::RwLock;
14use tracing::{debug, info};
15use uuid::Uuid;
16
17/// Enterprise audit logger configuration
18#[derive(Debug, Clone, Serialize, Deserialize)]
19pub struct EnterpriseAuditConfig {
20    /// Enable audit logging
21    pub enabled: bool,
22    /// Audit log storage backend
23    pub storage: AuditStorageConfig,
24    /// Event filtering configuration
25    pub filtering: AuditFilterConfig,
26    /// Retention policy
27    pub retention: AuditRetentionConfig,
28    /// Compliance configuration
29    pub compliance: ComplianceConfig,
30    /// Encryption configuration for audit logs
31    pub encryption: AuditEncryptionConfig,
32    /// Real-time streaming of audit events
33    pub streaming: AuditStreamingConfig,
34}
35
36impl Default for EnterpriseAuditConfig {
37    fn default() -> Self {
38        Self {
39            enabled: true,
40            storage: AuditStorageConfig::default(),
41            filtering: AuditFilterConfig::default(),
42            retention: AuditRetentionConfig::default(),
43            compliance: ComplianceConfig::default(),
44            encryption: AuditEncryptionConfig::default(),
45            streaming: AuditStreamingConfig::default(),
46        }
47    }
48}
49
50/// Audit log storage configuration
51#[derive(Debug, Clone, Serialize, Deserialize)]
52pub struct AuditStorageConfig {
53    /// Storage backend type
54    pub backend: AuditStorageBackend,
55    /// File storage path (for file backend)
56    pub file_path: Option<PathBuf>,
57    /// Database connection string (for database backend)
58    pub database_url: Option<String>,
59    /// S3 bucket configuration (for S3 backend)
60    pub s3_config: Option<S3AuditConfig>,
61    /// Buffer size for batching
62    pub buffer_size: usize,
63    /// Flush interval in seconds
64    pub flush_interval_secs: u64,
65}
66
67impl Default for AuditStorageConfig {
68    fn default() -> Self {
69        Self {
70            backend: AuditStorageBackend::File,
71            file_path: Some(PathBuf::from("/var/log/oxirs/audit.jsonl")),
72            database_url: None,
73            s3_config: None,
74            buffer_size: 1000,
75            flush_interval_secs: 60,
76        }
77    }
78}
79
80/// Audit storage backend types
81#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
82pub enum AuditStorageBackend {
83    /// File-based storage (JSONL format)
84    File,
85    /// Database storage (PostgreSQL, MySQL)
86    Database,
87    /// S3-compatible object storage
88    S3,
89    /// Elasticsearch
90    Elasticsearch,
91    /// Splunk
92    Splunk,
93    /// Custom backend
94    Custom,
95}
96
97/// S3 audit storage configuration
98#[derive(Debug, Clone, Serialize, Deserialize)]
99pub struct S3AuditConfig {
100    /// S3 bucket name
101    pub bucket: String,
102    /// S3 region
103    pub region: String,
104    /// S3 prefix for audit logs
105    pub prefix: String,
106    /// AWS access key ID
107    pub access_key_id: Option<String>,
108    /// AWS secret access key
109    pub secret_access_key: Option<String>,
110    /// Server-side encryption
111    pub server_side_encryption: bool,
112}
113
114/// Audit event filtering configuration
115#[derive(Debug, Clone, Serialize, Deserialize)]
116pub struct AuditFilterConfig {
117    /// Minimum severity level to log
118    pub min_severity: AuditSeverity,
119    /// Event types to include
120    pub include_event_types: Vec<AuditEventType>,
121    /// Event types to exclude
122    pub exclude_event_types: Vec<AuditEventType>,
123    /// User IDs to exclude from auditing (e.g., system users)
124    pub exclude_users: Vec<String>,
125    /// Resource patterns to exclude
126    pub exclude_resources: Vec<String>,
127}
128
129impl Default for AuditFilterConfig {
130    fn default() -> Self {
131        Self {
132            min_severity: AuditSeverity::Info,
133            include_event_types: vec![],
134            exclude_event_types: vec![],
135            exclude_users: vec![],
136            exclude_resources: vec![],
137        }
138    }
139}
140
141/// Audit retention policy configuration
142#[derive(Debug, Clone, Serialize, Deserialize)]
143pub struct AuditRetentionConfig {
144    /// Retention period in days
145    pub retention_days: u32,
146    /// Archive old logs
147    pub archive_enabled: bool,
148    /// Archive destination
149    pub archive_destination: Option<String>,
150    /// Compression for archived logs
151    pub archive_compression: CompressionType,
152    /// Automatic cleanup of expired logs
153    pub auto_cleanup: bool,
154}
155
156impl Default for AuditRetentionConfig {
157    fn default() -> Self {
158        Self {
159            retention_days: 365, // 1 year default
160            archive_enabled: true,
161            archive_destination: None,
162            archive_compression: CompressionType::Gzip,
163            auto_cleanup: true,
164        }
165    }
166}
167
168/// Compression types for audit logs
169#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
170pub enum CompressionType {
171    None,
172    Gzip,
173    Zstd,
174    Bzip2,
175}
176
177/// Compliance configuration
178#[derive(Debug, Clone, Serialize, Deserialize)]
179pub struct ComplianceConfig {
180    /// Compliance standards to adhere to
181    pub standards: Vec<ComplianceStandard>,
182    /// Enable compliance reporting
183    pub reporting_enabled: bool,
184    /// Report generation interval in days
185    pub report_interval_days: u32,
186    /// Report destination
187    pub report_destination: Option<String>,
188}
189
190impl Default for ComplianceConfig {
191    fn default() -> Self {
192        Self {
193            standards: vec![ComplianceStandard::SOC2],
194            reporting_enabled: true,
195            report_interval_days: 30,
196            report_destination: None,
197        }
198    }
199}
200
201/// Compliance standards
202#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
203pub enum ComplianceStandard {
204    /// GDPR (General Data Protection Regulation)
205    GDPR,
206    /// HIPAA (Health Insurance Portability and Accountability Act)
207    HIPAA,
208    /// SOC 2 (Service Organization Control 2)
209    SOC2,
210    /// PCI DSS (Payment Card Industry Data Security Standard)
211    PCIDSS,
212    /// ISO 27001
213    ISO27001,
214}
215
216impl fmt::Display for ComplianceStandard {
217    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
218        match self {
219            ComplianceStandard::GDPR => write!(f, "GDPR"),
220            ComplianceStandard::HIPAA => write!(f, "HIPAA"),
221            ComplianceStandard::SOC2 => write!(f, "SOC2"),
222            ComplianceStandard::PCIDSS => write!(f, "PCI-DSS"),
223            ComplianceStandard::ISO27001 => write!(f, "ISO 27001"),
224        }
225    }
226}
227
228/// Audit log encryption configuration
229#[derive(Debug, Clone, Serialize, Deserialize)]
230pub struct AuditEncryptionConfig {
231    /// Enable encryption for audit logs
232    pub enabled: bool,
233    /// Encryption algorithm
234    pub algorithm: EncryptionAlgorithm,
235    /// Key management system
236    pub key_management: KeyManagementConfig,
237}
238
239impl Default for AuditEncryptionConfig {
240    fn default() -> Self {
241        Self {
242            enabled: true,
243            algorithm: EncryptionAlgorithm::AES256GCM,
244            key_management: KeyManagementConfig::default(),
245        }
246    }
247}
248
249/// Encryption algorithms for audit logs
250#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
251pub enum EncryptionAlgorithm {
252    AES256GCM,
253    ChaCha20Poly1305,
254    AES256CBC,
255}
256
257/// Key management configuration
258#[derive(Debug, Clone, Serialize, Deserialize)]
259pub struct KeyManagementConfig {
260    /// Key management system type
261    pub kms_type: KmsType,
262    /// KMS endpoint URL
263    pub kms_url: Option<String>,
264    /// Key rotation interval in days
265    pub rotation_interval_days: u32,
266}
267
268impl Default for KeyManagementConfig {
269    fn default() -> Self {
270        Self {
271            kms_type: KmsType::Local,
272            kms_url: None,
273            rotation_interval_days: 90,
274        }
275    }
276}
277
278/// Key management system types
279#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
280pub enum KmsType {
281    /// Local key storage (for development)
282    Local,
283    /// AWS KMS
284    AwsKms,
285    /// Azure Key Vault
286    AzureKeyVault,
287    /// Google Cloud KMS
288    GcpKms,
289    /// HashiCorp Vault
290    HashiCorpVault,
291}
292
293/// Audit streaming configuration
294#[derive(Debug, Clone, Serialize, Deserialize, Default)]
295pub struct AuditStreamingConfig {
296    /// Enable real-time streaming of audit events
297    pub enabled: bool,
298    /// Streaming destinations
299    pub destinations: Vec<StreamingDestination>,
300}
301
302/// Streaming destination configuration
303#[derive(Debug, Clone, Serialize, Deserialize)]
304pub struct StreamingDestination {
305    /// Destination type
306    pub destination_type: DestinationType,
307    /// Destination endpoint URL
308    pub endpoint: String,
309    /// Authentication configuration
310    pub auth: Option<DestinationAuth>,
311}
312
313/// Destination types for audit streaming
314#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
315pub enum DestinationType {
316    Kafka,
317    Kinesis,
318    Webhook,
319    SIEM,
320}
321
322/// Authentication for streaming destinations
323#[derive(Debug, Clone, Serialize, Deserialize)]
324pub struct DestinationAuth {
325    pub auth_type: AuthType,
326    pub credentials: HashMap<String, String>,
327}
328
329/// Authentication types
330#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
331pub enum AuthType {
332    ApiKey,
333    OAuth2,
334    Basic,
335    Certificate,
336}
337
338/// Audit event types
339#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Hash)]
340pub enum AuditEventType {
341    /// Authentication events
342    Authentication,
343    /// Authorization events
344    Authorization,
345    /// Data access events
346    DataAccess,
347    /// Data modification events
348    DataModification,
349    /// Configuration changes
350    ConfigurationChange,
351    /// Security events
352    Security,
353    /// System events
354    System,
355    /// Administrative actions
356    Administrative,
357    /// Compliance events
358    Compliance,
359}
360
361impl fmt::Display for AuditEventType {
362    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
363        match self {
364            AuditEventType::Authentication => write!(f, "Authentication"),
365            AuditEventType::Authorization => write!(f, "Authorization"),
366            AuditEventType::DataAccess => write!(f, "DataAccess"),
367            AuditEventType::DataModification => write!(f, "DataModification"),
368            AuditEventType::ConfigurationChange => write!(f, "ConfigurationChange"),
369            AuditEventType::Security => write!(f, "Security"),
370            AuditEventType::System => write!(f, "System"),
371            AuditEventType::Administrative => write!(f, "Administrative"),
372            AuditEventType::Compliance => write!(f, "Compliance"),
373        }
374    }
375}
376
377/// Audit severity levels
378#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
379pub enum AuditSeverity {
380    Debug,
381    Info,
382    Warning,
383    Error,
384    Critical,
385}
386
387impl fmt::Display for AuditSeverity {
388    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
389        match self {
390            AuditSeverity::Debug => write!(f, "DEBUG"),
391            AuditSeverity::Info => write!(f, "INFO"),
392            AuditSeverity::Warning => write!(f, "WARNING"),
393            AuditSeverity::Error => write!(f, "ERROR"),
394            AuditSeverity::Critical => write!(f, "CRITICAL"),
395        }
396    }
397}
398
399/// Enterprise audit event
400#[derive(Debug, Clone, Serialize, Deserialize)]
401pub struct EnterpriseAuditEvent {
402    /// Unique event ID
403    pub event_id: String,
404    /// Event timestamp
405    pub timestamp: DateTime<Utc>,
406    /// Event type
407    pub event_type: AuditEventType,
408    /// Severity level
409    pub severity: AuditSeverity,
410    /// User who triggered the event
411    pub user_id: Option<String>,
412    /// User's IP address
413    pub source_ip: Option<String>,
414    /// Resource being accessed/modified
415    pub resource: String,
416    /// Action performed
417    pub action: String,
418    /// Result of the action
419    pub result: ActionResult,
420    /// Additional details
421    pub details: HashMap<String, serde_json::Value>,
422    /// Compliance tags
423    pub compliance_tags: Vec<ComplianceStandard>,
424    /// Session ID
425    pub session_id: Option<String>,
426    /// Request ID
427    pub request_id: Option<String>,
428    /// Correlation ID for distributed tracing
429    pub correlation_id: Option<String>,
430}
431
432/// Action result
433#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
434pub enum ActionResult {
435    Success,
436    Failure,
437    PartialSuccess,
438    Denied,
439}
440
441impl fmt::Display for ActionResult {
442    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
443        match self {
444            ActionResult::Success => write!(f, "SUCCESS"),
445            ActionResult::Failure => write!(f, "FAILURE"),
446            ActionResult::PartialSuccess => write!(f, "PARTIAL_SUCCESS"),
447            ActionResult::Denied => write!(f, "DENIED"),
448        }
449    }
450}
451
452/// Enterprise audit logger
453pub struct EnterpriseAuditLogger {
454    config: EnterpriseAuditConfig,
455    buffer: Arc<RwLock<Vec<EnterpriseAuditEvent>>>,
456    metrics: Arc<RwLock<AuditMetrics>>,
457}
458
459/// Audit metrics
460#[derive(Debug, Clone, Default, Serialize, Deserialize)]
461pub struct AuditMetrics {
462    /// Total events logged
463    pub events_total: u64,
464    /// Events by type
465    pub events_by_type: HashMap<String, u64>,
466    /// Events by severity
467    pub events_by_severity: HashMap<String, u64>,
468    /// Events by result
469    pub events_by_result: HashMap<String, u64>,
470    /// Failed to log events
471    pub events_failed: u64,
472    /// Buffer flushes
473    pub buffer_flushes: u64,
474    /// Last flush timestamp
475    pub last_flush: Option<DateTime<Utc>>,
476}
477
478impl EnterpriseAuditLogger {
479    /// Create a new enterprise audit logger
480    pub fn new(config: EnterpriseAuditConfig) -> Self {
481        Self {
482            config,
483            buffer: Arc::new(RwLock::new(Vec::new())),
484            metrics: Arc::new(RwLock::new(AuditMetrics::default())),
485        }
486    }
487
488    /// Initialize the audit logger
489    pub async fn initialize(&self) -> Result<()> {
490        if !self.config.enabled {
491            info!("Enterprise audit logging is disabled");
492            return Ok(());
493        }
494
495        info!("Initializing enterprise audit logger");
496
497        // Initialize storage backend
498        self.initialize_storage().await?;
499
500        // Start background flusher
501        self.start_background_flusher().await?;
502
503        // Initialize streaming if enabled
504        if self.config.streaming.enabled {
505            self.initialize_streaming().await?;
506        }
507
508        info!("Enterprise audit logger initialized successfully");
509        Ok(())
510    }
511
512    /// Initialize storage backend
513    async fn initialize_storage(&self) -> Result<()> {
514        match self.config.storage.backend {
515            AuditStorageBackend::File => {
516                if let Some(path) = &self.config.storage.file_path {
517                    // Ensure directory exists
518                    if let Some(parent) = path.parent() {
519                        tokio::fs::create_dir_all(parent).await?;
520                    }
521                    debug!("File storage initialized: {:?}", path);
522                }
523            }
524            AuditStorageBackend::Database => {
525                debug!("Database storage initialization (placeholder)");
526            }
527            AuditStorageBackend::S3 => {
528                debug!("S3 storage initialization (placeholder)");
529            }
530            AuditStorageBackend::Elasticsearch => {
531                debug!("Elasticsearch storage initialization (placeholder)");
532            }
533            AuditStorageBackend::Splunk => {
534                debug!("Splunk storage initialization (placeholder)");
535            }
536            AuditStorageBackend::Custom => {
537                debug!("Custom storage initialization (placeholder)");
538            }
539        }
540
541        Ok(())
542    }
543
544    /// Start background flusher task
545    async fn start_background_flusher(&self) -> Result<()> {
546        debug!("Starting background audit log flusher");
547        // In a real implementation, this would spawn a tokio task
548        // that periodically flushes the buffer to storage
549        Ok(())
550    }
551
552    /// Initialize streaming destinations
553    async fn initialize_streaming(&self) -> Result<()> {
554        debug!("Initializing audit event streaming");
555        for destination in &self.config.streaming.destinations {
556            debug!(
557                "Setting up streaming to {:?}: {}",
558                destination.destination_type, destination.endpoint
559            );
560        }
561        Ok(())
562    }
563
564    /// Log an audit event
565    pub async fn log_event(&self, event: EnterpriseAuditEvent) -> Result<()> {
566        if !self.config.enabled {
567            return Ok(());
568        }
569
570        // Apply filtering
571        if !self.should_log_event(&event).await {
572            return Ok(());
573        }
574
575        // Add to buffer
576        {
577            let mut buffer = self.buffer.write().await;
578            buffer.push(event.clone());
579
580            // Check if buffer needs flushing
581            if buffer.len() >= self.config.storage.buffer_size {
582                drop(buffer);
583                self.flush_buffer().await?;
584            }
585        }
586
587        // Update metrics
588        {
589            let mut metrics = self.metrics.write().await;
590            metrics.events_total += 1;
591            *metrics
592                .events_by_type
593                .entry(event.event_type.to_string())
594                .or_insert(0) += 1;
595            *metrics
596                .events_by_severity
597                .entry(event.severity.to_string())
598                .or_insert(0) += 1;
599            *metrics
600                .events_by_result
601                .entry(event.result.to_string())
602                .or_insert(0) += 1;
603        }
604
605        // Stream event if enabled
606        if self.config.streaming.enabled {
607            self.stream_event(&event).await?;
608        }
609
610        Ok(())
611    }
612
613    /// Check if event should be logged based on filters
614    async fn should_log_event(&self, event: &EnterpriseAuditEvent) -> bool {
615        // Check severity level
616        if event.severity < self.config.filtering.min_severity {
617            return false;
618        }
619
620        // Check excluded event types
621        if self
622            .config
623            .filtering
624            .exclude_event_types
625            .contains(&event.event_type)
626        {
627            return false;
628        }
629
630        // Check include list (if not empty)
631        if !self.config.filtering.include_event_types.is_empty()
632            && !self
633                .config
634                .filtering
635                .include_event_types
636                .contains(&event.event_type)
637        {
638            return false;
639        }
640
641        // Check excluded users
642        if let Some(user_id) = &event.user_id {
643            if self.config.filtering.exclude_users.contains(user_id) {
644                return false;
645            }
646        }
647
648        true
649    }
650
651    /// Flush buffer to storage
652    pub async fn flush_buffer(&self) -> Result<()> {
653        let events = {
654            let mut buffer = self.buffer.write().await;
655            if buffer.is_empty() {
656                return Ok(());
657            }
658            std::mem::take(&mut *buffer)
659        };
660
661        debug!("Flushing {} audit events to storage", events.len());
662
663        // Write to storage. If the write fails, restore the taken events to the
664        // front of the buffer so the audit trail is not silently discarded and a
665        // subsequent flush can retry.
666        if let Err(e) = self.write_to_storage(&events).await {
667            let mut buffer = self.buffer.write().await;
668            let mut restored = events;
669            restored.append(&mut buffer);
670            *buffer = restored;
671            return Err(e);
672        }
673
674        // Update metrics
675        {
676            let mut metrics = self.metrics.write().await;
677            metrics.buffer_flushes += 1;
678            metrics.last_flush = Some(Utc::now());
679        }
680
681        Ok(())
682    }
683
684    /// Write events to storage.
685    ///
686    /// The file backend performs a real appending JSONL write. Backends that are
687    /// not yet implemented return an explicit error (fail-loud) rather than
688    /// silently succeeding, so callers never believe an audit trail was
689    /// persisted when it was not.
690    async fn write_to_storage(&self, events: &[EnterpriseAuditEvent]) -> Result<()> {
691        match self.config.storage.backend {
692            AuditStorageBackend::File => {
693                let path = self.config.storage.file_path.as_ref().ok_or_else(|| {
694                    anyhow::anyhow!(
695                        "Audit File storage backend selected but no file_path configured"
696                    )
697                })?;
698
699                let mut content = String::new();
700                for event in events {
701                    let json = serde_json::to_string(event)?;
702                    content.push_str(&json);
703                    content.push('\n');
704                }
705
706                if let Some(parent) = path.parent() {
707                    if !parent.as_os_str().is_empty() {
708                        tokio::fs::create_dir_all(parent).await?;
709                    }
710                }
711
712                use tokio::io::AsyncWriteExt;
713                let mut file = tokio::fs::OpenOptions::new()
714                    .create(true)
715                    .append(true)
716                    .open(path)
717                    .await?;
718                file.write_all(content.as_bytes()).await?;
719                file.flush().await?;
720
721                debug!("Wrote {} bytes to audit log {:?}", content.len(), path);
722                Ok(())
723            }
724            other => Err(anyhow::anyhow!(
725                "Audit storage backend {:?} is not implemented; cannot persist {} event(s)",
726                other,
727                events.len()
728            )),
729        }
730    }
731
732    /// Stream event to real-time destinations
733    async fn stream_event(&self, event: &EnterpriseAuditEvent) -> Result<()> {
734        for destination in &self.config.streaming.destinations {
735            debug!(
736                "Streaming event {} to {:?}",
737                event.event_id, destination.destination_type
738            );
739            // In a real implementation, send to actual destination
740        }
741        Ok(())
742    }
743
744    /// Get audit metrics
745    pub async fn get_metrics(&self) -> AuditMetrics {
746        self.metrics.read().await.clone()
747    }
748
749    /// Generate compliance report
750    pub async fn generate_compliance_report(
751        &self,
752        standard: ComplianceStandard,
753        start_date: DateTime<Utc>,
754        end_date: DateTime<Utc>,
755    ) -> Result<ComplianceReport> {
756        info!(
757            "Generating compliance report for {} from {} to {}",
758            standard, start_date, end_date
759        );
760
761        // Query the persisted audit trail for events in range that carry this
762        // standard's compliance tag, then compute real totals and findings.
763        let events = self.query_events(start_date, end_date, standard).await?;
764        let total_events = events.len() as u64;
765
766        let mut findings = Vec::new();
767        for event in &events {
768            let finding = match event.result {
769                ActionResult::Denied => Some((
770                    FindingType::NonCompliance,
771                    format!("Denied {} on resource '{}'", event.action, event.resource),
772                    Some(
773                        "Review access control policy and verify the denial was expected"
774                            .to_string(),
775                    ),
776                )),
777                ActionResult::Failure => Some((
778                    FindingType::Warning,
779                    format!("Failed {} on resource '{}'", event.action, event.resource),
780                    Some(
781                        "Investigate the failure and confirm no data integrity impact".to_string(),
782                    ),
783                )),
784                _ => None,
785            };
786
787            if let Some((finding_type, description, remediation)) = finding {
788                findings.push(ComplianceFinding {
789                    finding_id: Uuid::new_v4().to_string(),
790                    finding_type,
791                    severity: event.severity,
792                    description,
793                    affected_events: vec![event.event_id.clone()],
794                    remediation,
795                });
796            }
797        }
798
799        let summary = format!(
800            "{standard} compliance report: {total_events} event(s) analyzed, {} finding(s) ({} non-compliance, {} warning)",
801            findings.len(),
802            findings
803                .iter()
804                .filter(|f| f.finding_type == FindingType::NonCompliance)
805                .count(),
806            findings
807                .iter()
808                .filter(|f| f.finding_type == FindingType::Warning)
809                .count(),
810        );
811
812        Ok(ComplianceReport {
813            standard,
814            report_id: Uuid::new_v4().to_string(),
815            generated_at: Utc::now(),
816            period_start: start_date,
817            period_end: end_date,
818            total_events,
819            findings,
820            summary,
821        })
822    }
823
824    /// Query persisted audit events within `[start, end]` carrying `standard`'s
825    /// compliance tag.
826    ///
827    /// Buffered events are flushed first so the report reflects the full trail.
828    /// Only the File backend supports querying today; other backends return an
829    /// explicit error rather than an empty result.
830    async fn query_events(
831        &self,
832        start: DateTime<Utc>,
833        end: DateTime<Utc>,
834        standard: ComplianceStandard,
835    ) -> Result<Vec<EnterpriseAuditEvent>> {
836        // Ensure everything buffered has been persisted before querying.
837        self.flush_buffer().await?;
838
839        match self.config.storage.backend {
840            AuditStorageBackend::File => {
841                let path = self.config.storage.file_path.as_ref().ok_or_else(|| {
842                    anyhow::anyhow!(
843                        "Audit File storage backend selected but no file_path configured"
844                    )
845                })?;
846
847                let contents = match tokio::fs::read_to_string(path).await {
848                    Ok(contents) => contents,
849                    // No file yet means no events have been recorded.
850                    Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
851                    Err(e) => return Err(e.into()),
852                };
853
854                let mut events = Vec::new();
855                for line in contents.lines() {
856                    let line = line.trim();
857                    if line.is_empty() {
858                        continue;
859                    }
860                    let event: EnterpriseAuditEvent = serde_json::from_str(line)?;
861                    if event.timestamp >= start
862                        && event.timestamp <= end
863                        && event.compliance_tags.contains(&standard)
864                    {
865                        events.push(event);
866                    }
867                }
868                Ok(events)
869            }
870            other => Err(anyhow::anyhow!(
871                "Compliance querying is not implemented for audit storage backend {:?}",
872                other
873            )),
874        }
875    }
876}
877
878/// Compliance report
879#[derive(Debug, Clone, Serialize, Deserialize)]
880pub struct ComplianceReport {
881    /// Compliance standard
882    pub standard: ComplianceStandard,
883    /// Report ID
884    pub report_id: String,
885    /// Report generation timestamp
886    pub generated_at: DateTime<Utc>,
887    /// Report period start
888    pub period_start: DateTime<Utc>,
889    /// Report period end
890    pub period_end: DateTime<Utc>,
891    /// Total events analyzed
892    pub total_events: u64,
893    /// Compliance findings
894    pub findings: Vec<ComplianceFinding>,
895    /// Report summary
896    pub summary: String,
897}
898
899/// Compliance finding
900#[derive(Debug, Clone, Serialize, Deserialize)]
901pub struct ComplianceFinding {
902    /// Finding ID
903    pub finding_id: String,
904    /// Finding type
905    pub finding_type: FindingType,
906    /// Severity
907    pub severity: AuditSeverity,
908    /// Description
909    pub description: String,
910    /// Affected events
911    pub affected_events: Vec<String>,
912    /// Remediation steps
913    pub remediation: Option<String>,
914}
915
916/// Compliance finding types
917#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
918pub enum FindingType {
919    NonCompliance,
920    Warning,
921    BestPractice,
922}
923
924#[cfg(test)]
925mod tests {
926    use super::*;
927
928    #[tokio::test]
929    async fn test_audit_config_default() {
930        let config = EnterpriseAuditConfig::default();
931        assert!(config.enabled);
932        assert_eq!(config.retention.retention_days, 365);
933    }
934
935    #[tokio::test]
936    async fn test_audit_logger_creation() {
937        let config = EnterpriseAuditConfig::default();
938        let logger = EnterpriseAuditLogger::new(config);
939        let metrics = logger.get_metrics().await;
940        assert_eq!(metrics.events_total, 0);
941    }
942
943    #[tokio::test]
944    async fn test_compliance_standard_display() {
945        assert_eq!(ComplianceStandard::GDPR.to_string(), "GDPR");
946        assert_eq!(ComplianceStandard::HIPAA.to_string(), "HIPAA");
947        assert_eq!(ComplianceStandard::SOC2.to_string(), "SOC2");
948    }
949
950    fn temp_audit_config() -> EnterpriseAuditConfig {
951        let mut config = EnterpriseAuditConfig::default();
952        let path = std::env::temp_dir().join(format!("oxirs-audit-{}.jsonl", Uuid::new_v4()));
953        config.storage.file_path = Some(path);
954        config
955    }
956
957    fn sample_event(result: ActionResult) -> EnterpriseAuditEvent {
958        EnterpriseAuditEvent {
959            event_id: Uuid::new_v4().to_string(),
960            timestamp: Utc::now(),
961            event_type: AuditEventType::DataAccess,
962            severity: AuditSeverity::Critical,
963            user_id: Some("user-1".to_string()),
964            source_ip: None,
965            resource: "graph:secret".to_string(),
966            action: "read".to_string(),
967            result,
968            details: HashMap::new(),
969            compliance_tags: vec![ComplianceStandard::SOC2],
970            session_id: None,
971            request_id: None,
972            correlation_id: None,
973        }
974    }
975
976    #[tokio::test]
977    async fn regression_audit_events_persisted_to_file() {
978        let config = temp_audit_config();
979        let path = config
980            .storage
981            .file_path
982            .clone()
983            .expect("file path configured");
984        let logger = EnterpriseAuditLogger::new(config);
985
986        logger
987            .log_event(sample_event(ActionResult::Success))
988            .await
989            .unwrap();
990        logger.flush_buffer().await.unwrap();
991
992        // The audit trail must actually be written to disk (previously a no-op).
993        let contents = tokio::fs::read_to_string(&path).await.unwrap();
994        assert!(contents.contains("graph:secret"));
995
996        let _ = tokio::fs::remove_file(&path).await;
997    }
998
999    #[tokio::test]
1000    async fn regression_unimplemented_backend_fails_loud() {
1001        let mut config = temp_audit_config();
1002        config.storage.backend = AuditStorageBackend::S3;
1003        let logger = EnterpriseAuditLogger::new(config);
1004
1005        logger
1006            .log_event(sample_event(ActionResult::Success))
1007            .await
1008            .unwrap();
1009        // Flushing to an unimplemented backend must error, not silently succeed.
1010        assert!(logger.flush_buffer().await.is_err());
1011        // Events must be retained in the buffer for a retry, not discarded.
1012        assert!(!logger.buffer.read().await.is_empty());
1013    }
1014
1015    #[tokio::test]
1016    async fn regression_compliance_report_reflects_events() {
1017        let config = temp_audit_config();
1018        let path = config
1019            .storage
1020            .file_path
1021            .clone()
1022            .expect("file path configured");
1023        let logger = EnterpriseAuditLogger::new(config);
1024
1025        logger
1026            .log_event(sample_event(ActionResult::Denied))
1027            .await
1028            .unwrap();
1029        logger
1030            .log_event(sample_event(ActionResult::Success))
1031            .await
1032            .unwrap();
1033
1034        let start = Utc::now() - chrono::Duration::hours(1);
1035        let end = Utc::now() + chrono::Duration::hours(1);
1036        let report = logger
1037            .generate_compliance_report(ComplianceStandard::SOC2, start, end)
1038            .await
1039            .unwrap();
1040
1041        // Both SOC2-tagged events must be counted (previously always 0).
1042        assert_eq!(report.total_events, 2);
1043        // The denied event produces a non-compliance finding.
1044        assert_eq!(report.findings.len(), 1);
1045        assert_eq!(report.findings[0].finding_type, FindingType::NonCompliance);
1046
1047        let _ = tokio::fs::remove_file(&path).await;
1048    }
1049
1050    #[tokio::test]
1051    async fn test_audit_severity_ordering() {
1052        assert!(AuditSeverity::Critical > AuditSeverity::Error);
1053        assert!(AuditSeverity::Error > AuditSeverity::Warning);
1054        assert!(AuditSeverity::Warning > AuditSeverity::Info);
1055        assert!(AuditSeverity::Info > AuditSeverity::Debug);
1056    }
1057
1058    #[tokio::test]
1059    async fn test_action_result_display() {
1060        assert_eq!(ActionResult::Success.to_string(), "SUCCESS");
1061        assert_eq!(ActionResult::Failure.to_string(), "FAILURE");
1062        assert_eq!(ActionResult::Denied.to_string(), "DENIED");
1063    }
1064}