Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/urllib3/util/request.py: 34%

1from __future__ import annotations 

2 

3import io 

4import sys 

5import typing 

6from base64 import b64encode 

7from enum import Enum 

8 

9from ..exceptions import UnrewindableBodyError 

10from .util import to_bytes 

11 

12if typing.TYPE_CHECKING: 

13 from typing import Final 

14 

15# Pass as a value within ``headers`` to skip 

16# emitting some HTTP headers that are added automatically. 

17# The only headers that are supported are ``Accept-Encoding``, 

18# ``Host``, and ``User-Agent``. 

19SKIP_HEADER = "@@@SKIP_HEADER@@@" 

20SKIPPABLE_HEADERS = frozenset(["accept-encoding", "host", "user-agent"]) 

21 

22ACCEPT_ENCODING = "gzip,deflate" 

23try: 

24 try: 

25 import brotlicffi as _unused_module_brotli # type: ignore[import-not-found] # noqa: F401 

26 except ImportError: 

27 import brotli as _unused_module_brotli # type: ignore[import-not-found] # noqa: F401 

28except ImportError: 

29 pass 

30else: 

31 ACCEPT_ENCODING += ",br" 

32 

33try: 

34 if sys.version_info >= (3, 14): 

35 from compression import zstd as _unused_module_zstd # noqa: F401 

36 else: 

37 from backports import zstd as _unused_module_zstd # noqa: F401 

38except ImportError: 

39 pass 

40else: 

41 ACCEPT_ENCODING += ",zstd" 

42 

43 

44class _TYPE_FAILEDTELL(Enum): 

45 token = 0 

46 

47 

48_FAILEDTELL: Final[_TYPE_FAILEDTELL] = _TYPE_FAILEDTELL.token 

49 

50_TYPE_BODY_POSITION = typing.Union[int, _TYPE_FAILEDTELL] 

51 

52# When sending a request with these methods we aren't expecting 

53# a body so don't need to set an explicit 'Content-Length: 0' 

54# The reason we do this in the negative instead of tracking methods 

55# which 'should' have a body is because unknown methods should be 

56# treated as if they were 'POST' which *does* expect a body. 

57_METHODS_NOT_EXPECTING_BODY = {"GET", "HEAD", "DELETE", "TRACE", "OPTIONS", "CONNECT"} 

58 

59 

60def make_headers( 

61 keep_alive: bool | None = None, 

62 accept_encoding: bool | list[str] | str | None = None, 

63 user_agent: str | None = None, 

64 basic_auth: str | None = None, 

65 proxy_basic_auth: str | None = None, 

66 disable_cache: bool | None = None, 

67 *, 

68 basic_auth_encoding: str = "latin-1", 

69 proxy_basic_auth_encoding: str = "latin-1", 

70) -> dict[str, str]: 

71 """ 

72 Shortcuts for generating request headers. 

73 

74 :param keep_alive: 

75 If ``True``, adds 'connection: keep-alive' header. 

76 

77 :param accept_encoding: 

78 Can be a boolean, list, or string. 

79 ``True`` translates to 'gzip,deflate'. If the dependencies for 

80 Brotli (either the ``brotli`` or ``brotlicffi`` package) and/or 

81 Zstandard (the ``backports.zstd`` package for Python before 3.14) 

82 algorithms are installed, then their encodings are 

83 included in the string ('br' and 'zstd', respectively). 

84 List will get joined by comma. 

85 String will be used as provided. 

86 

87 :param user_agent: 

88 String representing the user-agent you want, such as 

89 "python-urllib3/0.6" 

90 

91 :param basic_auth: 

92 Colon-separated username:password string for 'authorization: basic ...' 

93 auth header. Basic authentication transmits Base64-encoded bytes, not 

94 Unicode text, so the string is encoded with ``basic_auth_encoding`` 

95 before Base64 encoding. 

96 

97 Must contain decoded credentials (not percent-encoded). 

98 For URLs parsed with :func:`parse_url`, use :attr:`Url.auth_decoded_joined` 

99 to get the properly formatted string. 

100 

101 :param basic_auth_encoding: 

102 Text encoding used to encode ``basic_auth`` before Base64 encoding. 

103 Defaults to "latin-1" for backward compatibility. Use "utf-8" only 

104 when the peer expects UTF-8 Basic authentication credentials. 

105 

106 Must contain decoded credentials (not percent-encoded). 

107 For URLs parsed with :func:`parse_url`, use :attr:`Url.auth_decoded_joined` 

108 to get the properly formatted string. 

109 

110 :param proxy_basic_auth: 

111 Colon-separated username:password string for 'proxy-authorization: basic ...' 

112 auth header. Basic authentication transmits Base64-encoded bytes, not 

113 Unicode text, so the string is encoded with ``proxy_basic_auth_encoding`` 

114 before Base64 encoding. 

115 

116 Must contain decoded credentials (not percent-encoded). 

117 For URLs parsed with :func:`parse_url`, use :attr:`Url.auth_decoded_joined` 

118 to get the properly formatted string. 

119 

120 :param proxy_basic_auth_encoding: 

121 Text encoding used to encode ``proxy_basic_auth`` before Base64 

122 encoding. Defaults to "latin-1" for backward compatibility. Use 

123 "utf-8" only when the proxy expects UTF-8 Basic authentication 

124 credentials. 

125 

126 Must contain decoded credentials (not percent-encoded). 

127 For URLs parsed with :func:`parse_url`, use :attr:`Url.auth_decoded_joined` 

128 to get the properly formatted string. 

129 

130 :param disable_cache: 

131 If ``True``, adds 'cache-control: no-cache' header. 

132 

133 Example: 

134 

135 .. code-block:: python 

136 

137 import urllib3 

138 

139 print(urllib3.util.make_headers(keep_alive=True, user_agent="Batman/1.0")) 

140 # {'connection': 'keep-alive', 'user-agent': 'Batman/1.0'} 

141 print(urllib3.util.make_headers(accept_encoding=True)) 

142 # {'accept-encoding': 'gzip,deflate'} 

143 """ 

