Coverage for security / audit_log.py: 41.9%

1""" 

2Security Audit Logging 

3Filters sensitive data from log output and provides secure logging. 

4Prevents credential leakage via log files. 

5""" 

6 

7import re 

8import logging 

9from typing import List, Tuple 

10 

11# Patterns to redact from logs 

12_REDACTION_PATTERNS: List[Tuple[re.Pattern, str]] = [ 

13 # OpenAI API keys 

14 (re.compile(r'sk-[a-zA-Z0-9]{20,}'), '[REDACTED_OPENAI_KEY]'), 

15 # JWT tokens 

16 (re.compile(r'eyJ[a-zA-Z0-9_-]{10,}\.eyJ[a-zA-Z0-9_-]{10,}\.[a-zA-Z0-9_-]{10,}'), 

17 '[REDACTED_JWT]'), 

18 # Google API keys 

19 (re.compile(r'AIzaSy[a-zA-Z0-9_-]{33}'), '[REDACTED_GOOGLE_KEY]'), 

20 # Groq API keys 

21 (re.compile(r'gsk_[a-zA-Z0-9]{20,}'), '[REDACTED_GROQ_KEY]'), 

22 # Generic Bearer tokens in logs 

23 (re.compile(r'Bearer\s+[a-zA-Z0-9_.-]{20,}'), 'Bearer [REDACTED]'), 

24 # Password values in key=value format 

25 (re.compile(r'(password|passwd|pwd|secret|token|api_key|apikey)\s*[=:]\s*\S+', re.I), 

26 r'\1=[REDACTED]'), 

27 # AWS keys 

28 (re.compile(r'AKIA[0-9A-Z]{16}'), '[REDACTED_AWS_KEY]'), 

29 # Generic hex tokens (40+ chars) 

30 (re.compile(r'\b[0-9a-f]{40,}\b'), '[REDACTED_HEX_TOKEN]'), 

31] 

32 

33 

34class SensitiveFilter(logging.Filter): 

35 """ 

36 Logging filter that redacts sensitive data patterns. 

37 Attach to any logger handler to prevent credential leakage. 

38 """ 

39 

40 def filter(self, record: logging.LogRecord) -> bool: 

41 if isinstance(record.msg, str): 

42 record.msg = self._redact(record.msg) 

43 if record.args: 

44 if isinstance(record.args, dict): 

45 record.args = { 

46 k: self._redact(str(v)) if isinstance(v, str) else v 

47 for k, v in record.args.items() 

48 } 

49 elif isinstance(record.args, tuple): 

50 record.args = tuple( 

51 self._redact(str(a)) if isinstance(a, str) else a 

52 for a in record.args 

53 ) 

54 return True 

55 

56 @staticmethod 

57 def _redact(text: str) -> str: 

58 for pattern, replacement in _REDACTION_PATTERNS: 

59 text = pattern.sub(replacement, text) 

60 return text 

61 

62 

63def get_secure_logger(name: str, level: int = logging.INFO) -> logging.Logger: 

64 """ 

65 Get a logger with the SensitiveFilter already attached. 

66 Use this instead of logging.getLogger() for security-critical modules. 

67 """ 

68 logger = logging.getLogger(name) 

69 logger.setLevel(level) 

70 

71 # Add filter if not already present 

72 if not any(isinstance(f, SensitiveFilter) for f in logger.filters): 

73 logger.addFilter(SensitiveFilter()) 

74 

75 return logger 

76 

77 

78def apply_sensitive_filter_to_all(): 

79 """ 

80 Apply SensitiveFilter to the root logger so all log output is redacted. 

81 Call this once at application startup. 

82 """ 

83 root_logger = logging.getLogger() 

84 if not any(isinstance(f, SensitiveFilter) for f in root_logger.filters): 

85 root_logger.addFilter(SensitiveFilter()) 

86 

87 # Also add to all existing handlers 

88 for handler in root_logger.handlers: 

89 if not any(isinstance(f, SensitiveFilter) for f in handler.filters): 

90 handler.addFilter(SensitiveFilter())