codekingpro/portable-devtools
115k
1import collections2import enum3import time4from dataclasses import dataclass5from functools import cached_property6from logging import DEBUG7from logging import WARNING8 9import wsproto.handshake10 11from ...context import Context12from ...mode_specs import ReverseMode13from ...mode_specs import UpstreamMode14from ..quic import QuicStreamEvent15from ._base import HttpCommand16from ._base import HttpConnection17from ._base import ReceiveHttp18from ._base import StreamId19from ._events import ErrorCode20from ._events import HttpEvent21from ._events import RequestData22from ._events import RequestEndOfMessage23from ._events import RequestHeaders24from ._events import RequestProtocolError25from ._events import RequestTrailers26from ._events import ResponseData27from ._events import ResponseEndOfMessage28from ._events import ResponseHeaders29from ._events import ResponseProtocolError30from ._events import ResponseTrailers31from ._hooks import HttpConnectedHook32from ._hooks import HttpConnectErrorHook33from ._hooks import HttpConnectHook34from ._hooks import HttpErrorHook35from ._hooks import HttpRequestHeadersHook36from ._hooks import HttpRequestHook37from ._hooks import HttpResponseHeadersHook38from ._hooks import HttpResponseHook39from ._http1 import Http1Client40from ._http1 import Http1Connection41from ._http1 import Http1Server42from ._http2 import Http2Client43from ._http2 import Http2Server44from ._http3 import Http3Client45from ._http3 import Http3Server46from mitmproxy import flow47from mitmproxy import http48from mitmproxy.connection import Connection49from mitmproxy.connection import Server50from mitmproxy.connection import TransportProtocol51from mitmproxy.net import server_spec52from mitmproxy.net.http import url53from mitmproxy.net.http.http1 import expected_http_body_size54from mitmproxy.net.http.validate import validate_headers55from mitmproxy.proxy import commands56from mitmproxy.proxy import events57from mitmproxy.proxy import layer58from mitmproxy.proxy import tunnel59from mitmproxy.proxy.layers import quic60from mitmproxy.proxy.layers import tcp61from mitmproxy.proxy.layers import tls62from mitmproxy.proxy.layers import websocket63from mitmproxy.proxy.layers.http import _upstream_proxy64from mitmproxy.proxy.utils import expect65from mitmproxy.proxy.utils import ReceiveBuffer66from mitmproxy.utils import human67from mitmproxy.websocket import WebSocketData68 69 70class HTTPMode(enum.Enum):71 regular = 172 transparent = 273 upstream = 374 75 76def validate_request(77 mode: HTTPMode, request: http.Request, validate_inbound_headers: bool78) -> str | None:79 if request.scheme not in ("http", "https", ""):80 return f"Invalid request scheme: {request.scheme}"81 if mode is HTTPMode.transparent and request.method == "CONNECT":82 return (83 f"mitmproxy received an HTTP CONNECT request even though it is not running in regular/upstream mode. "84 f"This usually indicates a misconfiguration, please see the mitmproxy mode documentation for details."85 )86 if validate_inbound_headers:87 try:88 validate_headers(request)89 except ValueError as e:90 return (91 f"Received {e} from client, refusing to prevent request smuggling attacks. "92 "Disable the validate_inbound_headers option to skip this security check."93 )94 return None95 96 97def is_h3_alpn(alpn: bytes | None) -> bool:98 return alpn == b"h3" or (alpn is not None and alpn.startswith(b"h3-"))99 100 101@dataclass102class GetHttpConnection(HttpCommand):103 """104 Open an HTTP Connection. This may not actually open a connection, but return an existing HTTP connection instead.105 """106 107 blocking = True108 address: tuple[str, int]109 tls: bool110 via: server_spec.ServerSpec | None111 transport_protocol: TransportProtocol = "tcp"112 113 def __hash__(self):114 return id(self)115 116 def connection_spec_matches(self, connection: Connection) -> bool:117 return (118 isinstance(connection, Server)119 and self.address == connection.address120 and self.tls == connection.tls121 and self.via == connection.via122 and self.transport_protocol == connection.transport_protocol123 )124 125 126@dataclass127class GetHttpConnectionCompleted(events.CommandCompleted):128 command: GetHttpConnection129 reply: tuple[None, str] | tuple[Connection, None]130 """connection object, error message"""131 132 133@dataclass134class RegisterHttpConnection(HttpCommand):135 """136 Register that a HTTP connection attempt has been completed.137 """138 139 connection: Connection140 err: str | None141 142 143@dataclass144class SendHttp(HttpCommand):145 event: HttpEvent146 connection: Connection147 148 def __repr__(self) -> str:149 return f"Send({self.event})"150 151 152@dataclass153class DropStream(HttpCommand):154 """Signal to the HTTP layer that this stream is done processing and can be dropped from memory."""155 156 stream_id: StreamId157 158 159class HttpStream(layer.Layer):160 request_body_buf: ReceiveBuffer161 response_body_buf: ReceiveBuffer162 flow: http.HTTPFlow163 stream_id: StreamId164 child_layer: layer.Layer | None = None165 166 @cached_property167 def mode(self) -> HTTPMode:168 i = self.context.layers.index(self)169 parent = self.context.layers[i - 1]170 assert isinstance(parent, HttpLayer)171 return parent.mode172 173 def __init__(self, context: Context, stream_id: int) -> None:174 super().__init__(context)175 self.request_body_buf = ReceiveBuffer()176 self.response_body_buf = ReceiveBuffer()177 self.client_state = self.state_uninitialized178 self.server_state = self.state_uninitialized179 self.stream_id = stream_id180 181 def __repr__(self):182 if self._handle_event == self.passthrough:183 return f"HttpStream(id={self.stream_id}, passthrough)"184 else:185 return (186 f"HttpStream("187 f"id={self.stream_id}, "188 f"client_state={self.client_state.__name__}, "189 f"server_state={self.server_state.__name__}"190 f")"191 )192 193 @expect(events.Start, HttpEvent)194 def _handle_event(self, event: events.Event) -> layer.CommandGenerator[None]:195 if isinstance(event, events.Start):196 self.client_state = self.state_wait_for_request_headers197 elif isinstance(event, (RequestProtocolError, ResponseProtocolError)):198 yield from self.handle_protocol_error(event)199 elif isinstance(200 event, (RequestHeaders, RequestData, RequestTrailers, RequestEndOfMessage)201 ):202 yield from self.client_state(event)203 else:204 yield from self.server_state(event)205 206 @expect(RequestHeaders)207 def state_wait_for_request_headers(208 self, event: RequestHeaders209 ) -> layer.CommandGenerator[None]:210 if not event.replay_flow:211 self.flow = http.HTTPFlow(self.context.client, self.context.server)212 213 else:214 self.flow = event.replay_flow215 self.flow.request = event.request216 self.flow.live = True217 218 if (yield from self.check_invalid(True)):219 return220 221 if self.flow.request.method == "CONNECT":222 return (yield from self.handle_connect())223 224 if self.mode is HTTPMode.transparent:225 # Determine .scheme, .host and .port attributes for transparent requests226 assert self.context.server.address227 self.flow.request.data.host = self.context.server.address[0]228 self.flow.request.data.port = self.context.server.address[1]229 self.flow.request.scheme = "https" if self.context.server.tls else "http"230 elif not self.flow.request.host:231 # We need to extract destination information from the host header.232 try:233 host, port = url.parse_authority(234 self.flow.request.host_header or "", check=True235 )236 except ValueError:237 yield SendHttp(238 ResponseProtocolError(239 self.stream_id,240 "HTTP request has no host header, destination unknown.",241 ErrorCode.DESTINATION_UNKNOWN,242 ),243 self.context.client,244 )245 self.client_state = self.state_errored246 return247 else:248 if port is None:249 port = 443 if self.context.client.tls else 80250 self.flow.request.data.host = host251 self.flow.request.data.port = port252 self.flow.request.scheme = (253 "https" if self.context.client.tls else "http"254 )255 256 if self.mode is HTTPMode.regular and not (257 self.flow.request.is_http2 or self.flow.request.is_http3258 ):259 # Set the request target to origin-form for HTTP/1, some servers don't support absolute-form requests.260 # see https://github.com/mitmproxy/mitmproxy/issues/1759261 self.flow.request.authority = ""262 263 # update host header in reverse proxy mode264 if (265 isinstance(self.context.client.proxy_mode, ReverseMode)266 and not self.context.options.keep_host_header267 ):268 assert self.context.server.address269 self.flow.request.host_header = url.hostport(270 "https" if self.context.server.tls else "http",271 self.context.server.address[0],272 self.context.server.address[1],273 )274 275 if not event.end_stream and (yield from self.check_body_size(True)):276 return277 278 yield HttpRequestHeadersHook(self.flow)279 if (yield from self.check_killed(True)):280 return281 282 if self.flow.request.headers.get("expect", "").lower() == "100-continue":283 continue_response = http.Response.make(100)284 continue_response.headers.clear()285 yield SendHttp(286 ResponseHeaders(self.stream_id, continue_response), self.context.client287 )288 self.flow.request.headers.pop("expect")289 290 if self.flow.request.stream and not event.end_stream:291 yield from self.start_request_stream()292 else:293 self.client_state = self.state_consume_request_body294 self.server_state = self.state_wait_for_response_headers295 296 def start_request_stream(self) -> layer.CommandGenerator[None]:297 if self.flow.response:298 raise NotImplementedError(299 "Can't set a response and enable streaming at the same time."300 )301 ok = yield from self.make_server_connection()302 if not ok:303 self.client_state = self.state_errored304 return305 yield SendHttp(306 RequestHeaders(self.stream_id, self.flow.request, end_stream=False),307 self.context.server,308 )309 yield commands.Log(f"Streaming request to {self.flow.request.host}.")310 self.client_state = self.state_stream_request_body311 312 @expect(RequestData, RequestTrailers, RequestEndOfMessage)313 def state_stream_request_body(314 self, event: RequestData | RequestEndOfMessage315 ) -> layer.CommandGenerator[None]:316 if isinstance(event, RequestData):317 if callable(self.flow.request.stream):318 chunks = self.flow.request.stream(event.data)319 if isinstance(chunks, bytes):320 chunks = [chunks]321 else:322 chunks = [event.data]323 for chunk in chunks:324 if self.context.options.store_streamed_bodies:325 self.request_body_buf += chunk326 yield SendHttp(RequestData(self.stream_id, chunk), self.context.server)327 elif isinstance(event, RequestTrailers):328 # we don't do anything further here, we wait for RequestEndOfMessage first to trigger the request hook.329 self.flow.request.trailers = event.trailers330 elif isinstance(event, RequestEndOfMessage):331 if callable(self.flow.request.stream):332 chunks = self.flow.request.stream(b"")333 if chunks == b"":334 chunks = []335 elif isinstance(chunks, bytes):336 chunks = [chunks]337 for chunk in chunks:338 if self.context.options.store_streamed_bodies:339 self.request_body_buf += chunk340 yield SendHttp(341 RequestData(self.stream_id, chunk), self.context.server342 )343 344 if self.context.options.store_streamed_bodies:345 self.flow.request.data.content = bytes(self.request_body_buf)346 self.request_body_buf.clear()347 self.flow.request.timestamp_end = time.time()348 yield HttpRequestHook(self.flow)349 self.client_state = self.state_done350 351 if self.flow.request.trailers:352 # we've delayed sending trailers until after `request` has been triggered.353 yield SendHttp(354 RequestTrailers(self.stream_id, self.flow.request.trailers),355 self.context.server,356 )357 yield SendHttp(event, self.context.server)358 359 if self.server_state == self.state_done:360 yield from self.flow_done()361 362 @expect(RequestData, RequestTrailers, RequestEndOfMessage)363 def state_consume_request_body(364 self, event: events.Event365 ) -> layer.CommandGenerator[None]:366 if isinstance(event, RequestData):367 self.request_body_buf += event.data368 yield from self.check_body_size(True)369 elif isinstance(event, RequestTrailers):370 assert self.flow.request371 self.flow.request.trailers = event.trailers372 elif isinstance(event, RequestEndOfMessage):373 self.flow.request.timestamp_end = time.time()374 self.flow.request.data.content = bytes(self.request_body_buf)375 self.request_body_buf.clear()376 self.client_state = self.state_done377 yield HttpRequestHook(self.flow)378 if (yield from self.check_killed(True)):379 return380 elif self.flow.response:381 # response was set by an inline script.382 # we now need to emulate the responseheaders hook.383 self.flow.response.timestamp_start = time.time()384 yield HttpResponseHeadersHook(self.flow)385 if (yield from self.check_killed(True)):386 return387 yield from self.send_response()388 else:389 ok = yield from self.make_server_connection()390 if not ok:391 return392 393 content = self.flow.request.raw_content394 done_after_headers = not (content or self.flow.request.trailers)395 yield SendHttp(396 RequestHeaders(397 self.stream_id, self.flow.request, done_after_headers398 ),399 self.context.server,400 )401 if content:402 yield SendHttp(403 RequestData(self.stream_id, content), self.context.server404 )405 if self.flow.request.trailers:406 yield SendHttp(407 RequestTrailers(self.stream_id, self.flow.request.trailers),408 self.context.server,409 )410 yield SendHttp(RequestEndOfMessage(self.stream_id), self.context.server)411 412 @expect(ResponseHeaders)413 def state_wait_for_response_headers(414 self, event: ResponseHeaders415 ) -> layer.CommandGenerator[None]:416 self.flow.response = event.response417 418 if not event.end_stream and (yield from self.check_body_size(False)):419 return420 if (yield from self.check_invalid(False)):421 return422 423 yield HttpResponseHeadersHook(self.flow)424 if (yield from self.check_killed(True)):425 return426 427 elif self.flow.response.stream and not event.end_stream:428 yield from self.start_response_stream()429 else:430 self.server_state = self.state_consume_response_body431 432 def start_response_stream(self) -> layer.CommandGenerator[None]:433 assert self.flow.response434 yield SendHttp(435 ResponseHeaders(self.stream_id, self.flow.response, end_stream=False),436 self.context.client,437 )438 yield commands.Log(f"Streaming response from {self.flow.request.host}.")439 self.server_state = self.state_stream_response_body440 441 @expect(ResponseData, ResponseTrailers, ResponseEndOfMessage)442 def state_stream_response_body(443 self, event: events.Event444 ) -> layer.CommandGenerator[None]:445 assert self.flow.response446 if isinstance(event, ResponseData):447 if callable(self.flow.response.stream):448 chunks = self.flow.response.stream(event.data)449 if isinstance(chunks, bytes):450 chunks = [chunks]451 else:452 chunks = [event.data]453 for chunk in chunks:454 if self.context.options.store_streamed_bodies:455 self.response_body_buf += chunk456 yield SendHttp(ResponseData(self.stream_id, chunk), self.context.client)457 elif isinstance(event, ResponseTrailers):458 self.flow.response.trailers = event.trailers459 # will be sent in send_response() after the response hook.460 elif isinstance(event, ResponseEndOfMessage):461 if callable(self.flow.response.stream):462 chunks = self.flow.response.stream(b"")463 if chunks == b"":464 chunks = []465 elif isinstance(chunks, bytes):466 chunks = [chunks]467 for chunk in chunks:468 if self.context.options.store_streamed_bodies:469 self.response_body_buf += chunk470 yield SendHttp(471 ResponseData(self.stream_id, chunk), self.context.client472 )473 if self.context.options.store_streamed_bodies:474 self.flow.response.data.content = bytes(self.response_body_buf)475 self.response_body_buf.clear()476 yield from self.send_response(already_streamed=True)477 478 @expect(ResponseData, ResponseTrailers, ResponseEndOfMessage)479 def state_consume_response_body(480 self, event: events.Event481 ) -> layer.CommandGenerator[None]:482 if isinstance(event, ResponseData):483 self.response_body_buf += event.data484 yield from self.check_body_size(False)485 elif isinstance(event, ResponseTrailers):486 assert self.flow.response487 self.flow.response.trailers = event.trailers488 elif isinstance(event, ResponseEndOfMessage):489 assert self.flow.response490 self.flow.response.data.content = bytes(self.response_body_buf)491 self.response_body_buf.clear()492 yield from self.send_response()493 494 def send_response(self, already_streamed: bool = False):495 """We have either consumed the entire response from the server or the response was set by an addon."""496 assert self.flow.response497 self.flow.response.timestamp_end = time.time()498 499 is_websocket = (500 self.flow.response.status_code == 101501 and self.flow.response.headers.get("upgrade", "").lower() == "websocket"502 and self.flow.request.headers.get("Sec-WebSocket-Version", "").encode()503 == wsproto.handshake.WEBSOCKET_VERSION504 and self.context.options.websocket505 )506 if is_websocket:507 # We need to set this before calling the response hook508 # so that addons can determine if a WebSocket connection is following up.509 self.flow.websocket = WebSocketData()510 511 yield HttpResponseHook(self.flow)512 self.server_state = self.state_done513 if (yield from self.check_killed(False)):514 return515 516 if not already_streamed:517 content = self.flow.response.raw_content518 done_after_headers = not (content or self.flow.response.trailers)519 yield SendHttp(520 ResponseHeaders(self.stream_id, self.flow.response, done_after_headers),521 self.context.client,522 )523 if content:524 yield SendHttp(525 ResponseData(self.stream_id, content), self.context.client526 )527 528 if self.flow.response.trailers:529 yield SendHttp(530 ResponseTrailers(self.stream_id, self.flow.response.trailers),531 self.context.client,532 )533 534 if self.client_state == self.state_done:535 yield from self.flow_done()536 537 def flow_done(self) -> layer.CommandGenerator[None]:538 if not self.flow.websocket:539 self.flow.live = False540 541 assert self.flow.response542 if self.flow.response.status_code == 101:543 if self.flow.websocket:544 self.child_layer = websocket.WebsocketLayer(self.context, self.flow)545 elif self.context.options.rawtcp:546 self.child_layer = tcp.TCPLayer(self.context)547 else:548 yield commands.Log(549 f"Sent HTTP 101 response, but no protocol is enabled to upgrade to.",550 WARNING,551 )552 yield commands.CloseConnection(self.context.client)553 self.client_state = self.server_state = self.state_errored554 return555 if self.debug:556 yield commands.Log(557 f"{self.debug}[http] upgrading to {self.child_layer}", DEBUG558 )559 self._handle_event = self.passthrough560 yield from self.child_layer.handle_event(events.Start())561 else:562 yield DropStream(self.stream_id)563 564 # delay sending EOM until the child layer is set up,565 # we may get data immediately and need to be prepared to handle it.566 yield SendHttp(ResponseEndOfMessage(self.stream_id), self.context.client)567 568 def check_body_size(self, request: bool) -> layer.CommandGenerator[bool]:569 """570 Check if the body size exceeds limits imposed by stream_large_bodies or body_size_limit.571 572 Returns `True` if the body size exceeds body_size_limit and further processing should be stopped.573 """574 if not (575 self.context.options.stream_large_bodies576 or self.context.options.body_size_limit577 ):578 return False579 580 # Step 1: Determine the expected body size. This can either come from a known content-length header,581 # or from the amount of currently buffered bytes (e.g. for chunked encoding).582 response = not request583 expected_size: int | None584 # the 'late' case: we already started consuming the body585 if request and self.request_body_buf:586 expected_size = len(self.request_body_buf)587 elif response and self.response_body_buf:588 expected_size = len(self.response_body_buf)589 else:590 # the 'early' case: we have not started consuming the body591 try:592 expected_size = expected_http_body_size(593 self.flow.request, self.flow.response if response else None594 )595 except ValueError: # pragma: no cover596 # we just don't stream/kill malformed content-length headers.597 expected_size = None598 599 if expected_size is None or expected_size <= 0:600 return False601 602 # Step 2: Do we need to abort this?603 max_total_size = human.parse_size(self.context.options.body_size_limit)604 if max_total_size is not None and expected_size > max_total_size:605 if request and not self.request_body_buf:606 yield HttpRequestHeadersHook(self.flow)607 if response and not self.response_body_buf:608 yield HttpResponseHeadersHook(self.flow)609 610 err_msg = f"{'Request' if request else 'Response'} body exceeds mitmproxy's body_size_limit."611 err_code = (612 ErrorCode.REQUEST_TOO_LARGE if request else ErrorCode.RESPONSE_TOO_LARGE613 )614 615 self.flow.error = flow.Error(err_msg)616 yield HttpErrorHook(self.flow)617 yield SendHttp(618 ResponseProtocolError(self.stream_id, err_msg, err_code),619 self.context.client,620 )621 self.client_state = self.state_errored622 if response:623 yield SendHttp(624 RequestProtocolError(self.stream_id, err_msg, err_code),625 self.context.server,626 )627 self.server_state = self.state_errored628 self.flow.live = False629 return True630 631 # Step 3: Do we need to stream this?632 max_stream_size = human.parse_size(self.context.options.stream_large_bodies)633 if max_stream_size is not None and expected_size > max_stream_size:634 if request:635 self.flow.request.stream = True636 if self.request_body_buf:637 # clear buffer and then fake a DataReceived event with everything we had in the buffer so far.638 body_buf = bytes(self.request_body_buf)639 self.request_body_buf.clear()640 yield from self.start_request_stream()641 yield from self.handle_event(RequestData(self.stream_id, body_buf))642 if response:643 assert self.flow.response644 self.flow.response.stream = True645 if self.response_body_buf:646 body_buf = bytes(self.response_body_buf)647 self.response_body_buf.clear()648 yield from self.start_response_stream()649 yield from self.handle_event(ResponseData(self.stream_id, body_buf))650 return False651 652 def check_invalid(self, request: bool) -> layer.CommandGenerator[bool]:653 err: str | None = None654 if request:655 err = validate_request(656 self.mode,657 self.flow.request,658 self.context.options.validate_inbound_headers,659 )660 elif self.context.options.validate_inbound_headers:661 assert self.flow.response is not None662 try:663 validate_headers(self.flow.response)664 except ValueError as e:665 err = (666 f"Received {e} from server, refusing to prevent request smuggling attacks. "667 "Disable the validate_inbound_headers option to skip this security check."668 )669 670 if err:671 self.flow.error = flow.Error(err)672 673 if request:674 # flow has not been seen yet, register it.675 yield HttpRequestHeadersHook(self.flow)676 else:677 # immediately kill server connection678 yield commands.CloseConnection(self.flow.server_conn)679 yield HttpErrorHook(self.flow)680 yield SendHttp(681 ResponseProtocolError(682 self.stream_id,683 err,684 ErrorCode.REQUEST_VALIDATION_FAILED685 if request686 else ErrorCode.RESPONSE_VALIDATION_FAILED,687 ),688 self.context.client,689 )690 self.flow.live = False691 self.client_state = self.server_state = self.state_errored692 return True693 else:694 return False695 696 def check_killed(self, emit_error_hook: bool) -> layer.CommandGenerator[bool]:697 killed_by_us = (698 self.flow.error and self.flow.error.msg == flow.Error.KILLED_MESSAGE699 )700 # The client may have closed the connection while we were waiting for the hook to complete.701 # We peek into the event queue to see if that is the case.702 killed_by_remote = None703 for evt in self._paused_event_queue:704 if isinstance(evt, RequestProtocolError):705 killed_by_remote = evt.message706 break707 708 if killed_by_remote:709 if not self.flow.error:710 self.flow.error = flow.Error(killed_by_remote)711 if killed_by_us or killed_by_remote:712 if emit_error_hook:713 yield HttpErrorHook(self.flow)714 yield SendHttp(715 ResponseProtocolError(self.stream_id, "killed", ErrorCode.KILL),716 self.context.client,717 )718 self.flow.live = False719 self.client_state = self.server_state = self.state_errored720 return True721 return False722 723 def handle_protocol_error(724 self, event: RequestProtocolError | ResponseProtocolError725 ) -> layer.CommandGenerator[None]:726 is_client_error_but_we_already_talk_upstream = (727 isinstance(event, RequestProtocolError)728 and self.client_state in (self.state_stream_request_body, self.state_done)729 and self.server_state not in (self.state_done, self.state_errored)730 )731 need_error_hook = not (732 self.client_state == self.state_errored733 or self.server_state in (self.state_done, self.state_errored)734 )735 736 if is_client_error_but_we_already_talk_upstream:737 yield SendHttp(event, self.context.server)738 self.client_state = self.state_errored739 740 if need_error_hook:741 # We don't want to trigger both a response hook and an error hook,742 # so we need to check if the response is done yet or not.743 self.flow.error = flow.Error(event.message)744 yield HttpErrorHook(self.flow)745 746 if (yield from self.check_killed(False)):747 return748 749 if isinstance(event, ResponseProtocolError):750 if self.client_state != self.state_errored:751 yield SendHttp(event, self.context.client)752 self.server_state = self.state_errored753 754 self.flow.live = False755 yield DropStream(self.stream_id)756 757 def make_server_connection(self) -> layer.CommandGenerator[bool]:758 connection, err = yield GetHttpConnection(759 (self.flow.request.host, self.flow.request.port),760 self.flow.request.scheme == "https",761 self.flow.server_conn.via,762 self.flow.server_conn.transport_protocol,763 )764 if err:765 yield from self.handle_protocol_error(766 ResponseProtocolError(self.stream_id, err, ErrorCode.CONNECT_FAILED)767 )768 return False769 else:770 self.context.server = self.flow.server_conn = connection771 return True772 773 def handle_connect(self) -> layer.CommandGenerator[None]:774 self.client_state = self.state_done775 yield HttpConnectHook(self.flow)776 if (yield from self.check_killed(False)):777 return778 779 self.context.server.address = (self.flow.request.host, self.flow.request.port)780 781 if self.mode == HTTPMode.regular:782 yield from self.handle_connect_regular()783 else:784 yield from self.handle_connect_upstream()785 786 def handle_connect_regular(self):787 if (788 not self.flow.response789 and self.context.options.connection_strategy == "eager"790 ):791 err = yield commands.OpenConnection(self.context.server)792 if err:793 self.flow.response = http.Response.make(794 502,795 f"Cannot connect to {human.format_address(self.context.server.address)}: {err} "796 f"If you plan to redirect requests away from this server, "797 f"consider setting `connection_strategy` to `lazy` to suppress early connections.",798 )799 self.child_layer = layer.NextLayer(self.context)800 yield from self.handle_connect_finish()801 802 def handle_connect_upstream(self):803 self.child_layer = _upstream_proxy.HttpUpstreamProxy.make(self.context, True)[0]804 yield from self.handle_connect_finish()805 806 def handle_connect_finish(self):807 if not self.flow.response:808 # Do not send any response headers as it breaks proxying non-80 ports on809 # Android emulators using the -http-proxy option.810 self.flow.response = http.Response(811 self.flow.request.data.http_version,812 200,813 b"Connection established",814 http.Headers(),815 b"",816 None,817 time.time(),818 time.time(),819 )820 821 if 200 <= self.flow.response.status_code < 300:822 yield HttpConnectedHook(self.flow)823 self.child_layer = self.child_layer or layer.NextLayer(self.context)824 self._handle_event = self.passthrough825 yield from self.child_layer.handle_event(events.Start())826 else:827 yield HttpConnectErrorHook(self.flow)828 self.client_state = self.state_errored829 self.flow.live = False830 831 content = self.flow.response.raw_content832 done_after_headers = not (content or self.flow.response.trailers)833 yield SendHttp(834 ResponseHeaders(self.stream_id, self.flow.response, done_after_headers),835 self.context.client,836 )837 if content:838 yield SendHttp(ResponseData(self.stream_id, content), self.context.client)839 840 if self.flow.response.trailers:841 yield SendHttp(842 ResponseTrailers(self.stream_id, self.flow.response.trailers),843 self.context.client,844 )845 yield SendHttp(ResponseEndOfMessage(self.stream_id), self.context.client)846 847 @expect(RequestData, RequestEndOfMessage, events.Event)848 def passthrough(self, event: events.Event) -> layer.CommandGenerator[None]:849 assert self.flow.response850 assert self.child_layer851 # HTTP events -> normal connection events852 if isinstance(event, RequestData):853 event = events.DataReceived(self.context.client, event.data)854 elif isinstance(event, ResponseData):855 event = events.DataReceived(self.context.server, event.data)856 elif isinstance(event, RequestEndOfMessage):857 event = events.ConnectionClosed(self.context.client)858 elif isinstance(event, ResponseEndOfMessage):859 event = events.ConnectionClosed(self.context.server)860 861 for command in self.child_layer.handle_event(event):862 # normal connection events -> HTTP events863 if isinstance(command, commands.SendData):864 if command.connection == self.context.client:865 yield SendHttp(866 ResponseData(self.stream_id, command.data), self.context.client867 )868 elif (869 command.connection == self.context.server870 and self.flow.response.status_code == 101871 ):872 # there only is a HTTP server connection if we have switched protocols,873 # not if a connection is established via CONNECT.874 yield SendHttp(875 RequestData(self.stream_id, command.data), self.context.server876 )877 else:878 yield command879 elif isinstance(command, commands.CloseConnection):880 if command.connection == self.context.client:881 yield SendHttp(882 ResponseProtocolError(883 self.stream_id, "EOF", ErrorCode.PASSTHROUGH_CLOSE884 ),885 self.context.client,886 )887 elif (888 command.connection == self.context.server889 and self.flow.response.status_code == 101890 ):891 yield SendHttp(892 RequestProtocolError(893 self.stream_id, "EOF", ErrorCode.PASSTHROUGH_CLOSE894 ),895 self.context.server,896 )897 else:898 # If we are running TCP over HTTP we want to be consistent with half-closes.899 # The easiest approach for this is to just always full close for now.900 # Alternatively, we could signal that we want a half close only through ResponseProtocolError,901 # but that is more complex to implement.902 if isinstance(command, commands.CloseTcpConnection):903 command = commands.CloseConnection(command.connection)904 yield command905 else:906 yield command907 908 @expect()909 def state_uninitialized(self, _) -> layer.CommandGenerator[None]:910 yield from ()911 912 @expect()913 def state_done(self, _) -> layer.CommandGenerator[None]:914 yield from ()915 916 def state_errored(self, _) -> layer.CommandGenerator[None]:917 # silently consume every event.918 yield from ()919 920 921class HttpLayer(layer.Layer):922 """923 ConnectionEvent: We have received b"GET /\r\n\r\n" from the client.924 HttpEvent: We have received request headers925 HttpCommand: Send request headers to X926 ConnectionCommand: Send b"GET /\r\n\r\n" to server.927 928 ConnectionEvent -> HttpEvent -> HttpCommand -> ConnectionCommand929 """930 931 mode: HTTPMode932 command_sources: dict[commands.Command, layer.Layer]933 streams: dict[int, HttpStream]934 connections: dict[Connection, layer.Layer]935 waiting_for_establishment: collections.defaultdict[936 Connection, list[GetHttpConnection]937 ]938 939 def __init__(self, context: Context, mode: HTTPMode):940 super().__init__(context)941 self.mode = mode942 943 self.waiting_for_establishment = collections.defaultdict(list)944 self.streams = {}945 self.command_sources = {}946 self.connections = {}947 948 def __repr__(self):949 return f"HttpLayer({self.mode.name}, conns: {len(self.connections)})"950 951 def _handle_event(self, event: events.Event):952 if isinstance(event, events.Start):953 http_conn: HttpConnection954 if is_h3_alpn(self.context.client.alpn):955 http_conn = Http3Server(self.context.fork())956 elif self.context.client.alpn == b"h2":957 http_conn = Http2Server(self.context.fork())958 else:959 http_conn = Http1Server(self.context.fork())960 961 # may have been set by client playback.962 self.connections.setdefault(self.context.client, http_conn)963 yield from self.event_to_child(self.connections[self.context.client], event)964 if self.mode is HTTPMode.upstream:965 proxy_mode = self.context.client.proxy_mode966 assert isinstance(proxy_mode, UpstreamMode)967 self.context.server.via = (proxy_mode.scheme, proxy_mode.address)968 elif isinstance(event, events.CommandCompleted):969 stream = self.command_sources.pop(event.command)970 yield from self.event_to_child(stream, event)971 elif isinstance(event, events.MessageInjected):972 # For injected messages we pass the HTTP stacks entirely and directly address the stream.973 try:974 conn = self.connections[event.flow.server_conn]975 except KeyError:976 # We have a miss for the server connection, which means we're looking at a connection object977 # that is tunneled over another connection (for example: over an upstream HTTP proxy).978 # We now take the stream associated with the client connection. That won't work for HTTP/2,979 # but it's good enough for HTTP/1.980 conn = self.connections[event.flow.client_conn]981 if isinstance(conn, HttpStream):982 stream_id = conn.stream_id983 else:984 # We reach to the end of the connection's child stack to get the HTTP/1 client layer,985 # which tells us which stream we are dealing with.986 conn = conn.context.layers[-1]987 assert isinstance(conn, Http1Connection)988 assert conn.stream_id989 stream_id = conn.stream_id990 yield from self.event_to_child(self.streams[stream_id], event)991 elif isinstance(event, events.ConnectionEvent):992 if (993 event.connection == self.context.server994 and self.context.server not in self.connections995 ):996 # We didn't do anything with this connection yet, now the peer is doing something.997 if isinstance(event, events.ConnectionClosed):998 # The peer has closed it - let's close it too!999 yield commands.CloseConnection(event.connection)1000 elif isinstance(event, (events.DataReceived, QuicStreamEvent)):1001 # The peer has sent data or another connection activity occurred.1002 # This can happen with HTTP/2 servers that already send a settings frame.1003 child_layer: HttpConnection1004 if is_h3_alpn(self.context.server.alpn):1005 child_layer = Http3Client(self.context.fork())1006 elif self.context.server.alpn == b"h2":1007 child_layer = Http2Client(self.context.fork())1008 else:1009 child_layer = Http1Client(self.context.fork())1010 self.connections[self.context.server] = child_layer1011 yield from self.event_to_child(child_layer, events.Start())1012 yield from self.event_to_child(child_layer, event)1013 else:1014 raise AssertionError(f"Unexpected event: {event}")1015 else:1016 handler = self.connections[event.connection]1017 yield from self.event_to_child(handler, event)1018 else:1019 raise AssertionError(f"Unexpected event: {event}")1020 1021 def event_to_child(1022 self,1023 child: layer.Layer | HttpStream,1024 event: events.Event,1025 ) -> layer.CommandGenerator[None]:1026 for command in child.handle_event(event):1027 assert isinstance(command, commands.Command)1028 # Streams may yield blocking commands, which ultimately generate CommandCompleted events.1029 # Those need to be routed back to the correct stream, so we need to keep track of that.1030 1031 if command.blocking or isinstance(command, commands.RequestWakeup):1032 self.command_sources[command] = child1033 1034 if isinstance(command, ReceiveHttp):1035 if isinstance(command.event, RequestHeaders):1036 yield from self.make_stream(command.event.stream_id)1037 try:1038 stream = self.streams[command.event.stream_id]1039 except KeyError:1040 # We may be getting data or errors for a stream even though we've already finished handling it,1041 # see for example https://github.com/mitmproxy/mitmproxy/issues/5343.1042 pass1043 else:1044 yield from self.event_to_child(stream, command.event)1045 elif isinstance(command, SendHttp):1046 conn = self.connections[command.connection]1047 yield from self.event_to_child(conn, command.event)1048 elif isinstance(command, DropStream):1049 self.streams.pop(command.stream_id, None)1050 elif isinstance(command, GetHttpConnection):1051 yield from self.get_connection(command)1052 elif isinstance(command, RegisterHttpConnection):1053 yield from self.register_connection(command)1054 elif isinstance(command, commands.OpenConnection):1055 self.connections[command.connection] = child1056 yield command1057 elif isinstance(command, commands.Command):1058 yield command1059 else:1060 raise AssertionError(f"Not a command: {event}")1061 1062 def make_stream(self, stream_id: int) -> layer.CommandGenerator[None]:1063 ctx = self.context.fork()1064 self.streams[stream_id] = HttpStream(ctx, stream_id)1065 yield from self.event_to_child(self.streams[stream_id], events.Start())1066 1067 def get_connection(1068 self, event: GetHttpConnection, *, reuse: bool = True1069 ) -> layer.CommandGenerator[None]:1070 # Do we already have a connection we can re-use?1071 if reuse:1072 for connection in self.connections:1073 connection_suitable = event.connection_spec_matches(connection)1074 if connection_suitable:1075 if connection in self.waiting_for_establishment:1076 self.waiting_for_establishment[connection].append(event)1077 return1078 elif connection.error:1079 stream = self.command_sources.pop(event)1080 yield from self.event_to_child(1081 stream,1082 GetHttpConnectionCompleted(event, (None, connection.error)),1083 )1084 return1085 elif connection.connected:1086 # see "tricky multiplexing edge case" in make_http_connection for an explanation1087 h2_to_h1 = (1088 self.context.client.alpn == b"h2"1089 and connection.alpn != b"h2"1090 )1091 if not h2_to_h1:1092 stream = self.command_sources.pop(event)1093 yield from self.event_to_child(1094 stream,1095 GetHttpConnectionCompleted(event, (connection, None)),1096 )1097 return1098 else:1099 pass # the connection is at least half-closed already, we want a new one.1100 1101 context_connection_matches = (1102 self.context.server not in self.connections1103 and event.connection_spec_matches(self.context.server)1104 )1105 can_use_context_connection = (1106 context_connection_matches and self.context.server.connected1107 )1108 if context_connection_matches and self.context.server.error:1109 stream = self.command_sources.pop(event)1110 yield from self.event_to_child(1111 stream,1112 GetHttpConnectionCompleted(event, (None, self.context.server.error)),1113 )1114 return1115 1116 context = self.context.fork()1117 1118 stack = tunnel.LayerStack()1119 1120 if not can_use_context_connection:1121 context.server = Server(1122 address=event.address, transport_protocol=event.transport_protocol1123 )1124 1125 if event.via:1126 context.server.via = event.via1127 # We always send a CONNECT request, *except* for plaintext absolute-form HTTP requests in upstream mode.1128 send_connect = event.tls or self.mode != HTTPMode.upstream1129 stack /= _upstream_proxy.HttpUpstreamProxy.make(context, send_connect)1130 if event.tls:1131 # Assume that we are in transparent mode and lazily did not open a connection yet.1132 # We don't want the IP (which is the address) as the upstream SNI, but the client's SNI instead.1133 if (1134 self.mode == HTTPMode.transparent1135 and event.address == self.context.server.address1136 ):1137 # reverse proxy mode may set self.context.server.sni, which takes precedence.1138 context.server.sni = (1139 self.context.server.sni1140 or self.context.client.sni1141 or event.address[0]1142 )1143 else:1144 context.server.sni = event.address[0]1145 if context.server.transport_protocol == "tcp":1146 stack /= tls.ServerTLSLayer(context)1147 elif context.server.transport_protocol == "udp":1148 stack /= quic.ServerQuicLayer(context)1149 else:1150 raise AssertionError(1151 context.server.transport_protocol1152 ) # pragma: no cover1153 1154 stack /= HttpClient(context)1155 1156 self.connections[context.server] = stack[0]1157 self.waiting_for_establishment[context.server].append(event)1158 1159 yield from self.event_to_child(stack[0], events.Start())1160 1161 def register_connection(1162 self, command: RegisterHttpConnection1163 ) -> layer.CommandGenerator[None]:1164 waiting = self.waiting_for_establishment.pop(command.connection)1165 1166 reply: tuple[None, str] | tuple[Connection, None]1167 if command.err:1168 reply = (None, command.err)1169 else:1170 reply = (command.connection, None)1171 1172 for cmd in waiting:1173 stream = self.command_sources.pop(cmd)1174 yield from self.event_to_child(1175 stream, GetHttpConnectionCompleted(cmd, reply)1176 )1177 1178 # Tricky multiplexing edge case: Assume we are doing HTTP/2 -> HTTP/1 proxying and the destination server1179 # only serves responses with HTTP read-until-EOF semantics. In this case we can't process two flows on the1180 # same connection. The only workaround left is to open a separate connection for each flow.1181 if (1182 not command.err1183 and self.context.client.alpn == b"h2"1184 and command.connection.alpn != b"h2"1185 ):1186 for cmd in waiting[1:]:1187 yield from self.get_connection(cmd, reuse=False)1188 break1189 1190 1191class HttpClient(layer.Layer):1192 child_layer: layer.Layer1193 1194 @expect(events.Start)1195 def _handle_event(self, event: events.Event) -> layer.CommandGenerator[None]:1196 err: str | None1197 if self.context.server.connected:1198 err = None1199 else:1200 err = yield commands.OpenConnection(self.context.server)