Method handle_h2_event

mitmproxy/proxy/layers/http/_http2.py:580–637

(self, event: h2.events.Event)

            yield from super()._handle_event(event)

    def handle_h2_event(self, event: h2.events.Event) -> CommandGenerator[bool]:
        if isinstance(event, h2.events.ResponseReceived):
            if (
                self.streams.get(event.stream_id, None)
                is not StreamState.EXPECTING_HEADERS
            ):
                yield from self.protocol_error(f"Received unexpected HTTP/2 response.")
                return True

            try:
                status_code, headers = parse_h2_response_headers(event.headers)
            except ValueError as e:
                yield from self.protocol_error(f"Invalid HTTP/2 response headers: {e}")
                return True

            response = http.Response(
                http_version=b"HTTP/2.0",
                status_code=status_code,
                reason=b"",
                headers=headers,
                content=None,
                trailers=None,
                timestamp_start=time.time(),
                timestamp_end=None,
            )
            self.streams[event.stream_id] = StreamState.HEADERS_RECEIVED
            yield ReceiveHttp(
                ResponseHeaders(event.stream_id, response, bool(event.stream_ended))
            )
            return False
        elif isinstance(event, h2.events.InformationalResponseReceived):
            # We violate the spec here ("A proxy MUST forward 1xx responses", RFC 7231),
            # but that's probably fine:
            # - 100 Continue is sent by mitmproxy to clients (irrespective of what the server does).
            # - 101 Switching Protocols is not allowed for HTTP/2.
            # - 102 Processing is WebDAV only and also ignorable.
            # - 103 Early Hints is not mission-critical.
            headers = http.Headers(event.headers)
            status: str | int = "<unknown status>"
            try:
                status = int(headers[":status"])
                reason = status_codes.RESPONSES.get(status, "")
            except (KeyError, ValueError):
                reason = ""
            yield Log(f"Swallowing HTTP/2 informational response: {status} {reason}")
            return False
        elif isinstance(event, h2.events.RequestReceived):
            yield from self.protocol_error(
                f"HTTP/2 protocol error: received request from server"
            )
            return True
        elif isinstance(event, h2.events.RemoteSettingsChanged):
            # We have received at least one settings from now,
            # which means we can rely on the max concurrency in remote_settings
            self.provisional_max_concurrency = None
            return (yield from super().handle_h2_event(event))
        else:
            return (yield from super().handle_h2_event(event))