Coverage for /usr/local/lib/python3.10/site-packages/opal_common-0.0.0-py3.10.egg/opal_common/http_utils.py: 62%

1import re 

2from typing import Tuple, Union 

3from urllib.parse import urlsplit, urlunsplit 

4 

5import aiohttp 

6import httpx 

7 

8#: Query-string parameter names whose values may carry credentials. 

9SENSITIVE_QUERY_PARAMS = frozenset( 

10 { 

11 "token", 

12 "access_token", 

13 "api_key", 

14 "apikey", 

15 "key", 

16 "password", 

17 "secret", 

18 "sig", 

19 "signature", 

20 } 

21) 

22 

23#: Matches ``scheme://userinfo@`` anywhere in free text. 

24_USERINFO_RE = re.compile(r"(?P<scheme>[a-zA-Z][a-zA-Z0-9+.\-]*://)[^/@\s]+@") 

25 

26 

27def _mask_sensitive_params(component: str) -> Tuple[str, bool]: 

28 """Mask values of known sensitive params in a raw query/fragment component. 

29 

30 Operates directly on the raw ``key=value&...`` text (rather than 

31 ``parse_qsl`` + ``urlencode``) so the encoding of untouched params is 

32 preserved byte-for-byte - only the sensitive values are replaced with 

33 ``***``. Returns the (possibly rewritten) component and whether anything 

34 changed. 

35 """ 

36 changed = False 

37 out = [] 

38 for pair in component.split("&"): 

39 key, sep, _ = pair.partition("=") 

40 if sep and key.lower() in SENSITIVE_QUERY_PARAMS: 40 ↛ 41line 40 didn't jump to line 41 because the condition on line 40 was never true

41 out.append(f"{key}=***") 

42 changed = True 

43 else: 

44 out.append(pair) 

45 return "&".join(out), changed 

46 

47 

48def redact_url(url: str) -> str: 

49 """Strip embedded credentials from a URL so it is safe to log. 

50 

51 Data source / policy repo URLs may be of the form 

52 ``https://user:token@host/path`` or carry a credential in a query parameter 

53 (e.g. ``?token=...``). We replace any ``user:password@`` userinfo with 

54 ``***@`` and mask the values of known sensitive query (and fragment) 

55 parameters, while keeping the host, port, path and non-sensitive params 

56 intact for debugging. Returns the input byte-for-byte unchanged if it is 

57 empty, cannot be parsed, or carries nothing sensitive. 

58 

59 Never raises: this helper is called from log/except paths, so any parsing 

60 error yields the input unchanged rather than propagating. 

61 """ 

62 if not url: 

63 return url 

64 try: 

65 parts = urlsplit(url) 

66 except ValueError: 

67 return url 

68 

69 changed = False 

70 netloc = parts.netloc 

71 try: 

72 if parts.username or parts.password: 72 ↛ 73line 72 didn't jump to line 73 because the condition on line 72 was never true

73 host = parts.hostname or "" 

74 if ":" in host: # IPv6 literal - urlsplit strips the surrounding brackets 

75 host = f"[{host}]" 

76 if parts.port is not None: 

77 host = f"{host}:{parts.port}" 

78 netloc = f"***@{host}" 

79 changed = True 

80 except ValueError: 

81 # ``urlsplit`` is lazy - accessing ``.username``/``.password``/``.port`` 

82 # is what actually validates and can raise (e.g. "Port out of range"). 

83 # We must never throw from a log path, so bail out unchanged. 

84 return url 

85 

86 query, query_changed = _mask_sensitive_params(parts.query) 

87 fragment, fragment_changed = _mask_sensitive_params(parts.fragment) 

88 changed = changed or query_changed or fragment_changed 

89 

90 if not changed: 90 ↛ 92line 90 didn't jump to line 92 because the condition on line 90 was always true

91 return url 

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

93 

94 

95def redact_url_in_text(text: str, url: str = "") -> str: 

96 """Redact embedded credentials from free text such as a git error message. 

97 

98 Replaces verbatim occurrences of a known ``url`` with its fully redacted 

99 form (so query-string tokens of that URL are masked too), then scrubs any 

100 remaining ``scheme://user:password@`` userinfo found anywhere in the text 

101 (a regex, so it is robust to the exact URL form git happens to print). 

102 

103 The known-URL replacement runs *first*: the regex rewrites 

104 ``user:pw@`` -> ``***@``, which would otherwise destroy the verbatim ``url`` 

105 before its query tokens could be masked. 

106 """ 

107 if not text: 107 ↛ 108line 107 didn't jump to line 108 because the condition on line 107 was never true

108 return text 

109 scrubbed = text 

110 if url: 110 ↛ 111line 110 didn't jump to line 111 because the condition on line 110 was never true

111 scrubbed = scrubbed.replace(url, redact_url(url)) 

112 scrubbed = _USERINFO_RE.sub(lambda m: f"{m.group('scheme')}***@", scrubbed) 

113 return scrubbed 

114 

115 

116def is_http_error_response( 

117 response: Union[aiohttp.ClientResponse, httpx.Response] 

118) -> bool: 

119 """HTTP 400 and above are considered error responses.""" 

120 status: int = ( 

121 response.status 

122 if isinstance(response, aiohttp.ClientResponse) 

123 else response.status_code 

124 ) 

125 

126 return status >= 400