Class Http1Server

mitmproxy/proxy/layers/http/_http1.py:223–336

A simple HTTP/1 server with no pipelining support.

class Http1Server(Http1Connection):
    """A simple HTTP/1 server with no pipelining support."""

    ReceiveProtocolError = RequestProtocolError
    ReceiveData = RequestData
    ReceiveEndOfMessage = RequestEndOfMessage
    stream_id: int

    def __init__(self, context: Context):
        super().__init__(context, context.client)
        self.stream_id = 1

    def send(self, event: HttpEvent) -> layer.CommandGenerator[None]:
        assert event.stream_id == self.stream_id
        if isinstance(event, ResponseHeaders):
            self.response = response = event.response

            if response.is_http2 or response.is_http3:
                response = response.copy()
                # Convert to an HTTP/1 response.
                response.http_version = "HTTP/1.1"
                # not everyone supports empty reason phrases, so we better make up one.
                response.reason = status_codes.RESPONSES.get(response.status_code, "")
                # Shall we set a Content-Length header here if there is none?
                # For now, let's try to modify as little as possible.

            raw = http1.assemble_response_head(response)
            yield commands.SendData(self.conn, raw)
        elif isinstance(event, ResponseData):
            assert self.response
            if "chunked" in self.response.headers.get("transfer-encoding", "").lower():
                raw = b"%x\r\n%s\r\n" % (len(event.data), event.data)
            else:
                raw = event.data
            if raw:
                yield commands.SendData(self.conn, raw)
        elif isinstance(event, ResponseEndOfMessage):
            assert self.request
            assert self.response
            if (
                self.request.method.upper() != "HEAD"
                and "chunked"
                in self.response.headers.get("transfer-encoding", "").lower()
            ):
                yield commands.SendData(self.conn, b"0\r\n\r\n")
            yield from self.mark_done(response=True)
        elif isinstance(event, ResponseProtocolError):
            if not (self.conn.state & ConnectionState.CAN_WRITE):
                return
            status = event.code.http_status_code()
            if not self.response and status is not None:
                yield commands.SendData(
                    self.conn, make_error_response(status, event.message)
                )
            yield commands.CloseConnection(self.conn)
        else:
            raise AssertionError(f"Unexpected event: {event}")