Coverage for dataexcept/redaction.py: 100%

1"""Redaction helpers for values that must not reach a log. 

2 

3Several exceptions here are raised with credentials in hand: an authentication 

4token, a database URL carrying a password, a webhook URL whose *path* is the 

5secret. Those values end up in the exception message, and 

6:func:`dataexcept.logging_helpers.log_exception` logs ``str(exc)``, so without 

7redaction a failed delivery writes the credential to the log. 

8 

9The aim is to keep an error debuggable while giving up the secret. A redacted 

10value carries a short, non-reversible fingerprint, so repeated failures of the 

11*same* credential stay recognisable in a log without the credential appearing 

12in it. 

13 

14What this can and cannot do is stated in ``SECURITY.md``. In short: values the 

15library is *given* as credentials are redacted, and URLs are redacted wherever 

16they appear -- including inside a message you supplied and inside the text of a 

17wrapped exception. A bare, non-URL secret pasted into free-form text cannot be 

18recognised and is not redacted. 

19""" 

20 

21from __future__ import annotations 

22 

23import hashlib 

24import re 

25from typing import Optional 

26from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit 

27 

28__all__ = [ 

29 "fingerprint", 

30 "redact_if_url", 

31 "redact_secret", 

32 "redact_url", 

33 "redact_urls_in_text", 

34 "remove_secret", 

35] 

36 

37PLACEHOLDER = "***" 

38 

39#: Below this length a "secret" is not removed from free text. Substring 

40#: replacement of a short value corrupts ordinary words -- removing "tok" from 

41#: "Invalid authentication token" mangles the message and tells a reader 

42#: nothing. The structured field is redacted regardless of length. 

43MIN_REMOVABLE_SECRET_LENGTH = 8 

44 

45#: Parameter-name tokens that mark a value as a secret. A name is split into 

46#: tokens on separators and camelCase boundaries, and matched token by token. 

47#: 

48#: Substring matching was tried first and was wrong in both directions: it 

49#: redacted "monkey", "design", "assign", "keyword" and "authors" -- mangling 

50#: ordinary debugging information -- while still missing "passphrase". The 

51#: point of keeping host, port and path is that the error stays actionable, and 

52#: shredding a legitimate query parameter works against that. 

53#: 

54#: Deliberately absent: "code", "state", "nonce" and "client_id". An OAuth 

55#: authorization code is a secret, but "code" is far more often a country 

56#: code, an HTTP status or a discount code, and redacting those would destroy 

57#: more debugging information than it protects. Checked against the parameter 

58#: names used by AWS SigV4, Azure SAS, Google Cloud and OAuth 2. 

59SENSITIVE_PARAM_TOKENS = frozenset( 

60 { 

61 "apikey", 

62 "auth", 

63 "authorization", 

64 "bearer", 

65 "credential", 

66 "credentials", 

67 "hmac", 

68 "jwt", 

69 "key", 

70 "keys", 

71 "passphrase", 

72 "passwd", 

73 "password", 

74 "pwd", 

75 "sas", 

76 "secret", 

77 "secrets", 

78 "session", 

79 "sig", 

80 "signature", 

81 "token", 

82 "tokens", 

83 } 

84) 

85 

86#: Splits a parameter name into words: on separators, and between a lower-case 

87#: or digit character and an upper-case one, so ``accessToken`` yields 

88#: ``["access", "token"]``. 

89_NAME_TOKENS = re.compile(r"[A-Za-z0-9]+") 

90_CAMEL_BOUNDARY = re.compile(r"(?<=[a-z0-9])(?=[A-Z])") 

91 

92 

93def _tokens(name: str) -> list[str]: 

94 words: list[str] = [] 

95 for chunk in _NAME_TOKENS.findall(name): 

96 words.extend(part.lower() for part in _CAMEL_BOUNDARY.split(chunk) if part) 

97 return words 

98 

99 

100def _is_sensitive(name: str) -> bool: 

101 return any(token in SENSITIVE_PARAM_TOKENS for token in _tokens(name)) 

102 

103 

104#: Finds URLs inside free-form text, so a credential cannot slip through in a 

105#: caller-supplied message or in the text of a wrapped exception. 

106# No word-boundary anchor: a URL can directly follow a word character, as 

107# in the step name "feature_https://..." that FeaturePreprocessingError 

108# builds. The scheme character class excludes "_", so a match still starts 

109# at the scheme rather than mid-word. 

110_URL_IN_TEXT = re.compile(r"[a-zA-Z][a-zA-Z0-9+.\-]*://[^\s'\"<>,;)\]}]+") 

111 

112 

113def fingerprint(value: str) -> str: 

114 """Return a short, one-way fingerprint of *value*. 

115 

116 Enough to tell "the same bad token again" from "a different bad token", 

117 and not enough to recover the token. 

118 """ 

