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

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 elif body_seek is None: 

221 raise UnrewindableBodyError("body does not implement seek.") 

222 else: 

223 raise ValueError( 

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

225 ) 

226 

227 

228class ChunksAndContentLength(typing.NamedTuple): 

229 chunks: typing.Iterable[bytes] | None 

230 content_length: int | None 

231 

232 

233def body_to_chunks( 

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

235) -> ChunksAndContentLength: 

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

237 transforms them into an iterable of chunks to pass to 

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

239 

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

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

242 for framing instead. 

243 """ 

244 

245 chunks: typing.Iterable[bytes] | None 

246 content_length: int | None 

247 

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

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

250 # a body or not. 

251 if body is None: 

252 chunks = None 

253 if method.upper() not in _METHODS_NOT_EXPECTING_BODY: 

254 content_length = 0 

255 else: 

256 content_length = None 

257 

258 # Bytes or strings become bytes 

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

260 chunks = (to_bytes(body),) 

261 content_length = len(chunks[0]) 

262 

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

264 elif hasattr(body, "read"): 

265 

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

267 encode = isinstance(body, io.TextIOBase) 

268 while True: 

269 datablock = body.read(blocksize) 

270 if not datablock: 

271 break 

272 if encode: 

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

274 yield datablock 

275 

276 chunks = chunk_readable() 

277 content_length = None 

278 

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

280 else: 

281 try: 

282 # Check if the body implements the buffer API. 

283 mv = memoryview(body) 

284 except TypeError: 

285 try: 

286 # Check if the body is an iterable 

287 chunks = iter(body) 

288 content_length = None 

289 except TypeError: 

290 raise TypeError( 

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

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

293 ) from None 

294 else: 

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

296 chunks = (body,) 

297 content_length = mv.nbytes 

298 

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