Coverage for src / qdrant_loader_core / logging_processors.py: 79%

1"""Logging processors and formatters for structlog.""" 

2 

3from __future__ import annotations 

4 

5import logging 

6import re 

7from typing import Any 

8 

9 

10class CleanFormatter(logging.Formatter): 

11 """Formatter that removes ANSI color codes for clean file output.""" 

12 

13 def format(self, record: logging.LogRecord) -> str: 

14 message = super().format(record) 

15 try: 

16 ansi_escape = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])") 

17 return ansi_escape.sub("", message) 

18 except Exception: 

19 return message 

20 

21 

22def redact_processor( 

23 logger: Any, method_name: str, event_dict: dict[str, Any] 

24) -> dict[str, Any]: 

25 """Structlog processor to redact sensitive fields in event_dict.""" 

26 sensitive_keys = { 

27 "api_key", 

28 "llm_api_key", 

29 "authorization", 

30 "Authorization", 

31 "token", 

32 "access_token", 

33 "secret", 

34 "password", 

35 } 

36 sensitive_key_pattern = re.compile( 

37 r"(?:^|[_-]|(?<=[a-z]))(?i:token|secret|password|api[_-]?key|access[_-]?key|private[_-]?key|authorization)(?:$|[_-]|(?=[A-Z]))" 

38 ) 

39 

40 def is_sensitive_key(key: Any) -> bool: 

41 if not isinstance(key, str): 

42 return False 

43 return key in sensitive_keys or bool(sensitive_key_pattern.search(key)) 

44 

45 def mask(value: str) -> str: 

46 try: 

47 if not isinstance(value, str) or not value: 

48 return "***REDACTED***" 

49 if len(value) <= 8: 

50 return "***REDACTED***" 

51 return value[:2] + "***REDACTED***" + value[-2:] 

52 except Exception: 

53 return "***REDACTED***" 

54 

55 def deep_redact(obj: Any) -> Any: 

56 try: 

57 if isinstance(obj, dict): 

58 return { 

59 k: ( 

60 mask(v) 

61 if is_sensitive_key(k) and isinstance(v, str) 

62 else deep_redact(v) 

63 ) 

64 for k, v in obj.items() 

65 } 

66 if isinstance(obj, list): 

67 return [deep_redact(i) for i in obj] 

68 return obj 

69 except Exception: 

70 return obj 

71 

72 return deep_redact(event_dict)