codekingpro/portable-devtools
114k
1from __future__ import annotations2 3import asyncio4import functools5import hashlib6import json7import logging8import mimetypes9import os.path10import re11import secrets12import sys13from collections.abc import Callable14from collections.abc import Sequence15from io import BytesIO16from typing import Any17from typing import Awaitable18from typing import ClassVar19from typing import Concatenate20from typing import Literal21from typing import Optional22 23import tornado.escape24import tornado.web25import tornado.websocket26 27import mitmproxy.flow28import mitmproxy.tools.web.master29import mitmproxy_rs30from mitmproxy import certs31from mitmproxy import command32from mitmproxy import contentviews33from mitmproxy import flowfilter34from mitmproxy import http35from mitmproxy import io36from mitmproxy import log37from mitmproxy import optmanager38from mitmproxy import version39from mitmproxy.dns import DNSFlow40from mitmproxy.http import HTTPFlow41from mitmproxy.tcp import TCPFlow42from mitmproxy.tcp import TCPMessage43from mitmproxy.tools.web.webaddons import WebAuth44from mitmproxy.udp import UDPFlow45from mitmproxy.udp import UDPMessage46from mitmproxy.utils import asyncio_utils47from mitmproxy.utils.emoji import emoji48from mitmproxy.utils.strutils import always_str49from mitmproxy.utils.strutils import cut_after_n_lines50from mitmproxy.websocket import WebSocketMessage51 52# Fix for Windows systems where .js files may have text/plain MIME type in registry.53# Modern browsers with ES6 module scripts require proper JavaScript MIME types.54# See: https://github.com/mitmproxy/mitmproxy/issues/797155mimetypes.add_type("text/javascript", ".js")56 57TRANSPARENT_PNG = (58 b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08"59 b"\x04\x00\x00\x00\xb5\x1c\x0c\x02\x00\x00\x00\x0bIDATx\xdac\xfc\xff\x07"60 b"\x00\x02\x00\x01\xfc\xa8Q\rh\x00\x00\x00\x00IEND\xaeB`\x82"61)62 63logger = logging.getLogger(__name__)64 65 66def cert_to_json(certs: Sequence[certs.Cert]) -> dict | None:67 if not certs:68 return None69 cert = certs[0]70 return {71 "keyinfo": cert.keyinfo,72 "sha256": cert.fingerprint().hex(),73 "notbefore": int(cert.notbefore.timestamp()),74 "notafter": int(cert.notafter.timestamp()),75 "serial": str(cert.serial),76 "subject": cert.subject,77 "issuer": cert.issuer,78 "altnames": [str(x.value) for x in cert.altnames],79 }80 81 82def flow_to_json(flow: mitmproxy.flow.Flow) -> dict:83 """84 Remove flow message content and cert to save transmission space.85 Args:86 flow: The original flow.87 Sync with web/src/flow.ts.88 """89 f = {90 "id": flow.id,91 "intercepted": flow.intercepted,92 "is_replay": flow.is_replay,93 "type": flow.type,94 "modified": flow.modified(),95 "marked": emoji.get(flow.marked, "🔴") if flow.marked else "",96 "comment": flow.comment,97 "timestamp_created": flow.timestamp_created,98 }99 100 if flow.client_conn:101 f["client_conn"] = {102 "id": flow.client_conn.id,103 "peername": flow.client_conn.peername,104 "sockname": flow.client_conn.sockname,105 "tls_established": flow.client_conn.tls_established,106 "cert": cert_to_json(flow.client_conn.certificate_list),107 "sni": flow.client_conn.sni,108 "cipher": flow.client_conn.cipher,109 "alpn": always_str(flow.client_conn.alpn, "ascii", "backslashreplace"),110 "tls_version": flow.client_conn.tls_version,111 "timestamp_start": flow.client_conn.timestamp_start,112 "timestamp_tls_setup": flow.client_conn.timestamp_tls_setup,113 "timestamp_end": flow.client_conn.timestamp_end,114 }115 116 if flow.server_conn:117 f["server_conn"] = {118 "id": flow.server_conn.id,119 "peername": flow.server_conn.peername,120 "sockname": flow.server_conn.sockname,121 "address": flow.server_conn.address,122 "tls_established": flow.server_conn.tls_established,123 "cert": cert_to_json(flow.server_conn.certificate_list),124 "sni": flow.server_conn.sni,125 "cipher": flow.server_conn.cipher,126 "alpn": always_str(flow.server_conn.alpn, "ascii", "backslashreplace"),127 "tls_version": flow.server_conn.tls_version,128 "timestamp_start": flow.server_conn.timestamp_start,129 "timestamp_tcp_setup": flow.server_conn.timestamp_tcp_setup,130 "timestamp_tls_setup": flow.server_conn.timestamp_tls_setup,131 "timestamp_end": flow.server_conn.timestamp_end,132 }133 if flow.error:134 f["error"] = flow.error.get_state()135 136 if isinstance(flow, HTTPFlow):137 content_length: int | None138 content_hash: str | None139 140 if flow.request.raw_content is not None:141 content_length = len(flow.request.raw_content)142 content_hash = hashlib.sha256(flow.request.raw_content).hexdigest()143 else:144 content_length = None145 content_hash = None146 f["request"] = {147 "method": flow.request.method,148 "scheme": flow.request.scheme,149 "host": flow.request.host,150 "port": flow.request.port,151 "path": flow.request.path,152 "http_version": flow.request.http_version,153 "headers": tuple(flow.request.headers.items(True)),154 "contentLength": content_length,155 "contentHash": content_hash,156 "timestamp_start": flow.request.timestamp_start,157 "timestamp_end": flow.request.timestamp_end,158 "pretty_host": flow.request.pretty_host,159 }160 if flow.response:161 if flow.response.raw_content is not None:162 content_length = len(flow.response.raw_content)163 content_hash = hashlib.sha256(flow.response.raw_content).hexdigest()164 else:165 content_length = None166 content_hash = None167 f["response"] = {168 "http_version": flow.response.http_version,169 "status_code": flow.response.status_code,170 "reason": flow.response.reason,171 "headers": tuple(flow.response.headers.items(True)),172 "contentLength": content_length,173 "contentHash": content_hash,174 "timestamp_start": flow.response.timestamp_start,175 "timestamp_end": flow.response.timestamp_end,176 }177 if flow.response.data.trailers:178 f["response"]["trailers"] = tuple(179 flow.response.data.trailers.items(True)180 )181 182 if flow.websocket:183 f["websocket"] = {184 "messages_meta": {185 "contentLength": sum(186 len(x.content) for x in flow.websocket.messages187 ),188 "count": len(flow.websocket.messages),189 "timestamp_last": flow.websocket.messages[-1].timestamp190 if flow.websocket.messages191 else None,192 },193 "closed_by_client": flow.websocket.closed_by_client,194 "close_code": flow.websocket.close_code,195 "close_reason": flow.websocket.close_reason,196 "timestamp_end": flow.websocket.timestamp_end,197 }198 elif isinstance(flow, (TCPFlow, UDPFlow)):199 f["messages_meta"] = {200 "contentLength": sum(len(x.content) for x in flow.messages),201 "count": len(flow.messages),202 "timestamp_last": flow.messages[-1].timestamp if flow.messages else None,203 }204 elif isinstance(flow, DNSFlow):205 f["request"] = flow.request.to_json()206 if flow.response:207 f["response"] = flow.response.to_json()208 209 return f210 211 212def logentry_to_json(e: log.LogEntry) -> dict:213 return {214 "id": id(e), # we just need some kind of id.215 "message": e.msg,216 "level": e.level,217 }218 219 220class APIError(tornado.web.HTTPError):221 pass222 223 224class AuthRequestHandler(tornado.web.RequestHandler):225 AUTH_COOKIE_VALUE = b"y"226 227 def __init_subclass__(cls, **kwargs):228 """Automatically wrap all request handlers with `_require_auth`."""229 for method in cls.SUPPORTED_METHODS:230 method = method.lower()231 fn = getattr(cls, method)232 if fn is not tornado.web.RequestHandler._unimplemented_method:233 setattr(cls, method, AuthRequestHandler._require_auth(fn))234 235 def auth_fail(self, invalid_password: bool) -> None:236 """237 Will be called when returning a 403.238 May write a login form as the response.239 """240 241 @staticmethod242 def _require_auth[**P, R](243 fn: Callable[Concatenate[AuthRequestHandler, P], R],244 ) -> Callable[Concatenate[AuthRequestHandler, P], R | None]:245 @functools.wraps(fn)246 def wrapper(247 self: AuthRequestHandler, *args: P.args, **kwargs: P.kwargs248 ) -> R | None:249 if not self.current_user:250 password = ""251 if auth_header := self.request.headers.get("Authorization"):252 auth_scheme, _, auth_params = auth_header.partition(" ")253 if auth_scheme == "Bearer":254 password = auth_params255 256 if not password:257 password = self.get_argument("token", default="")258 259 if not self.settings["is_valid_password"](password):260 self.set_status(403)261 self.auth_fail(bool(password))262 return None263 self.set_signed_cookie(264 self.settings["auth_cookie_name"](),265 self.AUTH_COOKIE_VALUE,266 expires_days=400,267 httponly=True,268 samesite="Strict",269 )270 return fn(self, *args, **kwargs)271 272 return wrapper273 274 def get_current_user(self) -> bool:275 return (276 self.get_signed_cookie(self.settings["auth_cookie_name"](), min_version=2)277 == self.AUTH_COOKIE_VALUE278 )279 280 281class RequestHandler(AuthRequestHandler):282 application: Application283 284 def prepare(self):285 if (286 self.request.method not in ("GET", "HEAD", "OPTIONS")287 and "Sec-Fetch-Site" in self.request.headers288 and self.request.headers["Sec-Fetch-Site"] not in ("same-origin", "none")289 ):290 raise tornado.httpclient.HTTPError(403)291 292 def write(self, chunk: str | bytes | dict | list):293 # Writing arrays on the top level is ok nowadays.294 # http://flask.pocoo.org/docs/0.11/security/#json-security295 if isinstance(chunk, list):296 chunk = tornado.escape.json_encode(chunk)297 self.set_header("Content-Type", "application/json; charset=UTF-8")298 super().write(chunk)299 300 def set_default_headers(self):301 super().set_default_headers()302 self.set_header("Server", version.MITMPROXY)303 self.set_header("X-Frame-Options", "DENY")304 self.add_header("X-XSS-Protection", "1; mode=block")305 self.add_header("X-Content-Type-Options", "nosniff")306 self.add_header(307 "Content-Security-Policy",308 "default-src 'self'; "309 "connect-src 'self' ws:; "310 "img-src 'self' data:; "311 "style-src 'self' 'unsafe-inline'",312 )313 314 @property315 def json(self):316 if not self.request.headers.get("Content-Type", "").startswith(317 "application/json"318 ):319 raise APIError(400, "Invalid Content-Type, expected application/json.")320 try:321 return json.loads(self.request.body.decode())322 except Exception as e:323 raise APIError(400, f"Malformed JSON: {e}")324 325 @property326 def filecontents(self):327 """328 Accept either a multipart/form file upload or just take the plain request body.329 330 """331 if self.request.files:332 return next(iter(self.request.files.values()))[0].body333 else:334 return self.request.body335 336 @property337 def view(self) -> mitmproxy.addons.view.View:338 return self.application.master.view339 340 @property341 def master(self) -> mitmproxy.tools.web.master.WebMaster:342 return self.application.master343 344 @property345 def flow(self) -> mitmproxy.flow.Flow:346 flow_id = str(self.path_kwargs["flow_id"])347 # FIXME: Add a facility to addon.view to safely access the store348 flow = self.view.get_by_id(flow_id)349 if flow:350 return flow351 else:352 raise APIError(404, "Flow not found.")353 354 def write_error(self, status_code: int, **kwargs):355 if "exc_info" in kwargs and isinstance(kwargs["exc_info"][1], APIError):356 self.finish(kwargs["exc_info"][1].log_message)357 else:358 super().write_error(status_code, **kwargs)359 360 361class IndexHandler(RequestHandler):362 def auth_fail(self, invalid_password: bool) -> None:363 self.render("login.html", invalid_password=invalid_password)364 365 def get(self):366 self.render("../index.html")367 368 post = get # login form369 370 371class FilterHelp(RequestHandler):372 def get(self):373 self.write(dict(commands=flowfilter.help))374 375 376class WebSocketEventBroadcaster(tornado.websocket.WebSocketHandler, AuthRequestHandler):377 # raise an error if inherited class doesn't specify its own instance.378 connections: ClassVar[set[WebSocketEventBroadcaster]]379 380 _send_queue: asyncio.Queue[bytes]381 _send_task: asyncio.Task[None]382 383 def prepare(self) -> Optional[Awaitable[None]]:384 token = self.xsrf_token # https://github.com/tornadoweb/tornado/issues/645385 assert token386 return None387 388 def open(self, *args, **kwargs):389 self.connections.add(self)390 self._send_queue = asyncio.Queue()391 # Python 3.13+: use _send_queue.shutdown() and we can use keep_ref=True here.392 self._send_task = asyncio_utils.create_task(393 self.send_task(),394 name="WebSocket send task",395 keep_ref=False,396 )397 398 def on_close(self):399 self.connections.discard(self)400 self._send_task.cancel()401 402 @classmethod403 def broadcast(cls, **kwargs):404 message = cls._json_dumps(kwargs)405 for conn in cls.connections:406 conn.send(message)407 408 def send(self, message: bytes):409 self._send_queue.put_nowait(message)410 411 async def send_task(self):412 while True:413 message = await self._send_queue.get()414 try:415 await self.write_message(message)416 except tornado.websocket.WebSocketClosedError:417 self.on_close()418 419 @staticmethod420 def _json_dumps(d):421 return json.dumps(d, ensure_ascii=False).encode("utf8", "surrogateescape")422 423 424class ClientConnection(WebSocketEventBroadcaster):425 connections: ClassVar[set[ClientConnection]] = set() # type: ignore426 application: Application427 428 def __init__(self, application: Application, request, **kwargs):429 super().__init__(application, request, **kwargs)430 self.filters: dict[str, flowfilter.TFilter] = {} # filters per connection431 432 @classmethod433 def broadcast_flow_reset(cls) -> None:434 for conn in cls.connections:435 conn.send(cls._json_dumps({"type": "flows/reset"}))436 for name, expr in conn.filters.copy().items():437 conn.update_filter(name, expr.pattern)438 439 @classmethod440 def broadcast_flow(441 cls,442 type: Literal["flows/add", "flows/update"],443 f: mitmproxy.flow.Flow,444 ) -> None:445 flow_json = flow_to_json(f)446 for conn in cls.connections:447 conn._broadcast_flow(type, f, flow_json)448 449 def _broadcast_flow(450 self,451 type: Literal["flows/add", "flows/update"],452 f: mitmproxy.flow.Flow,453 flow_json: dict, # Passing the flow_json dictionary to avoid recalculating it for each client454 ) -> None:455 filters = {name: bool(expr(f)) for name, expr in self.filters.items()}456 message = self._json_dumps(457 {458 "type": type,459 "payload": {460 "flow": flow_json,461 "matching_filters": filters,462 },463 },464 )465 self.send(message)466 467 def update_filter(self, name: str, expr: str) -> None:468 if expr:469 filt = flowfilter.parse(expr)470 self.filters[name] = filt471 matching_flow_ids = [f.id for f in self.application.master.view if filt(f)]472 else:473 self.filters.pop(name, None)474 matching_flow_ids = None475 476 message = self._json_dumps(477 {478 "type": "flows/filterUpdate",479 "payload": {480 "name": name,481 "matching_flow_ids": matching_flow_ids,482 },483 },484 )485 self.send(message=message)486 487 async def on_message(self, message: str | bytes):488 try:489 data = json.loads(message)490 match data["type"]:491 case "flows/updateFilter":492 self.update_filter(data["payload"]["name"], data["payload"]["expr"])493 case other:494 raise ValueError(f"Unsupported command: {other}")495 except Exception as e:496 logger.error(f"Error processing message from {self}: {e}")497 self.close(code=1011, reason="Internal server error.")498 499 500class Flows(RequestHandler):501 def get(self):502 self.write([flow_to_json(f) for f in self.view])503 504 505class DumpFlows(RequestHandler):506 def get(self) -> None:507 self.set_header("Content-Disposition", "attachment; filename=flows")508 self.set_header("Content-Type", "application/octet-stream")509 510 match: Callable[[mitmproxy.flow.Flow], bool]511 try:512 match = flowfilter.parse(self.request.arguments["filter"][0].decode())513 except ValueError: # thrown py flowfilter.parse if filter is invalid514 raise APIError(400, f"Invalid filter argument / regex")515 except (516 KeyError,517 IndexError,518 ): # Key+Index: ["filter"][0] can fail, if it's not set519 520 def match(_) -> bool:521 return True522 523 with BytesIO() as bio:524 fw = io.FlowWriter(bio)525 for f in self.view:526 if match(f):527 fw.add(f)528 self.write(bio.getvalue())529 530 async def post(self):531 self.view.clear()532 bio = BytesIO(self.filecontents)533 for f in io.FlowReader(bio).stream():534 await self.master.load_flow(f)535 bio.close()536 537 538class ClearAll(RequestHandler):539 def post(self):540 self.view.clear()541 self.master.events.clear()542 543 544class ResumeFlows(RequestHandler):545 def post(self):546 for f in self.view:547 if not f.intercepted:548 continue549 f.resume()550 self.view.update([f])551 552 553class KillFlows(RequestHandler):554 def post(self):555 for f in self.view:556 if f.killable:557 f.kill()558 self.view.update([f])559 560 561class ResumeFlow(RequestHandler):562 def post(self, flow_id):563 self.flow.resume()564 self.view.update([self.flow])565 566 567class KillFlow(RequestHandler):568 def post(self, flow_id):569 if self.flow.killable:570 self.flow.kill()571 self.view.update([self.flow])572 573 574class FlowHandler(RequestHandler):575 def delete(self, flow_id):576 if self.flow.killable:577 self.flow.kill()578 self.view.remove([self.flow])579 580 def put(self, flow_id) -> None:581 flow: mitmproxy.flow.Flow = self.flow582 flow.backup()583 try:584 for a, b in self.json.items():585 if a == "request" and hasattr(flow, "request"):586 request: mitmproxy.http.Request = flow.request587 for k, v in b.items():588 if k in ["method", "scheme", "host", "path", "http_version"]:589 setattr(request, k, str(v))590 elif k == "port":591 request.port = int(v)592 elif k == "headers":593 request.headers.clear()594 for header in v:595 request.headers.add(*header)596 elif k == "trailers":597 if request.trailers is not None:598 request.trailers.clear()599 else:600 request.trailers = mitmproxy.http.Headers()601 for trailer in v:602 request.trailers.add(*trailer)603 elif k == "content":604 request.text = v605 else:606 raise APIError(400, f"Unknown update request.{k}: {v}")607 608 elif a == "response" and hasattr(flow, "response"):609 response: mitmproxy.http.Response = flow.response610 for k, v in b.items():611 if k in ["msg", "http_version"]:612 setattr(response, k, str(v))613 elif k == "code":614 response.status_code = int(v)615 elif k == "headers":616 response.headers.clear()617 for header in v:618 response.headers.add(*header)619 elif k == "trailers":620 if response.trailers is not None:621 response.trailers.clear()622 else:623 response.trailers = mitmproxy.http.Headers()624 for trailer in v:625 response.trailers.add(*trailer)626 elif k == "content":627 response.text = v628 else:629 raise APIError(400, f"Unknown update response.{k}: {v}")630 elif a == "marked":631 flow.marked = b632 elif a == "comment":633 flow.comment = b634 else:635 raise APIError(400, f"Unknown update {a}: {b}")636 except APIError:637 flow.revert()638 raise639 self.view.update([flow])640 641 642class DuplicateFlow(RequestHandler):643 def post(self, flow_id):644 f = self.flow.copy()645 self.view.add([f])646 self.write(f.id)647 648 649class RevertFlow(RequestHandler):650 def post(self, flow_id):651 if self.flow.modified():652 self.flow.revert()653 self.view.update([self.flow])654 655 656class ReplayFlow(RequestHandler):657 def post(self, flow_id):658 self.master.commands.call("replay.client", [self.flow])659 660 661class FlowContent(RequestHandler):662 def post(self, flow_id, message):663 self.flow.backup()664 message = getattr(self.flow, message)665 message.content = self.filecontents666 self.view.update([self.flow])667 668 def get(self, flow_id, message):669 message = getattr(self.flow, message)670 assert isinstance(self.flow, HTTPFlow)671 672 original_cd = message.headers.get("Content-Disposition", None)673 filename = None674 if original_cd:675 if m := re.search(r'filename=([-\w" .()]+)', original_cd):676 filename = m.group(1)677 if not filename:678 filename = self.flow.request.path.split("?")[0].split("/")[-1]679 680 filename = re.sub(r'[^-\w" .()]', "", filename)681 cd = f"attachment; {filename=!s}"682 self.set_header("Content-Disposition", cd)683 self.set_header("Content-Type", "application/text")684 self.set_header("X-Content-Type-Options", "nosniff")685 self.set_header("X-Frame-Options", "DENY")686 self.write(message.get_content(strict=False))687 688 689class FlowContentView(RequestHandler):690 def message_to_json(691 self,692 view_name: str,693 message: http.Message | TCPMessage | UDPMessage | WebSocketMessage,694 flow: HTTPFlow | TCPFlow | UDPFlow,695 max_lines: int | None = None,696 from_client: bool | None = None,697 timestamp: float | None = None,698 ):699 if view_name and view_name.lower() == "auto":700 view_name = "auto"701 pretty = contentviews.prettify_message(message, flow, view_name=view_name)702 if max_lines:703 pretty.text = cut_after_n_lines(pretty.text, max_lines)704 705 ret: dict[str, Any] = dict(706 text=pretty.text,707 view_name=pretty.view_name,708 syntax_highlight=pretty.syntax_highlight,709 description=pretty.description,710 )711 if from_client is not None:712 ret["from_client"] = from_client713 if timestamp is not None:714 ret["timestamp"] = timestamp715 return ret716 717 def get(self, flow_id, message, content_view) -> None:718 flow = self.flow719 assert isinstance(flow, (HTTPFlow, TCPFlow, UDPFlow))720 721 if self.request.arguments.get("lines"):722 max_lines = int(self.request.arguments["lines"][0])723 else:724 max_lines = None725 726 if message == "messages":727 messages: list[TCPMessage] | list[UDPMessage] | list[WebSocketMessage]728 if isinstance(flow, HTTPFlow) and flow.websocket:729 messages = flow.websocket.messages730 elif isinstance(flow, (TCPFlow, UDPFlow)):731 messages = flow.messages732 else:733 raise APIError(400, f"This flow has no messages.")734 msgs = []735 for m in messages:736 d = self.message_to_json(737 view_name=content_view,738 message=m,739 flow=flow,740 max_lines=max_lines,741 from_client=m.from_client,742 timestamp=m.timestamp,743 )744 msgs.append(d)745 if max_lines:746 max_lines -= d["text"].count("\n") + 1747 assert max_lines is not None748 if max_lines <= 0:749 break750 self.write(msgs)751 else:752 message = getattr(self.flow, message)753 self.write(self.message_to_json(content_view, message, flow, max_lines))754 755 756class Commands(RequestHandler):757 def get(self) -> None:758 commands = {}759 for name, cmd in self.master.commands.commands.items():760 commands[name] = {761 "help": cmd.help,762 "parameters": [763 {764 "name": param.name,765 "type": command.typename(param.type),766 "kind": str(param.kind),767 }768 for param in cmd.parameters769 ],770 "return_type": command.typename(cmd.return_type)771 if cmd.return_type772 else None,773 "signature_help": cmd.signature_help(),774 }775 self.write(commands)776 777 778class ExecuteCommand(RequestHandler):779 def post(self, cmd: str):780 # TODO: We should parse query strings here, this API is painful.781 try:782 args = self.json["arguments"]783 except APIError:784 args = []785 try:786 result = self.master.commands.call_strings(cmd, args)787 except Exception as e:788 self.write({"error": str(e)})789 else:790 self.write(791 {792 "value": result,793 # "type": command.typename(type(result)) if result is not None else "none"794 }795 )796 797 798class Events(RequestHandler):799 def get(self):800 self.write([logentry_to_json(e) for e in self.master.events.data])801 802 803class Options(RequestHandler):804 def get(self):805 self.write(optmanager.dump_dicts(self.master.options))806 807 def put(self):808 update = self.json809 try:810 self.master.options.update(**update)811 except Exception as err:812 raise APIError(400, f"{err}")813 814 815class SaveOptions(RequestHandler):816 def post(self):817 # try:818 # optmanager.save(self.master.options, CONFIG_PATH, True)819 # except Exception as err:820 # raise APIError(400, "{}".format(err))821 pass822 823 824class State(RequestHandler):825 # Separate method for testability.826 @staticmethod827 def get_json(master: mitmproxy.tools.web.master.WebMaster):828 return {829 "version": version.VERSION,830 "contentViews": [831 v for v in contentviews.registry.available_views() if v != "query"832 ],833 "servers": {834 s.mode.full_spec: s.to_json() for s in master.proxyserver.servers835 },836 "platform": sys.platform,837 "localModeUnavailable": mitmproxy_rs.local.LocalRedirector.unavailable_reason(),838 }839 840 def get(self):841 self.write(State.get_json(self.master))842 843 844class ProcessList(RequestHandler):845 @staticmethod846 def get_json():847 processes = mitmproxy_rs.process_info.active_executables()848 return [849 {850 "is_visible": process.is_visible,851 "executable": str(process.executable),852 "is_system": process.is_system,853 "display_name": process.display_name,854 }855 for process in processes856 ]857 858 def get(self):859 self.write(ProcessList.get_json())860 861 862class ProcessImage(RequestHandler):863 def get(self):864 path = self.get_query_argument("path", None)865 866 if not path:867 raise APIError(400, "Missing 'path' parameter.")868 869 try:870 icon_bytes = mitmproxy_rs.process_info.executable_icon(path)871 except Exception:872 icon_bytes = TRANSPARENT_PNG873 874 self.set_header("Content-Type", "image/png")875 self.set_header("X-Content-Type-Options", "nosniff")876 self.set_header("Cache-Control", "max-age=604800")877 self.write(icon_bytes)878 879 880class GZipContentAndFlowFiles(tornado.web.GZipContentEncoding):881 CONTENT_TYPES = {882 "application/octet-stream",883 *tornado.web.GZipContentEncoding.CONTENT_TYPES,884 }885 886 887handlers = [888 (r"/", IndexHandler),889 (r"/filter-help(?:\.json)?", FilterHelp),890 (r"/updates", ClientConnection),891 (r"/commands(?:\.json)?", Commands),892 (r"/commands/(?P<cmd>[a-z.]+)", ExecuteCommand),893 (r"/events(?:\.json)?", Events),894 (r"/flows(?:\.json)?", Flows),895 (r"/flows/dump", DumpFlows),896 (r"/flows/resume", ResumeFlows),897 (r"/flows/kill", KillFlows),898 (r"/flows/(?P<flow_id>[0-9a-f\-]+)", FlowHandler),899 (r"/flows/(?P<flow_id>[0-9a-f\-]+)/resume", ResumeFlow),900 (r"/flows/(?P<flow_id>[0-9a-f\-]+)/kill", KillFlow),901 (r"/flows/(?P<flow_id>[0-9a-f\-]+)/duplicate", DuplicateFlow),902 (r"/flows/(?P<flow_id>[0-9a-f\-]+)/replay", ReplayFlow),903 (r"/flows/(?P<flow_id>[0-9a-f\-]+)/revert", RevertFlow),904 (r"/flows/(?P<flow_id>[0-9a-f\-]+)/(?P<message>request|response|messages)/content.data", FlowContent),905 (r"/flows/(?P<flow_id>[0-9a-f\-]+)/(?P<message>request|response|messages)/content/(?P<content_view>[0-9a-zA-Z\-\_%]+)(?:\.json)?", FlowContentView),906 (r"/clear", ClearAll),907 (r"/options(?:\.json)?", Options),908 (r"/options/save", SaveOptions),909 (r"/state(?:\.json)?", State),910 (r"/processes", ProcessList),911 (r"/executable-icon", ProcessImage),912] # fmt: skip913 914 915class Application(tornado.web.Application):916 master: mitmproxy.tools.web.master.WebMaster917 918 def __init__(919 self, master: mitmproxy.tools.web.master.WebMaster, debug: bool920 ) -> None:921 self.master = master922 auth_addon: WebAuth = master.addons.get("webauth")923 super().__init__(924 handlers=handlers, # type: ignore # https://github.com/tornadoweb/tornado/pull/3455925 template_path=os.path.join(os.path.dirname(__file__), "templates"),926 static_path=os.path.join(os.path.dirname(__file__), "static"),927 xsrf_cookies=True,928 # https://github.com/mitmproxy/mitmproxy/issues/8194929 # 2026-05: We can move back to the default cookie name in a few years.930 xsrf_cookie_name="_mitmproxy_xsrf",931 xsrf_cookie_kwargs=dict(samesite="Strict"),932 cookie_secret=secrets.token_bytes(32),933 debug=debug,934 autoreload=False,935 transforms=[GZipContentAndFlowFiles],936 is_valid_password=auth_addon.is_valid_password,937 auth_cookie_name=auth_addon.auth_cookie_name,938 compiled_template_cache=False, # Vite939 )940