144 headers: dict[str, str] = {} 

145 if accept_encoding: 

146 if isinstance(accept_encoding, str): 

147 pass 

148 elif isinstance(accept_encoding, list): 

149 accept_encoding = ",".join(accept_encoding) 

150 else: 

151 accept_encoding = ACCEPT_ENCODING 

152 headers["accept-encoding"] = accept_encoding 

153 

154 if user_agent: 

155 headers["user-agent"] = user_agent 

156 

157 if keep_alive: 

158 headers["connection"] = "keep-alive" 

159 

160 if basic_auth: 

161 headers["authorization"] = ( 

162 f"Basic {b64encode(basic_auth.encode(basic_auth_encoding)).decode()}" 

163 ) 

164 

165 if proxy_basic_auth: 

166 headers["proxy-authorization"] = ( 

167 f"Basic {b64encode(proxy_basic_auth.encode(proxy_basic_auth_encoding)).decode()}" 

168 ) 

169 

170 if disable_cache: 

171 headers["cache-control"] = "no-cache" 

172 

173 return headers 

174 

175 

176def set_file_position( 

177 body: typing.Any, pos: _TYPE_BODY_POSITION | None 

178) -> _TYPE_BODY_POSITION | None: 

179 """ 

180 If a position is provided, move file to that point. 

181 Otherwise, we'll attempt to record a position for future use. 

182 """ 

183 if pos is not None: 

184 rewind_body(body, pos) 

185 elif getattr(body, "tell", None) is not None: 

186 try: 

187 pos = body.tell() 

188 except OSError: 

189 # This differentiates from None, allowing us to catch 

190 # a failed `tell()` later when trying to rewind the body. 

191 pos = _FAILEDTELL 

192 

193 return pos 

194 

195 

196def rewind_body(body: typing.IO[typing.AnyStr], body_pos: _TYPE_BODY_POSITION) -> None: 

197 """ 

198 Attempt to rewind body to a certain position. 

199 Primarily used for request redirects and retries. 

200 

201 :param body: 

202 File-like object that supports seek. 

203 

204 :param int pos: 

205 Position to seek to in file. 

206 """ 

207 body_seek = getattr(body, "seek", None) 

208 if body_seek is not None and isinstance(body_pos, int): 

209 try: 

210 body_seek(body_pos) 

211 except OSError as e: 

212 raise UnrewindableBodyError( 

213 "An error occurred when rewinding request body for redirect/retry." 

214 ) from e 

215 elif body_pos is _FAILEDTELL: 

216 raise UnrewindableBodyError( 

217 "Unable to record file position for rewinding " 

218 "request body during a redirect/retry." 

219 ) 

220 else: 

221 raise ValueError( 

222 f"body_pos must be of type integer, instead it was {type(body_pos)}." 

223 ) 

224 

225 

226class ChunksAndContentLength(typing.NamedTuple): 

227 chunks: typing.Iterable[bytes] | None 

228 content_length: int | None 

229 

230 

231def body_to_chunks( 

232 body: typing.Any | None, method: str, blocksize: int 

233) -> ChunksAndContentLength: 

234 """Takes the HTTP request method, body, and blocksize and 

235 transforms them into an iterable of chunks to pass to 

236 socket.sendall() and an optional 'Content-Length' header. 

237 

238 A 'Content-Length' of 'None' indicates the length of the body 

239 can't be determined so should use 'Transfer-Encoding: chunked' 

240 for framing instead. 

241 """ 

242 

243 chunks: typing.Iterable[bytes] | None 

244 content_length: int | None 

245 

246 # No body, we need to make a recommendation on 'Content-Length' 

247 # based on whether that request method is expected to have 

248 # a body or not. 

249 if body is None: 

250 chunks = None 

251 if method.upper() not in _METHODS_NOT_EXPECTING_BODY: 

252 content_length = 0 

253 else: 

254 content_length = None 

255 

256 # Bytes or strings become bytes 

257 elif isinstance(body, (str, bytes)): 

258 chunks = (to_bytes(body),) 

259 content_length = len(chunks[0]) 

260 

261 # File-like object, TODO: use seek() and tell() for length? 

262 elif hasattr(body, "read"): 

263 

264 def chunk_readable() -> typing.Iterable[bytes]: 

265 encode = isinstance(body, io.TextIOBase) 

266 while True: 

267 datablock = body.read(blocksize) 

268 if not datablock: 

269 break 

270 if encode: 

271 datablock = datablock.encode("utf-8") 

272 yield datablock 

273 

274 chunks = chunk_readable() 

275 content_length = None 

276 

277 # Otherwise we need to start checking via duck-typing. 

278 else: 

279 try: 

280 # Check if the body implements the buffer API. 

281 mv = memoryview(body) 

282 except TypeError: 

283 try: 

284 # Check if the body is an iterable 

285 chunks = iter(body) 

286 content_length = None 

287 except TypeError: 

288 raise TypeError( 

289 f"'body' must be a bytes-like object, file-like " 

290 f"object, or iterable. Instead was {body!r}" 

291 ) from None 

292 else: 

293 # Since it implements the buffer API can be passed directly to socket.sendall() 

294 chunks = (body,) 

295 content_length = mv.nbytes 

296 

297 return ChunksAndContentLength(chunks=chunks, content_length=content_length)