119 digest = hashlib.sha256(value.encode("utf-8", "replace")).hexdigest() 

120 return digest[:8] 

121 

122 

123def redact_secret(value: Optional[str]) -> Optional[str]: 

124 """Replace a secret with a placeholder and its fingerprint.""" 

125 if value is None: 

126 return None 

127 if not value: 

128 return PLACEHOLDER 

129 return f"{PLACEHOLDER}({fingerprint(value)})" 

130 

131 

132def remove_secret(text: str, secret: Optional[str]) -> str: 

133 """Replace every occurrence of a known *secret* in *text*. 

134 

135 Used where the library was handed the secret explicitly, so it can be 

136 removed even from a message the caller wrote themselves. 

137 """ 

138 if not secret or not text or len(secret) < MIN_REMOVABLE_SECRET_LENGTH: 

139 return text 

140 return text.replace(secret, f"{PLACEHOLDER}({fingerprint(secret)})") 

141 

142 

143def _redact_params(query: str) -> tuple[str, bool]: 

144 if not query: 

145 return query, False 

146 pairs = parse_qsl(query, keep_blank_values=True) 

147 if not any(_is_sensitive(key) for key, _ in pairs): 

148 return query, False 

149 return ( 

150 urlencode( 

151 [ 

152 (key, PLACEHOLDER if _is_sensitive(key) else value) 

153 for key, value in pairs 

154 ], 

155 # Keep the placeholder legible rather than percent-encoded. 

156 safe="*", 

157 ), 

158 True, 

159 ) 

160 

161 

162def redact_url(url: Optional[str], *, keep_path: bool = True) -> Optional[str]: 

163 """Strip credentials from *url*. 

164 

165 Scheme, host and port are always kept: those are what make an error 

166 actionable. Userinfo, sensitive query parameters and sensitive fragment 

167 parameters are always removed. 

168 

169 Pass ``keep_path=False`` where the path itself is the credential. An 

170 incoming webhook URL is the common case -- Slack, Discord and others put 

171 the secret in the path, so preserving it would defeat the point. 

172 """ 

173 if not url or not isinstance(url, str): 

174 # Anything that is not a string is handed back untouched. urlsplit 

175 # would raise AttributeError from inside urllib, masking whatever the 

176 # caller's real mistake was with a message about `.decode`. 

177 return url 

178 

179 try: 

180 parts = urlsplit(url) 

181 except ValueError: # pragma: no cover - urlsplit is extremely permissive 

182 return PLACEHOLDER 

183 

184 if not parts.scheme or not parts.netloc: 

185 # Not a URL with a host; a bare path or plain string is returned 

186 # untouched rather than mangled. 

187 return url 

188 

189 redacted = False 

190 

191 netloc = parts.netloc 

192 if "@" in netloc: 

193 _, _, host = netloc.rpartition("@") 

194 netloc = f"{PLACEHOLDER}:{PLACEHOLDER}@{host}" 

195 redacted = True 

196 

197 query, query_redacted = _redact_params(parts.query) 

198 redacted = redacted or query_redacted 

199 

200 fragment = parts.fragment 

201 if "=" in fragment: 

202 # OAuth implicit flow returns the token in the fragment. 

203 fragment, fragment_redacted = _redact_params(fragment) 

204 redacted = redacted or fragment_redacted 

205 

206 path = parts.path 

207 if not keep_path and path.strip("/"): 

208 path = f"/{PLACEHOLDER}" 

209 redacted = True 

210 

211 if not redacted: 

212 # Hand back exactly what was passed in. Rebuilding would normalise it, 

213 # and "sqlite://" loses its slashes on the way through. 

214 return url 

215 

216 return urlunsplit((parts.scheme, netloc, path, query, fragment)) 

217 

218 

219def redact_if_url(value: Optional[str], *, keep_path: bool = True) -> Optional[str]: 

220 """Redact *value* only if it is a URL, leaving file paths untouched. 

221 

222 Fields such as ``DataLoadingError.source`` document themselves as "file 

223 path or URL", so they cannot be redacted unconditionally without mangling 

224 ordinary paths. 

225 """ 

226 if not isinstance(value, str) or "://" not in value: 

227 return value 

228 return redact_url(value, keep_path=keep_path) 

229 

230 

231def redact_urls_in_text(text: str, *, keep_path: bool = True) -> str: 

232 """Redact every URL found in free-form *text*. 

233 

234 This is the boundary that stops a secret being reintroduced after the 

235 structured argument was redacted -- through a caller-supplied ``message``, 

236 or through the text of a wrapped exception that quotes the original URL. 

237 """ 

238 if not text or "://" not in text: 

239 return text 

240 return _URL_IN_TEXT.sub( 

241 lambda match: redact_url(match.group(0), keep_path=keep_path) or "", 

242 text, 

243 )