codekingpro/portable-devtools

1"""Reads HAR files into flow objects"""2 3import base644import logging5import time6from datetime import datetime7 8from mitmproxy import connection9from mitmproxy import exceptions10from mitmproxy import http11from mitmproxy.net.http.headers import infer_content_encoding12 13logger = logging.getLogger(__name__)14 15 16def fix_headers(17    request_headers: list[dict[str, str]] | list[tuple[str, str]],18) -> http.Headers:19    """Converts provided headers into (b"header-name", b"header-value") tuples"""20    flow_headers: list[tuple[bytes, bytes]] = []21    for header in request_headers:22        # Applications that use the {"name":item,"value":item} notation are Brave,Chrome,Edge,Firefox,Charles,Fiddler,Insomnia,Safari23        if isinstance(header, dict):24            key = header["name"]25            value = header["value"]26 27        # Application that uses the [name, value] notation is Slack28 29        else:30            try:31                key = header[0]32                value = header[1]33            except IndexError as e:34                raise exceptions.OptionsError(str(e)) from e35        flow_headers.append((key.encode(), value.encode()))36 37    return http.Headers(flow_headers)38 39 40def request_to_flow(request_json: dict) -> http.HTTPFlow:41    """42    Creates a HTTPFlow object from a given entry in HAR file43    """44 45    timestamp_start = datetime.fromisoformat(46        request_json["startedDateTime"].replace("Z", "+00:00")47    ).timestamp()48    timestamp_end = timestamp_start + request_json["time"] / 1000.049    request_method = request_json["request"]["method"]50    request_url = request_json["request"]["url"]51    server_address = request_json.get("serverIPAddress", None)52    request_headers = fix_headers(request_json["request"]["headers"])53 54    http_version_req = request_json["request"]["httpVersion"]55    http_version_resp = request_json["response"]["httpVersion"]56 57    request_content = ""58    # List contains all the representations of an http request across different HAR files59    if request_url.startswith("http://"):60        port = 8061    else:62        port = 44363 64    client_conn = connection.Client(65        peername=("127.0.0.1", 0),66        sockname=("127.0.0.1", 0),67        # TODO Get time info from HAR File68        timestamp_start=time.time(),69    )70 71    if server_address:72        server_conn = connection.Server(address=(server_address, port))73    else:74        server_conn = connection.Server(address=None)75 76    new_flow = http.HTTPFlow(client_conn, server_conn)77 78    if "postData" in request_json["request"]:79        request_content = request_json["request"]["postData"]["text"]80 81    new_flow.request = http.Request.make(82        request_method, request_url, request_content, request_headers83    )84 85    response_code = request_json["response"]["status"]86 87    # In Firefox HAR files images don't include response bodies88    response_content = request_json["response"]["content"].get("text", "")89    content_encoding = request_json["response"]["content"].get("encoding", None)90    response_headers = fix_headers(request_json["response"]["headers"])91 92    if content_encoding == "base64":93        response_content = base64.b64decode(response_content)94    elif isinstance(response_content, str):95        # Convert text to bytes, as in `Response.set_text`96        try:97            response_content = http.encoding.encode(98                response_content,99                (100                    content_encoding101                    or infer_content_encoding(response_headers.get("content-type", ""))102                ),103            )104        except ValueError:105            # Fallback to UTF-8106            response_content = response_content.encode(107                "utf-8", errors="surrogateescape"108            )109 110    # Then encode the content, as in `Response.set_content`111    response_content = http.encoding.encode(112        response_content, response_headers.get("content-encoding") or "identity"113    )114 115    new_flow.response = http.Response(116        b"HTTP/1.1",117        response_code,118        http.status_codes.RESPONSES.get(response_code, "").encode(),119        response_headers,120        response_content,121        None,122        timestamp_start,123        timestamp_end,124    )125 126    # Update timestamps127 128    new_flow.request.timestamp_start = timestamp_start129    new_flow.request.timestamp_end = timestamp_end130 131    new_flow.client_conn.timestamp_start = timestamp_start132    new_flow.client_conn.timestamp_end = timestamp_end133 134    # Update HTTP version135 136    match http_version_req:137        case "http/2.0":138            new_flow.request.http_version = "HTTP/2"139        case "HTTP/2":140            new_flow.request.http_version = "HTTP/2"141        case "HTTP/3":142            new_flow.request.http_version = "HTTP/3"143        case _:144            new_flow.request.http_version = "HTTP/1.1"145    match http_version_resp:146        case "http/2.0":147            new_flow.response.http_version = "HTTP/2"148        case "HTTP/2":149            new_flow.response.http_version = "HTTP/2"150        case "HTTP/3":151            new_flow.response.http_version = "HTTP/3"152        case _:153            new_flow.response.http_version = "HTTP/1.1"154 155    # Remove compression because that may generate different sizes between versions156    new_flow.request.decode()157    new_flow.response.decode()158 159    return new_flow160