Coverage for src/qdrant_loader_core/logging_filters.py: 78%

1"""Logging filters for redaction and noise suppression.""" 

2 

3from __future__ import annotations 

4 

5import logging 

6import re 

7from urllib.parse import unquote_plus 

8 

9 

10class QdrantVersionFilter(logging.Filter): 

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

12 try: 

13 return "version check" not in record.getMessage().lower() 

14 except Exception: 

15 return True 

16 

17 

18class ApplicationFilter(logging.Filter): 

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

20 # Allow all logs by default; app packages may add their own filters 

21 return True 

22 

23 

24class UvicornAccessRedactFilter(logging.Filter): 

25 """Redacts secret/token query-string values from uvicorn's access log line. 

26 

27 uvicorn logs the raw request line (including the query string) via 

28 ``record.args`` on the "uvicorn.access" logger, e.g. 

29 ``args = (client_addr, method, "/path?token=abc123", http_version, status)``. 

30 That logger has ``propagate=False`` in uvicorn's default logging config, so 

31 it never reaches the root logger's handlers/filters (including 

32 :class:`RedactionFilter`) — the secret would otherwise be written to the 

33 access log in plaintext. Attach this filter directly to the 

34 "uvicorn.access" logger; per-logger filters run in ``Logger.handle()`` 

35 before any handler, so this applies regardless of what handlers uvicorn's 

36 own dictConfig installs (dictConfig only replaces handlers, not filters). 

37 """ 

38 

39 _SENSITIVE_QUERY_KEYS = { 

40 "token", 

41 "secret", 

42 "signature", 

43 "password", 

44 "authorization", 

45 "api_key", 

46 "api-key", 

47 "access_key", 

48 "access-key", 

49 "private_key", 

50 "private-key", 

51 "access_token", 

52 "access-token", 

53 } 

54 _QUERY_PAIR = re.compile(r'([?&])([^=&\s"]+)=([^&\s"]*)') 

55 

56 @classmethod 

57 def _redact_query_pair(cls, match: re.Match[str]) -> str: 

58 sep, raw_key, _value = match.groups() 

59 # uvicorn logs the raw, still percent-encoded request line, so a key 

60 # like "sec%72et" would slip past a literal match while Starlette/ 

61 # FastAPI (which percent-decodes query keys during parsing) still 

62 # resolves it to "secret" and accepts it as the webhook secret. 

63 # Decode before comparing so encoded keys are caught too. 

64 if unquote_plus(raw_key).lower() in cls._SENSITIVE_QUERY_KEYS: 

65 return f"{sep}{raw_key}=***REDACTED***" 

66 return match.group(0) 

67 

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

69 try: 

70 if isinstance(record.args, tuple) and len(record.args) >= 3: 

71 path = record.args[2] 

72 if isinstance(path, str) and "?" in path: 

73 redacted = self._QUERY_PAIR.sub(self._redact_query_pair, path) 

74 if redacted != path: 

75 record.args = ( 

76 record.args[0], 

77 record.args[1], 

78 redacted, 

79 *record.args[3:], 

80 ) 

81 except Exception: 

82 pass 

83 return True 

84 

85 

86class RedactionFilter(logging.Filter): 

87 """Redacts obvious secrets from stdlib log records.""" 

88 

89 # Heuristics for tokens/keys in plain strings 

90 TOKEN_PATTERNS = [ 

91 re.compile(r"sk-[A-Za-z0-9_\-]{6,}"), 

92 re.compile(r"tok-[A-Za-z0-9_\-]{6,}"), 

93 re.compile( 

94 r"(?i)(api_key|authorization|token|access_token|secret|password)\s*[:=]\s*([^\s]+)" 

95 ), 

96 re.compile(r"Bearer\s+[A-Za-z0-9_\-\.]+"), 

97 ] 

98 

99 # Keys commonly used for secrets in structlog event dictionaries 

100 SENSITIVE_KEYS = { 

101 "api_key", 

102 "llm_api_key", 

103 "authorization", 

104 "Authorization", 

105 "token", 

106 "access_token", 

107 "secret", 

108 "password", 

109 } 

110 

111 def _redact_text(self, text: str) -> str: 

112 def mask(m: re.Match[str]) -> str: 

113 s = m.group(0) 

114 if len(s) <= 8: 

115 return "***REDACTED***" 

116 return s[:2] + "***REDACTED***" + s[-2:] 

117 

118 redacted = text 

119 for pat in self.TOKEN_PATTERNS: 

120 redacted = pat.sub(mask, redacted) 

121 return redacted 

122 

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

124 try: 

125 redaction_detected = False 

126 

127 # Args may contain secrets; best-effort mask strings and detect changes 

128 if isinstance(record.args, tuple): 

129 new_args = [] 

130 for a in record.args: 

131 if isinstance(a, str): 

132 red_a = self._redact_text(a) 

133 if red_a != a: 

134 redaction_detected = True 

135 new_args.append(red_a) 

136 else: 

137 new_args.append(a) 

138 record.args = tuple(new_args) 

139 

140 # Redact raw message only when it contains no formatting placeholders 

141 # to avoid interfering with %-style or {}-style formatting 

142 if isinstance(record.msg, str): 

143 try: 

144 has_placeholders = ("%" in record.msg) or ("{" in record.msg) 

145 except Exception: 

146 has_placeholders = True 

147 if not has_placeholders: 

148 red_msg = self._redact_text(record.msg) 

149 if red_msg != record.msg: 

150 record.msg = red_msg 

151 redaction_detected = True 

152 

153 # If structlog extras contain sensitive keys, mark as redacted 

154 try: 

155 if any( 

156 (k in self.SENSITIVE_KEYS and bool(record.__dict__.get(k))) 

157 for k in record.__dict__.keys() 

158 ): 

159 redaction_detected = True 

160 except Exception: 

161 pass 

162 

163 # Ensure a visible redaction marker appears in the captured message 

164 if redaction_detected: 

165 try: 

166 if ( 

167 isinstance(record.msg, str) 

168 and "***REDACTED***" not in record.msg 

169 ): 

170 # Append a marker in a way that won't interfere with %-formatting 

171 record.msg = f"{record.msg} ***REDACTED***" 

172 except Exception: 

173 pass 

174 except Exception: 

175 pass 

176 return True