grapsus_proxy/
logging.rs

1//! Logging infrastructure for Grapsus proxy
2//!
3//! This module provides structured logging to files for:
4//! - Access logs (request/response data with trace_id)
5//! - Error logs (errors and warnings)
6//! - Audit logs (security events)
7//!
8//! Access log formats supported:
9//! - `json` (default): Structured JSON with all fields
10//! - `combined`: Apache/nginx Combined Log Format with trace_id extension
11
12use anyhow::{Context, Result};
13use parking_lot::Mutex;
14use serde::Serialize;
15use std::fs::{File, OpenOptions};
16use std::io::{BufWriter, Write};
17use std::path::Path;
18use std::sync::Arc;
19use tracing::{error, warn};
20
21use grapsus_config::{AuditLogConfig, LoggingConfig};
22
23/// Access log format
24#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25pub enum AccessLogFormat {
26    /// Structured JSON format (default)
27    Json,
28    /// Apache/nginx Combined Log Format with trace_id extension
29    Combined,
30}
31
32/// Access log entry with trace_id for request correlation
33#[derive(Debug, Serialize)]
34pub struct AccessLogEntry {
35    /// Timestamp in RFC3339 format
36    pub timestamp: String,
37    /// Unique trace ID for request correlation
38    pub trace_id: String,
39    /// HTTP method
40    pub method: String,
41    /// Request path
42    pub path: String,
43    /// Query string (if any)
44    #[serde(skip_serializing_if = "Option::is_none")]
45    pub query: Option<String>,
46    /// HTTP protocol version
47    pub protocol: String,
48    /// Response status code
49    pub status: u16,
50    /// Response body size in bytes
51    pub body_bytes: u64,
52    /// Request duration in milliseconds
53    pub duration_ms: u64,
54    /// Client IP address
55    pub client_ip: String,
56    /// User-Agent header
57    #[serde(skip_serializing_if = "Option::is_none")]
58    pub user_agent: Option<String>,
59    /// Referer header
60    #[serde(skip_serializing_if = "Option::is_none")]
61    pub referer: Option<String>,
62    /// Host header
63    #[serde(skip_serializing_if = "Option::is_none")]
64    pub host: Option<String>,
65    /// Matched route ID
66    #[serde(skip_serializing_if = "Option::is_none")]
67    pub route_id: Option<String>,
68    /// Selected upstream
69    #[serde(skip_serializing_if = "Option::is_none")]
70    pub upstream: Option<String>,
71    /// Number of upstream attempts
72    pub upstream_attempts: u32,
73    /// Instance ID of the proxy
74    pub instance_id: String,
75    /// Namespace (for scoped requests)
76    #[serde(skip_serializing_if = "Option::is_none")]
77    pub namespace: Option<String>,
78    /// Service (for scoped requests)
79    #[serde(skip_serializing_if = "Option::is_none")]
80    pub service: Option<String>,
81    /// Response body size in bytes (bytes sent to client)
82    pub body_bytes_sent: u64,
83    /// Upstream address that handled the request (IP:port)
84    #[serde(skip_serializing_if = "Option::is_none")]
85    pub upstream_addr: Option<String>,
86    /// Whether the connection to upstream was reused (connection pooling)
87    pub connection_reused: bool,
88    /// Whether the request hit a rate limit (429 response)
89    pub rate_limit_hit: bool,
90    /// GeoIP country code (ISO 3166-1 alpha-2)
91    #[serde(skip_serializing_if = "Option::is_none")]
92    pub geo_country: Option<String>,
93}
94
95impl AccessLogEntry {
96    /// Format the entry as a string based on the specified format.
97    ///
98    /// When `fields` is provided, disabled fields are omitted from JSON output.
99    pub fn format(
100        &self,
101        format: AccessLogFormat,
102        fields: Option<&grapsus_config::AccessLogFields>,
103    ) -> String {
104        match format {
105            AccessLogFormat::Json => self.format_json(fields),
106            AccessLogFormat::Combined => self.format_combined(),
107        }
108    }
109
110    /// Format as JSON, optionally filtering fields based on config
111    fn format_json(&self, fields: Option<&grapsus_config::AccessLogFields>) -> String {
112        let fields = match fields {
113            Some(f) => f,
114            None => return serde_json::to_string(self).unwrap_or_else(|_| "{}".to_string()),
115        };
116
117        // Build a filtered JSON object based on enabled fields
118        let mut map = serde_json::Map::new();
119        if fields.timestamp {
120            map.insert(
121                "timestamp".to_string(),
122                serde_json::Value::String(self.timestamp.clone()),
123            );
124        }
125        if fields.trace_id {
126            map.insert(
127                "trace_id".to_string(),
128                serde_json::Value::String(self.trace_id.clone()),
129            );
130        }
131        if fields.method {
132            map.insert(
133                "method".to_string(),
134                serde_json::Value::String(self.method.clone()),
135            );
136        }
137        if fields.path {
138            map.insert(
139                "path".to_string(),
140                serde_json::Value::String(self.path.clone()),
141            );
142        }
143        if fields.query {
144            if let Some(ref q) = self.query {
145                map.insert("query".to_string(), serde_json::Value::String(q.clone()));
146            }
147        }
148        if fields.status {
149            map.insert("status".to_string(), serde_json::json!(self.status));
150        }
151        if fields.latency_ms {
152            map.insert(
153                "duration_ms".to_string(),
154                serde_json::json!(self.duration_ms),
155            );
156        }
157        if fields.client_ip {
158            map.insert(
159                "client_ip".to_string(),
160                serde_json::Value::String(self.client_ip.clone()),
161            );
162        }
163        if fields.user_agent {
164            if let Some(ref ua) = self.user_agent {
165                map.insert(
166                    "user_agent".to_string(),
167                    serde_json::Value::String(ua.clone()),
168                );
169            }
170        }
171        if fields.referer {
172            if let Some(ref r) = self.referer {
173                map.insert("referer".to_string(), serde_json::Value::String(r.clone()));
174            }
175        }
176        if fields.body_bytes_sent {
177            map.insert(
178                "body_bytes_sent".to_string(),
179                serde_json::json!(self.body_bytes_sent),
180            );
181        }
182        if fields.upstream_addr {
183            if let Some(ref addr) = self.upstream_addr {
184                map.insert(
185                    "upstream_addr".to_string(),
186                    serde_json::Value::String(addr.clone()),
187                );
188            }
189        }
190        if fields.connection_reused {
191            map.insert(
192                "connection_reused".to_string(),
193                serde_json::json!(self.connection_reused),
194            );
195        }
196        if fields.rate_limit_hit {
197            map.insert(
198                "rate_limit_hit".to_string(),
199                serde_json::json!(self.rate_limit_hit),
200            );
201        }
202        if fields.geo_country {
203            if let Some(ref cc) = self.geo_country {
204                map.insert(
205                    "geo_country".to_string(),
206                    serde_json::Value::String(cc.clone()),
207                );
208            }
209        }
210        // Always include these core fields (not configurable)
211        if let Some(ref route) = self.route_id {
212            map.insert(
213                "route_id".to_string(),
214                serde_json::Value::String(route.clone()),
215            );
216        }
217        if let Some(ref upstream) = self.upstream {
218            map.insert(
219                "upstream".to_string(),
220                serde_json::Value::String(upstream.clone()),
221            );
222        }
223        map.insert(
224            "instance_id".to_string(),
225            serde_json::Value::String(self.instance_id.clone()),
226        );
227
228        serde_json::to_string(&map).unwrap_or_else(|_| "{}".to_string())
229    }
230
231    /// Format as Combined Log Format with trace_id extension
232    /// Format: client_ip - - [timestamp] "method path?query protocol" status bytes "referer" "user_agent" trace_id duration_ms
233    fn format_combined(&self) -> String {
234        // Parse RFC3339 timestamp to CLF format [day/month/year:hour:min:sec zone]
235        let clf_timestamp = self.format_clf_timestamp();
236
237        // Build request line
238        let request_line = if let Some(ref query) = self.query {
239            format!("{} {}?{} {}", self.method, self.path, query, self.protocol)
240        } else {
241            format!("{} {} {}", self.method, self.path, self.protocol)
242        };
243
244        // Escape and format optional fields
245        let referer = self.referer.as_deref().unwrap_or("-");
246        let user_agent = self.user_agent.as_deref().unwrap_or("-");
247
248        // Combined format with trace_id and duration extensions
249        format!(
250            "{} - - [{}] \"{}\" {} {} \"{}\" \"{}\" {} {}ms",
251            self.client_ip,
252            clf_timestamp,
253            request_line,
254            self.status,
255            self.body_bytes,
256            referer,
257            user_agent,
258            self.trace_id,
259            self.duration_ms
260        )
261    }
262
263    /// Convert RFC3339 timestamp to Common Log Format timestamp
264    fn format_clf_timestamp(&self) -> String {
265        // Try to parse and reformat, fallback to original if parsing fails
266        if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(&self.timestamp) {
267            dt.format("%d/%b/%Y:%H:%M:%S %z").to_string()
268        } else {
269            self.timestamp.clone()
270        }
271    }
272}
273
274/// Error log entry
275#[derive(Debug, Serialize)]
276pub struct ErrorLogEntry {
277    /// Timestamp in RFC3339 format
278    pub timestamp: String,
279    /// Trace ID for correlation
280    pub trace_id: String,
281    /// Log level (warn, error)
282    pub level: String,
283    /// Error message
284    pub message: String,
285    /// Route ID if available
286    #[serde(skip_serializing_if = "Option::is_none")]
287    pub route_id: Option<String>,
288    /// Upstream if available
289    #[serde(skip_serializing_if = "Option::is_none")]
290    pub upstream: Option<String>,
291    /// Error details/context
292    #[serde(skip_serializing_if = "Option::is_none")]
293    pub details: Option<String>,
294}
295
296/// Audit event type for categorizing security events
297#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
298#[serde(rename_all = "snake_case")]
299pub enum AuditEventType {
300    /// Request blocked by policy
301    Blocked,
302    /// Agent made a decision
303    AgentDecision,
304    /// WAF rule matched
305    WafMatch,
306    /// WAF blocked request
307    WafBlock,
308    /// Rate limit exceeded
309    RateLimitExceeded,
310    /// Authentication event
311    AuthEvent,
312    /// Configuration change
313    ConfigChange,
314    /// Certificate reload
315    CertReload,
316    /// Circuit breaker state change
317    CircuitBreakerChange,
318    /// Cache purge request
319    CachePurge,
320    /// Admin action
321    AdminAction,
322    /// Custom event
323    Custom,
324}
325
326impl std::fmt::Display for AuditEventType {
327    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
328        match self {
329            AuditEventType::Blocked => write!(f, "blocked"),
330            AuditEventType::AgentDecision => write!(f, "agent_decision"),
331            AuditEventType::WafMatch => write!(f, "waf_match"),
332            AuditEventType::WafBlock => write!(f, "waf_block"),
333            AuditEventType::RateLimitExceeded => write!(f, "rate_limit_exceeded"),
334            AuditEventType::AuthEvent => write!(f, "auth_event"),
335            AuditEventType::ConfigChange => write!(f, "config_change"),
336            AuditEventType::CertReload => write!(f, "cert_reload"),
337            AuditEventType::CircuitBreakerChange => write!(f, "circuit_breaker_change"),
338            AuditEventType::CachePurge => write!(f, "cache_purge"),
339            AuditEventType::AdminAction => write!(f, "admin_action"),
340            AuditEventType::Custom => write!(f, "custom"),
341        }
342    }
343}
344
345/// Audit log entry for security events
346#[derive(Debug, Serialize)]
347pub struct AuditLogEntry {
348    /// Timestamp in RFC3339 format
349    pub timestamp: String,
350    /// Trace ID for correlation
351    pub trace_id: String,
352    /// Event type (blocked, agent_decision, waf_match, etc.)
353    pub event_type: String,
354    /// HTTP method
355    pub method: String,
356    /// Request path
357    pub path: String,
358    /// Client IP
359    pub client_ip: String,
360    /// Route ID
361    #[serde(skip_serializing_if = "Option::is_none")]
362    pub route_id: Option<String>,
363    /// Block reason if blocked
364    #[serde(skip_serializing_if = "Option::is_none")]
365    pub reason: Option<String>,
366    /// Agent that made the decision
367    #[serde(skip_serializing_if = "Option::is_none")]
368    pub agent_id: Option<String>,
369    /// WAF rule IDs matched
370    #[serde(skip_serializing_if = "Vec::is_empty")]
371    pub rule_ids: Vec<String>,
372    /// Additional tags
373    #[serde(skip_serializing_if = "Vec::is_empty")]
374    pub tags: Vec<String>,
375    /// Action taken (allow, block, challenge, redirect)
376    #[serde(skip_serializing_if = "Option::is_none")]
377    pub action: Option<String>,
378    /// Response status code
379    #[serde(skip_serializing_if = "Option::is_none")]
380    pub status_code: Option<u16>,
381    /// User ID if authenticated
382    #[serde(skip_serializing_if = "Option::is_none")]
383    pub user_id: Option<String>,
384    /// Session ID if available
385    #[serde(skip_serializing_if = "Option::is_none")]
386    pub session_id: Option<String>,
387    /// Additional metadata as key-value pairs
388    #[serde(skip_serializing_if = "std::collections::HashMap::is_empty")]
389    pub metadata: std::collections::HashMap<String, String>,
390    /// Namespace (for scoped requests)
391    #[serde(skip_serializing_if = "Option::is_none")]
392    pub namespace: Option<String>,
393    /// Service (for scoped requests)
394    #[serde(skip_serializing_if = "Option::is_none")]
395    pub service: Option<String>,
396}
397
398impl AuditLogEntry {
399    /// Create a new audit log entry with required fields
400    pub fn new(
401        trace_id: impl Into<String>,
402        event_type: AuditEventType,
403        method: impl Into<String>,
404        path: impl Into<String>,
405        client_ip: impl Into<String>,
406    ) -> Self {
407        Self {
408            timestamp: chrono::Utc::now().to_rfc3339(),
409            trace_id: trace_id.into(),
410            event_type: event_type.to_string(),
411            method: method.into(),
412            path: path.into(),
413            client_ip: client_ip.into(),
414            route_id: None,
415            reason: None,
416            agent_id: None,
417            rule_ids: Vec::new(),
418            tags: Vec::new(),
419            action: None,
420            status_code: None,
421            user_id: None,
422            session_id: None,
423            metadata: std::collections::HashMap::new(),
424            namespace: None,
425            service: None,
426        }
427    }
428
429    /// Builder: set namespace
430    pub fn with_namespace(mut self, namespace: impl Into<String>) -> Self {
431        self.namespace = Some(namespace.into());
432        self
433    }
434
435    /// Builder: set service
436    pub fn with_service(mut self, service: impl Into<String>) -> Self {
437        self.service = Some(service.into());
438        self
439    }
440
441    /// Builder: set scope from namespace and service
442    pub fn with_scope(mut self, namespace: Option<String>, service: Option<String>) -> Self {
443        self.namespace = namespace;
444        self.service = service;
445        self
446    }
447
448    /// Builder: set route ID
449    pub fn with_route_id(mut self, route_id: impl Into<String>) -> Self {
450        self.route_id = Some(route_id.into());
451        self
452    }
453
454    /// Builder: set reason
455    pub fn with_reason(mut self, reason: impl Into<String>) -> Self {
456        self.reason = Some(reason.into());
457        self
458    }
459
460    /// Builder: set agent ID
461    pub fn with_agent_id(mut self, agent_id: impl Into<String>) -> Self {
462        self.agent_id = Some(agent_id.into());
463        self
464    }
465
466    /// Builder: add rule IDs
467    pub fn with_rule_ids(mut self, rule_ids: Vec<String>) -> Self {
468        self.rule_ids = rule_ids;
469        self
470    }
471
472    /// Builder: add tags
473    pub fn with_tags(mut self, tags: Vec<String>) -> Self {
474        self.tags = tags;
475        self
476    }
477
478    /// Builder: set action
479    pub fn with_action(mut self, action: impl Into<String>) -> Self {
480        self.action = Some(action.into());
481        self
482    }
483
484    /// Builder: set status code
485    pub fn with_status_code(mut self, status_code: u16) -> Self {
486        self.status_code = Some(status_code);
487        self
488    }
489
490    /// Builder: set user ID
491    pub fn with_user_id(mut self, user_id: impl Into<String>) -> Self {
492        self.user_id = Some(user_id.into());
493        self
494    }
495
496    /// Builder: set session ID
497    pub fn with_session_id(mut self, session_id: impl Into<String>) -> Self {
498        self.session_id = Some(session_id.into());
499        self
500    }
501
502    /// Builder: add metadata
503    pub fn with_metadata(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
504        self.metadata.insert(key.into(), value.into());
505        self
506    }
507
508    /// Create an entry for a blocked request
509    pub fn blocked(
510        trace_id: impl Into<String>,
511        method: impl Into<String>,
512        path: impl Into<String>,
513        client_ip: impl Into<String>,
514        reason: impl Into<String>,
515    ) -> Self {
516        Self::new(trace_id, AuditEventType::Blocked, method, path, client_ip)
517            .with_reason(reason)
518            .with_action("block")
519    }
520
521    /// Create an entry for rate limit exceeded
522    pub fn rate_limited(
523        trace_id: impl Into<String>,
524        method: impl Into<String>,
525        path: impl Into<String>,
526        client_ip: impl Into<String>,
527        limit_key: impl Into<String>,
528    ) -> Self {
529        Self::new(
530            trace_id,
531            AuditEventType::RateLimitExceeded,
532            method,
533            path,
534            client_ip,
535        )
536        .with_reason("Rate limit exceeded")
537        .with_action("block")
538        .with_metadata("limit_key", limit_key)
539    }
540
541    /// Create an entry for WAF block
542    pub fn waf_blocked(
543        trace_id: impl Into<String>,
544        method: impl Into<String>,
545        path: impl Into<String>,
546        client_ip: impl Into<String>,
547        rule_ids: Vec<String>,
548    ) -> Self {
549        Self::new(trace_id, AuditEventType::WafBlock, method, path, client_ip)
550            .with_rule_ids(rule_ids)
551            .with_action("block")
552    }
553
554    /// Create an entry for configuration change
555    pub fn config_change(
556        trace_id: impl Into<String>,
557        change_type: impl Into<String>,
558        details: impl Into<String>,
559    ) -> Self {
560        Self::new(
561            trace_id,
562            AuditEventType::ConfigChange,
563            "-",
564            "/-/config",
565            "internal",
566        )
567        .with_reason(change_type)
568        .with_metadata("details", details)
569    }
570
571    /// Create an entry for certificate reload
572    pub fn cert_reload(
573        trace_id: impl Into<String>,
574        listener_id: impl Into<String>,
575        success: bool,
576    ) -> Self {
577        Self::new(
578            trace_id,
579            AuditEventType::CertReload,
580            "-",
581            "/-/certs",
582            "internal",
583        )
584        .with_metadata("listener_id", listener_id)
585        .with_metadata("success", success.to_string())
586    }
587
588    /// Create an entry for cache purge
589    pub fn cache_purge(
590        trace_id: impl Into<String>,
591        method: impl Into<String>,
592        path: impl Into<String>,
593        client_ip: impl Into<String>,
594        pattern: impl Into<String>,
595    ) -> Self {
596        Self::new(
597            trace_id,
598            AuditEventType::CachePurge,
599            method,
600            path,
601            client_ip,
602        )
603        .with_metadata("pattern", pattern)
604        .with_action("purge")
605    }
606
607    /// Create an entry for admin action
608    pub fn admin_action(
609        trace_id: impl Into<String>,
610        method: impl Into<String>,
611        path: impl Into<String>,
612        client_ip: impl Into<String>,
613        action: impl Into<String>,
614    ) -> Self {
615        Self::new(
616            trace_id,
617            AuditEventType::AdminAction,
618            method,
619            path,
620            client_ip,
621        )
622        .with_action(action)
623    }
624}
625
626/// Buffered file writer for log files
627struct LogFileWriter {
628    writer: BufWriter<File>,
629}
630
631impl LogFileWriter {
632    fn new(path: &Path, buffer_size: usize) -> Result<Self> {
633        // Create parent directories if they don't exist
634        if let Some(parent) = path.parent() {
635            std::fs::create_dir_all(parent)
636                .with_context(|| format!("Failed to create log directory: {:?}", parent))?;
637        }
638
639        let file = OpenOptions::new()
640            .create(true)
641            .append(true)
642            .open(path)
643            .with_context(|| format!("Failed to open log file: {:?}", path))?;
644
645        Ok(Self {
646            writer: BufWriter::with_capacity(buffer_size, file),
647        })
648    }
649
650    fn write_line(&mut self, line: &str) -> Result<()> {
651        writeln!(self.writer, "{}", line)?;
652        Ok(())
653    }
654
655    fn flush(&mut self) -> Result<()> {
656        self.writer.flush()?;
657        Ok(())
658    }
659}
660
661/// Log manager handling all log file writers
662pub struct LogManager {
663    access_log: Option<Mutex<LogFileWriter>>,
664    access_log_format: AccessLogFormat,
665    access_log_config: Option<grapsus_config::AccessLogConfig>,
666    error_log: Option<Mutex<LogFileWriter>>,
667    error_log_level: String,
668    audit_log: Option<Mutex<LogFileWriter>>,
669    audit_config: Option<AuditLogConfig>,
670}
671
672impl LogManager {
673    /// Create a new log manager from configuration
674    pub fn new(config: &LoggingConfig) -> Result<Self> {
675        let (access_log, access_log_format) = if let Some(ref access_config) = config.access_log {
676            if access_config.enabled {
677                let format = Self::parse_access_format(&access_config.format);
678                let writer = Mutex::new(LogFileWriter::new(
679                    &access_config.file,
680                    access_config.buffer_size,
681                )?);
682                (Some(writer), format)
683            } else {
684                (None, AccessLogFormat::Json)
685            }
686        } else {
687            (None, AccessLogFormat::Json)
688        };
689
690        let (error_log, error_log_level) = if let Some(ref error_config) = config.error_log {
691            if error_config.enabled {
692                (
693                    Some(Mutex::new(LogFileWriter::new(
694                        &error_config.file,
695                        error_config.buffer_size,
696                    )?)),
697                    error_config.level.clone(),
698                )
699            } else {
700                (None, "warn".to_string())
701            }
702        } else {
703            (None, "warn".to_string())
704        };
705
706        let audit_log = if let Some(ref audit_config) = config.audit_log {
707            if audit_config.enabled {
708                Some(Mutex::new(LogFileWriter::new(
709                    &audit_config.file,
710                    audit_config.buffer_size,
711                )?))
712            } else {
713                None
714            }
715        } else {
716            None
717        };
718
719        Ok(Self {
720            access_log,
721            access_log_format,
722            access_log_config: config.access_log.clone(),
723            error_log,
724            error_log_level,
725            audit_log,
726            audit_config: config.audit_log.clone(),
727        })
728    }
729
730    /// Create a disabled log manager (no file logging)
731    pub fn disabled() -> Self {
732        Self {
733            access_log: None,
734            access_log_format: AccessLogFormat::Json,
735            access_log_config: None,
736            error_log: None,
737            error_log_level: "warn".to_string(),
738            audit_log: None,
739            audit_config: None,
740        }
741    }
742
743    /// Parse access log format from config string
744    fn parse_access_format(format: &str) -> AccessLogFormat {
745        match format.to_lowercase().as_str() {
746            "combined" | "clf" | "common" => AccessLogFormat::Combined,
747            _ => AccessLogFormat::Json, // Default to JSON
748        }
749    }
750
751    /// Check whether an access log entry with the given status should be logged,
752    /// accounting for sampling rate. Call this before constructing `AccessLogEntry`
753    /// to avoid allocating fields that would be discarded.
754    pub fn should_log_access(&self, status: u16) -> bool {
755        if self.access_log.is_none() {
756            return false;
757        }
758        if let Some(ref config) = self.access_log_config {
759            if config.sample_errors_always && status >= 400 {
760                return true;
761            }
762            use rand::RngExt;
763            let mut rng = rand::rng();
764            return rng.random::<f64>() < config.sample_rate;
765        }
766        true
767    }
768
769    /// Write an access log entry. Callers should check `should_log_access` first
770    /// to avoid building the entry when it would be discarded by sampling.
771    pub fn log_access(&self, entry: &AccessLogEntry) {
772        if let Some(ref writer) = self.access_log {
773            let fields = self.access_log_config.as_ref().map(|c| &c.fields);
774            let formatted = entry.format(self.access_log_format, fields);
775            let mut guard = writer.lock();
776            if let Err(e) = guard.write_line(&formatted) {
777                error!("Failed to write access log: {}", e);
778            }
779        }
780    }
781
782    /// Write an error log entry
783    pub fn log_error(&self, entry: &ErrorLogEntry) {
784        if let Some(ref writer) = self.error_log {
785            match serde_json::to_string(entry) {
786                Ok(json) => {
787                    let mut guard = writer.lock();
788                    if let Err(e) = guard.write_line(&json) {
789                        error!("Failed to write error log: {}", e);
790                    }
791                }
792                Err(e) => {
793                    error!("Failed to serialize error log entry: {}", e);
794                }
795            }
796        }
797    }
798
799    /// Write an audit log entry
800    pub fn log_audit(&self, entry: &AuditLogEntry) {
801        if let Some(ref writer) = self.audit_log {
802            if let Some(ref config) = self.audit_config {
803                // Check if we should log this event type
804                let should_log = match entry.event_type.as_str() {
805                    "blocked" => config.log_blocked,
806                    "agent_decision" => config.log_agent_decisions,
807                    "waf_match" | "waf_block" => config.log_waf_events,
808                    _ => true, // Log other event types by default
809                };
810
811                if !should_log {
812                    return;
813                }
814            }
815
816            match serde_json::to_string(entry) {
817                Ok(json) => {
818                    let mut guard = writer.lock();
819                    if let Err(e) = guard.write_line(&json) {
820                        error!("Failed to write audit log: {}", e);
821                    }
822                }
823                Err(e) => {
824                    error!("Failed to serialize audit log entry: {}", e);
825                }
826            }
827        }
828    }
829
830    /// Flush all log buffers
831    pub fn flush(&self) {
832        if let Some(ref writer) = self.access_log {
833            if let Err(e) = writer.lock().flush() {
834                warn!("Failed to flush access log: {}", e);
835            }
836        }
837        if let Some(ref writer) = self.error_log {
838            if let Err(e) = writer.lock().flush() {
839                warn!("Failed to flush error log: {}", e);
840            }
841        }
842        if let Some(ref writer) = self.audit_log {
843            if let Err(e) = writer.lock().flush() {
844                warn!("Failed to flush audit log: {}", e);
845            }
846        }
847    }
848
849    /// Check if access logging is enabled
850    pub fn access_log_enabled(&self) -> bool {
851        self.access_log.is_some()
852    }
853
854    /// Check if error logging is enabled
855    pub fn error_log_enabled(&self) -> bool {
856        self.error_log.is_some()
857    }
858
859    /// Check if audit logging is enabled
860    pub fn audit_log_enabled(&self) -> bool {
861        self.audit_log.is_some()
862    }
863
864    /// Check if the given level meets the configured minimum error log level.
865    /// Level hierarchy: "warn" logs both warn and error, "error" logs only error.
866    fn should_log_error_level(&self, level: &str) -> bool {
867        match self.error_log_level.as_str() {
868            "error" => level == "error",
869            _ => level == "warn" || level == "error", // "warn" (default) logs both
870        }
871    }
872
873    /// Convenience method to construct an ErrorLogEntry and write it to the error log.
874    /// Checks both that error logging is enabled and that the level meets the configured minimum.
875    pub fn log_request_error(
876        &self,
877        level: &str,
878        message: &str,
879        trace_id: &str,
880        route_id: Option<&str>,
881        upstream: Option<&str>,
882        details: Option<String>,
883    ) {
884        if self.error_log.is_none() || !self.should_log_error_level(level) {
885            return;
886        }
887        let entry = ErrorLogEntry {
888            timestamp: chrono::Utc::now().to_rfc3339(),
889            trace_id: trace_id.to_string(),
890            level: level.to_string(),
891            message: message.to_string(),
892            route_id: route_id.map(|s| s.to_string()),
893            upstream: upstream.map(|s| s.to_string()),
894            details,
895        };
896        self.log_error(&entry);
897    }
898}
899
900impl Drop for LogManager {
901    fn drop(&mut self) {
902        self.flush();
903    }
904}
905
906/// Shared log manager that can be passed around
907pub type SharedLogManager = Arc<LogManager>;
908
909#[cfg(test)]
910mod tests {
911    use super::*;
912    use tempfile::tempdir;
913    use grapsus_config::{AccessLogConfig, ErrorLogConfig};
914
915    #[test]
916    fn test_access_log_entry_serialization() {
917        let entry = AccessLogEntry {
918            timestamp: "2024-01-01T00:00:00Z".to_string(),
919            trace_id: "abc123".to_string(),
920            method: "GET".to_string(),
921            path: "/api/users".to_string(),
922            query: Some("page=1".to_string()),
923            protocol: "HTTP/1.1".to_string(),
924            status: 200,
925            body_bytes: 1024,
926            duration_ms: 50,
927            client_ip: "192.168.1.1".to_string(),
928            user_agent: Some("Mozilla/5.0".to_string()),
929            referer: None,
930            host: Some("example.com".to_string()),
931            route_id: Some("api-route".to_string()),
932            upstream: Some("backend-1".to_string()),
933            upstream_attempts: 1,
934            instance_id: "instance-1".to_string(),
935            namespace: None,
936            service: None,
937            body_bytes_sent: 2048,
938            upstream_addr: Some("10.0.1.5:8080".to_string()),
939            connection_reused: true,
940            rate_limit_hit: false,
941            geo_country: None,
942        };
943
944        let json = serde_json::to_string(&entry).unwrap();
945        assert!(json.contains("\"trace_id\":\"abc123\""));
946        assert!(json.contains("\"status\":200"));
947    }
948
949    #[test]
950    fn test_access_log_entry_with_scope() {
951        let entry = AccessLogEntry {
952            timestamp: "2024-01-01T00:00:00Z".to_string(),
953            trace_id: "abc123".to_string(),
954            method: "GET".to_string(),
955            path: "/api/users".to_string(),
956            query: None,
957            protocol: "HTTP/1.1".to_string(),
958            status: 200,
959            body_bytes: 1024,
960            duration_ms: 50,
961            client_ip: "192.168.1.1".to_string(),
962            user_agent: None,
963            referer: None,
964            host: None,
965            route_id: Some("api-route".to_string()),
966            upstream: Some("backend-1".to_string()),
967            upstream_attempts: 1,
968            instance_id: "instance-1".to_string(),
969            namespace: Some("api".to_string()),
970            service: Some("payments".to_string()),
971            body_bytes_sent: 2048,
972            upstream_addr: None,
973            connection_reused: false,
974            rate_limit_hit: false,
975            geo_country: Some("US".to_string()),
976        };
977
978        let json = serde_json::to_string(&entry).unwrap();
979        assert!(json.contains("\"namespace\":\"api\""));
980        assert!(json.contains("\"service\":\"payments\""));
981    }
982
983    #[test]
984    fn test_log_manager_creation() {
985        let dir = tempdir().unwrap();
986        let access_log_path = dir.path().join("access.log");
987        let error_log_path = dir.path().join("error.log");
988        let audit_log_path = dir.path().join("audit.log");
989
990        let config = LoggingConfig {
991            level: "info".to_string(),
992            format: "json".to_string(),
993            timestamps: true,
994            file: None,
995            access_log: Some(AccessLogConfig {
996                enabled: true,
997                file: access_log_path.clone(),
998                format: "json".to_string(),
999                buffer_size: 8192,
1000                include_trace_id: true,
1001                sample_rate: 1.0,
1002                sample_errors_always: true,
1003                fields: grapsus_config::AccessLogFields::default(),
1004            }),
1005            error_log: Some(ErrorLogConfig {
1006                enabled: true,
1007                file: error_log_path.clone(),
1008                level: "warn".to_string(),
1009                buffer_size: 8192,
1010            }),
1011            audit_log: Some(AuditLogConfig {
1012                enabled: true,
1013                file: audit_log_path.clone(),
1014                buffer_size: 8192,
1015                log_blocked: true,
1016                log_agent_decisions: true,
1017                log_waf_events: true,
1018            }),
1019        };
1020
1021        let manager = LogManager::new(&config).unwrap();
1022        assert!(manager.access_log_enabled());
1023        assert!(manager.error_log_enabled());
1024        assert!(manager.audit_log_enabled());
1025    }
1026
1027    #[test]
1028    fn test_access_log_combined_format() {
1029        let entry = AccessLogEntry {
1030            timestamp: "2024-01-15T10:30:00+00:00".to_string(),
1031            trace_id: "trace-abc123".to_string(),
1032            method: "GET".to_string(),
1033            path: "/api/users".to_string(),
1034            query: Some("page=1".to_string()),
1035            protocol: "HTTP/1.1".to_string(),
1036            status: 200,
1037            body_bytes: 1024,
1038            duration_ms: 50,
1039            client_ip: "192.168.1.1".to_string(),
1040            user_agent: Some("Mozilla/5.0".to_string()),
1041            referer: Some("https://example.com/".to_string()),
1042            host: Some("api.example.com".to_string()),
1043            route_id: Some("api-route".to_string()),
1044            upstream: Some("backend-1".to_string()),
1045            upstream_attempts: 1,
1046            instance_id: "instance-1".to_string(),
1047            namespace: None,
1048            service: None,
1049            body_bytes_sent: 2048,
1050            upstream_addr: Some("10.0.1.5:8080".to_string()),
1051            connection_reused: true,
1052            rate_limit_hit: false,
1053            geo_country: Some("US".to_string()),
1054        };
1055
1056        let combined = entry.format(AccessLogFormat::Combined, None);
1057
1058        // Check Combined format structure
1059        assert!(combined.starts_with("192.168.1.1 - - ["));
1060        assert!(combined.contains("\"GET /api/users?page=1 HTTP/1.1\""));
1061        assert!(combined.contains(" 200 1024 "));
1062        assert!(combined.contains("\"https://example.com/\""));
1063        assert!(combined.contains("\"Mozilla/5.0\""));
1064        assert!(combined.contains("trace-abc123"));
1065        assert!(combined.ends_with("50ms"));
1066    }
1067
1068    #[test]
1069    fn test_access_log_format_parsing() {
1070        assert_eq!(
1071            LogManager::parse_access_format("json"),
1072            AccessLogFormat::Json
1073        );
1074        assert_eq!(
1075            LogManager::parse_access_format("JSON"),
1076            AccessLogFormat::Json
1077        );
1078        assert_eq!(
1079            LogManager::parse_access_format("combined"),
1080            AccessLogFormat::Combined
1081        );
1082        assert_eq!(
1083            LogManager::parse_access_format("COMBINED"),
1084            AccessLogFormat::Combined
1085        );
1086        assert_eq!(
1087            LogManager::parse_access_format("clf"),
1088            AccessLogFormat::Combined
1089        );
1090        assert_eq!(
1091            LogManager::parse_access_format("unknown"),
1092            AccessLogFormat::Json
1093        ); // Default to JSON
1094    }
1095}
grapsus_proxy/logging.rs

grapsus_proxy/
logging.rs
