|
jpayne@7
|
1 from __future__ import annotations
|
|
jpayne@7
|
2
|
|
jpayne@7
|
3 import http.client as httplib
|
|
jpayne@7
|
4 from email.errors import MultipartInvariantViolationDefect, StartBoundaryNotFoundDefect
|
|
jpayne@7
|
5
|
|
jpayne@7
|
6 from ..exceptions import HeaderParsingError
|
|
jpayne@7
|
7
|
|
jpayne@7
|
8
|
|
jpayne@7
|
9 def is_fp_closed(obj: object) -> bool:
|
|
jpayne@7
|
10 """
|
|
jpayne@7
|
11 Checks whether a given file-like object is closed.
|
|
jpayne@7
|
12
|
|
jpayne@7
|
13 :param obj:
|
|
jpayne@7
|
14 The file-like object to check.
|
|
jpayne@7
|
15 """
|
|
jpayne@7
|
16
|
|
jpayne@7
|
17 try:
|
|
jpayne@7
|
18 # Check `isclosed()` first, in case Python3 doesn't set `closed`.
|
|
jpayne@7
|
19 # GH Issue #928
|
|
jpayne@7
|
20 return obj.isclosed() # type: ignore[no-any-return, attr-defined]
|
|
jpayne@7
|
21 except AttributeError:
|
|
jpayne@7
|
22 pass
|
|
jpayne@7
|
23
|
|
jpayne@7
|
24 try:
|
|
jpayne@7
|
25 # Check via the official file-like-object way.
|
|
jpayne@7
|
26 return obj.closed # type: ignore[no-any-return, attr-defined]
|
|
jpayne@7
|
27 except AttributeError:
|
|
jpayne@7
|
28 pass
|
|
jpayne@7
|
29
|
|
jpayne@7
|
30 try:
|
|
jpayne@7
|
31 # Check if the object is a container for another file-like object that
|
|
jpayne@7
|
32 # gets released on exhaustion (e.g. HTTPResponse).
|
|
jpayne@7
|
33 return obj.fp is None # type: ignore[attr-defined]
|
|
jpayne@7
|
34 except AttributeError:
|
|
jpayne@7
|
35 pass
|
|
jpayne@7
|
36
|
|
jpayne@7
|
37 raise ValueError("Unable to determine whether fp is closed.")
|
|
jpayne@7
|
38
|
|
jpayne@7
|
39
|
|
jpayne@7
|
40 def assert_header_parsing(headers: httplib.HTTPMessage) -> None:
|
|
jpayne@7
|
41 """
|
|
jpayne@7
|
42 Asserts whether all headers have been successfully parsed.
|
|
jpayne@7
|
43 Extracts encountered errors from the result of parsing headers.
|
|
jpayne@7
|
44
|
|
jpayne@7
|
45 Only works on Python 3.
|
|
jpayne@7
|
46
|
|
jpayne@7
|
47 :param http.client.HTTPMessage headers: Headers to verify.
|
|
jpayne@7
|
48
|
|
jpayne@7
|
49 :raises urllib3.exceptions.HeaderParsingError:
|
|
jpayne@7
|
50 If parsing errors are found.
|
|
jpayne@7
|
51 """
|
|
jpayne@7
|
52
|
|
jpayne@7
|
53 # This will fail silently if we pass in the wrong kind of parameter.
|
|
jpayne@7
|
54 # To make debugging easier add an explicit check.
|
|
jpayne@7
|
55 if not isinstance(headers, httplib.HTTPMessage):
|
|
jpayne@7
|
56 raise TypeError(f"expected httplib.Message, got {type(headers)}.")
|
|
jpayne@7
|
57
|
|
jpayne@7
|
58 unparsed_data = None
|
|
jpayne@7
|
59
|
|
jpayne@7
|
60 # get_payload is actually email.message.Message.get_payload;
|
|
jpayne@7
|
61 # we're only interested in the result if it's not a multipart message
|
|
jpayne@7
|
62 if not headers.is_multipart():
|
|
jpayne@7
|
63 payload = headers.get_payload()
|
|
jpayne@7
|
64
|
|
jpayne@7
|
65 if isinstance(payload, (bytes, str)):
|
|
jpayne@7
|
66 unparsed_data = payload
|
|
jpayne@7
|
67
|
|
jpayne@7
|
68 # httplib is assuming a response body is available
|
|
jpayne@7
|
69 # when parsing headers even when httplib only sends
|
|
jpayne@7
|
70 # header data to parse_headers() This results in
|
|
jpayne@7
|
71 # defects on multipart responses in particular.
|
|
jpayne@7
|
72 # See: https://github.com/urllib3/urllib3/issues/800
|
|
jpayne@7
|
73
|
|
jpayne@7
|
74 # So we ignore the following defects:
|
|
jpayne@7
|
75 # - StartBoundaryNotFoundDefect:
|
|
jpayne@7
|
76 # The claimed start boundary was never found.
|
|
jpayne@7
|
77 # - MultipartInvariantViolationDefect:
|
|
jpayne@7
|
78 # A message claimed to be a multipart but no subparts were found.
|
|
jpayne@7
|
79 defects = [
|
|
jpayne@7
|
80 defect
|
|
jpayne@7
|
81 for defect in headers.defects
|
|
jpayne@7
|
82 if not isinstance(
|
|
jpayne@7
|
83 defect, (StartBoundaryNotFoundDefect, MultipartInvariantViolationDefect)
|
|
jpayne@7
|
84 )
|
|
jpayne@7
|
85 ]
|
|
jpayne@7
|
86
|
|
jpayne@7
|
87 if defects or unparsed_data:
|
|
jpayne@7
|
88 raise HeaderParsingError(defects=defects, unparsed_data=unparsed_data)
|
|
jpayne@7
|
89
|
|
jpayne@7
|
90
|
|
jpayne@7
|
91 def is_response_to_head(response: httplib.HTTPResponse) -> bool:
|
|
jpayne@7
|
92 """
|
|
jpayne@7
|
93 Checks whether the request of a response has been a HEAD-request.
|
|
jpayne@7
|
94
|
|
jpayne@7
|
95 :param http.client.HTTPResponse response:
|
|
jpayne@7
|
96 Response to check if the originating request
|
|
jpayne@7
|
97 used 'HEAD' as a method.
|
|
jpayne@7
|
98 """
|
|
jpayne@7
|
99 # FIXME: Can we do this somehow without accessing private httplib _method?
|
|
jpayne@7
|
100 method_str = response._method # type: str # type: ignore[attr-defined]
|
|
jpayne@7
|
101 return method_str.upper() == "HEAD"
|
