Coverage for utils/log_sanitize.py: 100.00%

1import json 

2import re 

3from typing import Any 

4from urllib.parse import parse_qs 

5 

6REDACTED = "[REDACTED]" 

7ELLIPSIS = "..." 

8PREVIEW_CHARS = 8192 

9JWT_RE = re.compile(r"^[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+$") 

10SENSITIVE_TEXT_RE = re.compile( 

11 r"(?i)\b(password|passwd|pwd|secret|token|authorization|api[_-]?key)\b\s*[:=]\s*\S+" 

12) 

13SENSITIVE_KEY_MARKERS = ( 

14 "password", 

15 "passwd", 

16 "pwd", 

17 "secret", 

18 "token", 

19 "authorization", 

20 "cookie", 

21 "apikey", 

22 "setcookie", 

23 "privatekey", 

24 "csrf", 

25 "creditcard", 

26 "cardnumber", 

27 "cvv", 

28 "ssn", 

29) 

30SKIP_CONTENT_TYPES = ( 

31 "multipart/", 

32 "image/", 

33 "audio/", 

34 "video/", 

35 "octet-stream", 

36 "gzip", 

37 "zip", 

38 "text/event-stream", 

39) 

40 

41 

42def _normalize_key(key: str) -> str: 

43 return re.sub(r"[^a-z0-9]", "", key.lower()) 

44 

45 

46def is_sensitive_key(key: str) -> bool: 

47 normalized = _normalize_key(str(key)) 

48 return any(marker in normalized for marker in SENSITIVE_KEY_MARKERS) 

49 

50 

51def should_omit_body(content_type: str) -> bool: 

52 lowered = (content_type or "").lower() 

53 return any(kind in lowered for kind in SKIP_CONTENT_TYPES) 

54 

55 

56def quote_log_field(value: str) -> str: 

57 escaped = value.replace("\\", "\\\\").replace('"', '\\"') 

58 return f'"{escaped}"' 

59 

60 

61def format_log_value(value: str) -> str: 

62 """Wrap values so Grafana can extract original JSON, including spaces.""" 

63 compact = value.replace("\n", "\\n").replace("\r", "").replace(">>", "\\u003e\\u003e") 

64 return f"<<{compact}>>" 

65 

66 

67def _redact_string(value: str) -> str: 

68 stripped = value.strip() 

69 if stripped.lower().startswith("bearer ") or JWT_RE.match(stripped): 

70 return REDACTED 

71 return value 

72 

73 

74def redact(value: Any) -> Any: 

75 if isinstance(value, dict): 

76 return { 

77 key: REDACTED if is_sensitive_key(str(key)) else redact(item) 

78 for key, item in value.items() 

79 } 

80 if isinstance(value, list): 

81 return [redact(item) for item in value] 

82 if isinstance(value, str): 

83 return _redact_string(value) 

84 return value 

85 

86 

87def _redact_loose_text(text: str) -> str: 

88 return SENSITIVE_TEXT_RE.sub(lambda match: f"{match.group(1)}={REDACTED}", text) 

89 

90 

91def _decode(raw: bytes, max_bytes: int) -> tuple[str, bool]: 

92 truncated = len(raw) > max_bytes 

93 return raw[:max_bytes].decode("utf-8", errors="replace"), truncated 

94 

95 

96def _loads_json_prefix(text: str) -> Any | None: 

97 try: 

98 return json.loads(text) 

99 except json.JSONDecodeError: 

100 pass 

101 

102 in_string = False 

103 escape = False 

104 stack: list[str] = [] 

105 for char in text: 

106 if in_string: 

107 if escape: 

108 escape = False 

109 elif char == "\\": 

110 escape = True 

111 elif char == '"': 

112 in_string = False 

113 continue 

114 if char == '"': 

115 in_string = True 

116 elif char == "{": 

117 stack.append("}") 

118 elif char == "[": 

119 stack.append("]") 

120 elif stack and char == stack[-1]: 

121 stack.pop() 

122 

123 candidate = text.rstrip() 

124 if in_string: 

125 candidate += '"' 

126 candidate = re.sub(r",\s*$", "", candidate) 

127 candidate += "".join(reversed(stack)) 

128 try: 

129 return json.loads(candidate) 

130 except json.JSONDecodeError: 

131 return None 

132 

133 

134def _with_ellipsis(rendered: str, limit: int = PREVIEW_CHARS) -> str: 

135 if len(rendered) <= limit: 

136 return rendered + ELLIPSIS 

137 cutoff = max(16, limit - len(ELLIPSIS)) 

138 preview = rendered[:cutoff] 

139 for separator in ("},", "],", ",", "}", "]"): 

140 index = preview.rfind(separator) 

141 if index >= cutoff // 3: 

142 preview = preview[: index + len(separator)].rstrip(",") 

143 break 

144 return preview + ELLIPSIS 

145 

146 

147def sanitize_body(raw: bytes, content_type: str, max_bytes: int = 8192) -> str: 

148 if not raw: 

149 return "" 

150 if should_omit_body(content_type): 

151 return "[omitted]" 

152 

153 text, truncated = _decode(raw, max_bytes) 

154 lowered = (content_type or "").lower() 

155 payload: Any = None 

156 parsed = False 

157 

158 if "application/json" in lowered or text[:1] in "{[": 

159 parsed_json = _loads_json_prefix(text) 

160 if parsed_json is not None: 

161 payload = redact(parsed_json) 

162 parsed = True 

163 elif "application/x-www-form-urlencoded" in lowered: 

164 parsed_form = { 

165 key: values[-1] if values else "" 

166 for key, values in parse_qs(text, keep_blank_values=True).items() 

167 } 

168 payload = redact(parsed_form) 

169 parsed = True 

170 

171 if not parsed: 

172 preview = _redact_loose_text(text).replace("\n", "\\n") 

173 if truncated: 

174 return _with_ellipsis(preview) 

175 return preview 

176 

177 if isinstance(payload, (dict, list)): 

178 rendered = json.dumps(payload, ensure_ascii=False, separators=(",", ":")) 

179 else: 

180 rendered = str(payload).replace("\n", "\\n") 

181 

182 if truncated or len(rendered) > PREVIEW_CHARS: 

183 return _with_ellipsis(rendered) 

184 return rendered 

185 

186 

187def sanitize_query(query_params: Any) -> str: 

188 data = {str(key): value for key, value in dict(query_params).items()} 

189 if not data: 

190 return "" 

191 return json.dumps(redact(data), ensure_ascii=False, separators=(",", ":"))