jpayne@7: from __future__ import annotations
jpayne@7: 
jpayne@7: import http.client as httplib
jpayne@7: from email.errors import MultipartInvariantViolationDefect, StartBoundaryNotFoundDefect
jpayne@7: 
jpayne@7: from ..exceptions import HeaderParsingError
jpayne@7: 
jpayne@7: 
jpayne@7: def is_fp_closed(obj: object) -> bool:
jpayne@7:     """
jpayne@7:     Checks whether a given file-like object is closed.
jpayne@7: 
jpayne@7:     :param obj:
jpayne@7:         The file-like object to check.
jpayne@7:     """
jpayne@7: 
jpayne@7:     try:
jpayne@7:         # Check `isclosed()` first, in case Python3 doesn't set `closed`.
jpayne@7:         # GH Issue #928
jpayne@7:         return obj.isclosed()  # type: ignore[no-any-return, attr-defined]
jpayne@7:     except AttributeError:
jpayne@7:         pass
jpayne@7: 
jpayne@7:     try:
jpayne@7:         # Check via the official file-like-object way.
jpayne@7:         return obj.closed  # type: ignore[no-any-return, attr-defined]
jpayne@7:     except AttributeError:
jpayne@7:         pass
jpayne@7: 
jpayne@7:     try:
jpayne@7:         # Check if the object is a container for another file-like object that
jpayne@7:         # gets released on exhaustion (e.g. HTTPResponse).
jpayne@7:         return obj.fp is None  # type: ignore[attr-defined]
jpayne@7:     except AttributeError:
jpayne@7:         pass
jpayne@7: 
jpayne@7:     raise ValueError("Unable to determine whether fp is closed.")
jpayne@7: 
jpayne@7: 
jpayne@7: def assert_header_parsing(headers: httplib.HTTPMessage) -> None:
jpayne@7:     """
jpayne@7:     Asserts whether all headers have been successfully parsed.
jpayne@7:     Extracts encountered errors from the result of parsing headers.
jpayne@7: 
jpayne@7:     Only works on Python 3.
jpayne@7: 
jpayne@7:     :param http.client.HTTPMessage headers: Headers to verify.
jpayne@7: 
jpayne@7:     :raises urllib3.exceptions.HeaderParsingError:
jpayne@7:         If parsing errors are found.
jpayne@7:     """
jpayne@7: 
jpayne@7:     # This will fail silently if we pass in the wrong kind of parameter.
jpayne@7:     # To make debugging easier add an explicit check.
jpayne@7:     if not isinstance(headers, httplib.HTTPMessage):
jpayne@7:         raise TypeError(f"expected httplib.Message, got {type(headers)}.")
jpayne@7: 
jpayne@7:     unparsed_data = None
jpayne@7: 
jpayne@7:     # get_payload is actually email.message.Message.get_payload;
jpayne@7:     # we're only interested in the result if it's not a multipart message
jpayne@7:     if not headers.is_multipart():
jpayne@7:         payload = headers.get_payload()
jpayne@7: 
jpayne@7:         if isinstance(payload, (bytes, str)):
jpayne@7:             unparsed_data = payload
jpayne@7: 
jpayne@7:     # httplib is assuming a response body is available
jpayne@7:     # when parsing headers even when httplib only sends
jpayne@7:     # header data to parse_headers() This results in
jpayne@7:     # defects on multipart responses in particular.
jpayne@7:     # See: https://github.com/urllib3/urllib3/issues/800
jpayne@7: 
jpayne@7:     # So we ignore the following defects:
jpayne@7:     # - StartBoundaryNotFoundDefect:
jpayne@7:     #     The claimed start boundary was never found.
jpayne@7:     # - MultipartInvariantViolationDefect:
jpayne@7:     #     A message claimed to be a multipart but no subparts were found.
jpayne@7:     defects = [
jpayne@7:         defect
jpayne@7:         for defect in headers.defects
jpayne@7:         if not isinstance(
jpayne@7:             defect, (StartBoundaryNotFoundDefect, MultipartInvariantViolationDefect)
jpayne@7:         )
jpayne@7:     ]
jpayne@7: 
jpayne@7:     if defects or unparsed_data:
jpayne@7:         raise HeaderParsingError(defects=defects, unparsed_data=unparsed_data)
jpayne@7: 
jpayne@7: 
jpayne@7: def is_response_to_head(response: httplib.HTTPResponse) -> bool:
jpayne@7:     """
jpayne@7:     Checks whether the request of a response has been a HEAD-request.
jpayne@7: 
jpayne@7:     :param http.client.HTTPResponse response:
jpayne@7:         Response to check if the originating request
jpayne@7:         used 'HEAD' as a method.
jpayne@7:     """
jpayne@7:     # FIXME: Can we do this somehow without accessing private httplib _method?
jpayne@7:     method_str = response._method  # type: str  # type: ignore[attr-defined]
jpayne@7:     return method_str.upper() == "HEAD"