Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/h11/_writers.py: 38%

1# Code to read HTTP data 

2# 

3# Strategy: each writer takes an event + a write-some-bytes function, which is 

4# calls. 

5# 

6# WRITERS is a dict describing how to pick a reader. It maps states to either: 

7# - a writer 

8# - or, for body writers, a dict of framin-dependent writer factories 

9 

10from typing import Any, Callable, Dict, List, Tuple, Type, Union 

11 

12from ._events import Data, EndOfMessage, Event, InformationalResponse, Request, Response 

13from ._headers import Headers 

14from ._state import CLIENT, IDLE, SEND_BODY, SEND_RESPONSE, SERVER 

15from ._util import LocalProtocolError, Sentinel 

16 

17__all__ = ["WRITERS"] 

18 

19Writer = Callable[[bytes], Any] 

20 

21 

22def write_headers(headers: Headers, write: Writer) -> None: 

23 # "Since the Host field-value is critical information for handling a 

24 # request, a user agent SHOULD generate Host as the first header field 

25 # following the request-line." - RFC 7230 

26 raw_items = headers._full_items 

27 for raw_name, name, value in raw_items: 

28 if name == b"host": 

29 write(b"%s: %s\r\n" % (raw_name, value)) 

30 for raw_name, name, value in raw_items: 

31 if name != b"host": 

32 write(b"%s: %s\r\n" % (raw_name, value)) 

33 write(b"\r\n") 

34 

35 

36def write_request(request: Request, write: Writer) -> None: 

37 if request.http_version != b"1.1": 

38 raise LocalProtocolError("I only send HTTP/1.1") 

39 write(b"%s %s HTTP/1.1\r\n" % (request.method, request.target)) 

40 write_headers(request.headers, write) 

41 

42 

43# Shared between InformationalResponse and Response 

44def write_any_response( 

45 response: Union[InformationalResponse, Response], write: Writer 

46) -> None: 

47 if response.http_version != b"1.1": 

48 raise LocalProtocolError("I only send HTTP/1.1") 

49 status_bytes = str(response.status_code).encode("ascii") 

50 # We don't bother sending ascii status messages like "OK"; they're 

51 # optional and ignored by the protocol. (But the space after the numeric 

52 # status code is mandatory.) 

53 # 

54 # XX FIXME: could at least make an effort to pull out the status message 

55 # from stdlib's http.HTTPStatus table. Or maybe just steal their enums 

56 # (either by import or copy/paste). We already accept them as status codes 

57 # since they're of type IntEnum < int. 

58 write(b"HTTP/1.1 %s %s\r\n" % (status_bytes, response.reason)) 

59 write_headers(response.headers, write) 

60 

61 

62class BodyWriter: 

63 def __call__(self, event: Event, write: Writer) -> None: 

64 if type(event) is Data: 

65 self.send_data(event.data, write) 

66 elif type(event) is EndOfMessage: 

67 self.send_eom(event.headers, write) 

68 else: # pragma: no cover 

69 assert False 

70 

71 def send_data(self, data: bytes, write: Writer) -> None: 

72 pass 

73 

74 def send_eom(self, headers: Headers, write: Writer) -> None: 

75 pass 

76 

77 

78# 

79# These are all careful not to do anything to 'data' except call len(data) and 

80# write(data). This allows us to transparently pass-through funny objects, 

81# like placeholder objects referring to files on disk that will be sent via 

82# sendfile(2). 

83# 

84class ContentLengthWriter(BodyWriter): 

85 def __init__(self, length: int) -> None: 

86 self._length = length 

87 

88 def send_data(self, data: bytes, write: Writer) -> None: 

89 self._length -= len(data) 

90 if self._length < 0: 

91 raise LocalProtocolError("Too much data for declared Content-Length") 

92 write(data) 

93 

94 def send_eom(self, headers: Headers, write: Writer) -> None: 

95 if self._length != 0: 

96 raise LocalProtocolError("Too little data for declared Content-Length") 

97 if headers: 

98 raise LocalProtocolError("Content-Length and trailers don't mix") 

99 

100 

101class ChunkedWriter(BodyWriter): 

102 def send_data(self, data: bytes, write: Writer) -> None: 

103 # if we encoded 0-length data in the naive way, it would look like an 

104 # end-of-message. 

105 if not data: 

106 return 

107 write(b"%x\r\n" % len(data)) 

108 write(data) 

109 write(b"\r\n") 

110 

111 def send_eom(self, headers: Headers, write: Writer) -> None: 

112 write(b"0\r\n") 

113 write_headers(headers, write) 

114 

115 

116class Http10Writer(BodyWriter): 

117 def send_data(self, data: bytes, write: Writer) -> None: 

118 write(data) 

119 

120 def send_eom(self, headers: Headers, write: Writer) -> None: 

121 if headers: 

122 raise LocalProtocolError("can't send trailers to HTTP/1.0 client") 

123 # no need to close the socket ourselves, that will be taken care of by 

124 # Connection: close machinery 

125 

126 

127WritersType = Dict[ 

128 Union[Tuple[Type[Sentinel], Type[Sentinel]], Type[Sentinel]], 

129 Union[ 

130 Dict[str, Type[BodyWriter]], 

131 Callable[[Union[InformationalResponse, Response], Writer], None], 

132 Callable[[Request, Writer], None], 

133 ], 

134] 

135 

136WRITERS: WritersType = { 

137 (CLIENT, IDLE): write_request, 

138 (SERVER, IDLE): write_any_response, 

139 (SERVER, SEND_RESPONSE): write_any_response, 

140 SEND_BODY: { 

141 "chunked": ChunkedWriter, 

142 "content-length": ContentLengthWriter, 

143 "http/1.0": Http10Writer, 

144 }, 

145}