codekingpro/portable-devtools

1# Code to read HTTP data2#3# Strategy: each writer takes an event + a write-some-bytes function, which is4# calls.5#6# WRITERS is a dict describing how to pick a reader. It maps states to either:7# - a writer8# - or, for body writers, a dict of framin-dependent writer factories9 10from typing import Any, Callable, Dict, List, Tuple, Type, Union11 12from ._events import Data, EndOfMessage, Event, InformationalResponse, Request, Response13from ._headers import Headers14from ._state import CLIENT, IDLE, SEND_BODY, SEND_RESPONSE, SERVER15from ._util import LocalProtocolError, Sentinel16 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 a24    # request, a user agent SHOULD generate Host as the first header field25    # following the request-line." - RFC 723026    raw_items = headers._full_items27    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 Response44def write_any_response(45    response: Union[InformationalResponse, Response], write: Writer46) -> 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're51    # optional and ignored by the protocol. (But the space after the numeric52    # status code is mandatory.)53    #54    # XX FIXME: could at least make an effort to pull out the status message55    # from stdlib's http.HTTPStatus table. Or maybe just steal their enums56    # (either by import or copy/paste). We already accept them as status codes57    # 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 cover69            assert False70 71    def send_data(self, data: bytes, write: Writer) -> None:72        pass73 74    def send_eom(self, headers: Headers, write: Writer) -> None:75        pass76 77 78#79# These are all careful not to do anything to 'data' except call len(data) and80# write(data). This allows us to transparently pass-through funny objects,81# like placeholder objects referring to files on disk that will be sent via82# sendfile(2).83#84class ContentLengthWriter(BodyWriter):85    def __init__(self, length: int) -> None:86        self._length = length87 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 an104        # end-of-message.105        if not data:106            return107        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 by124        # Connection: close machinery125 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}146