Add tests for msgpack parsing

Also checked in scripts used to generate test data

mitmproxy2swagger/test_mitmproxy2swagger.py

@@ -68,3 +68,38 @@ def test_mitmproxy2swagger_generates_headers_for_flow_files():
         get_nested_key(data, "paths./post.post.responses.200.headers.content-type")
         is not None
     )
+
+
+def test_mitmproxy2swagger_parses_msgpack_requests_and_responses():
+    data = mitmproxy2swagger_e2e_test(
+        "testdata/msgpack_flows",
+        "http://localhost:8082/",
+        [
+            "--format",
+            "flow",
+        ],
+    )
+    assert data is not None
+
+    assert (
+        get_nested_key(data, "paths./.post.responses.200.content.application/msgpack")
+        is not None
+    )
+    assert (
+        get_nested_key(
+            data,
+            "paths./.post.responses.200.content.application/msgpack.schema.properties.new_field.type",
+        )
+        == "string"
+    )
+    assert (
+        get_nested_key(data, "paths./.post.requestBody.content.application/msgpack")
+        is not None
+    )
+    assert (
+        get_nested_key(
+            data,
+            "paths./.post.requestBody.content.application/msgpack.schema.properties.field1.type",
+        )
+        == "string"
+    )

mitmproxy2swagger/test_openapi_compliance.py

@@ -57,3 +57,17 @@ def test_mitmproxy2swagger_compliance_from_form_data_file_with_headers():
     )
     assert data is not None
     validate_spec(data)
+
+
+def test_mitmproxy2swagger_compliance_from_msgpack_file_with_headers():
+    data = mitmproxy2swagger_e2e_test(
+        "testdata/msgpack_flows",
+        "http://localhost:8082/",
+        [
+            "--format",
+            "flow",
+            "--headers",
+        ],
+    )
+    assert data is not None
+    validate_spec(data)

testdata/msgpack_flows

@@ -0,0 +1 @@
+2099:9:websocket;0:~8:response;393:6:reason;2:OK,11:status_code;3:200#13:timestamp_end;18:1704284722.2074344^15:timestamp_start;18:1704284722.2070074^8:trailers;0:~7:content;51:ƒ¦field1¦value1¦field2¦value2©new_field«Added Field,7:headers;157:39:6:Server,26:BaseHTTP/0.6 Python/3.11.6,]40:4:Date,29:Wed, 03 Jan 2024 12:25:22 GMT,]39:12:Content-type,19:application/msgpack,]23:14:Content-length,2:51,]]12:http_version;8:HTTP/1.0,}7:request;514:4:path;1:/,9:authority;0:,6:scheme;4:http,6:method;4:POST,4:port;4:8082#4:host;9:localhost;13:timestamp_end;18:1704284722.2042184^15:timestamp_start;18:1704284722.2037127^8:trailers;0:~7:content;29:‚¦field1¦value1¦field2¦value2,7:headers;244:25:4:Host,14:localhost:8082,]40:10:User-Agent,22:python-requests/2.31.0,]46:15:Accept-Encoding,23:gzip, deflate, br, zstd,]15:6:Accept,3:*/*,]28:10:Connection,10:keep-alive,]39:12:Content-Type,19:application/msgpack,]23:14:Content-Length,2:29,]]12:http_version;8:HTTP/1.1,}6:backup;0:~17:timestamp_created;17:1704284722.203833^7:comment;0:;8:metadata;0:}6:marked;0:;9:is_replay;0:~11:intercepted;5:false!11:server_conn;467:3:via;0:~19:timestamp_tcp_setup;18:1704284722.2062707^7:address;19:9:localhost;4:8082#]19:timestamp_tls_setup;0:~13:timestamp_end;17:1704284722.208372^15:timestamp_start;18:1704284722.2047434^3:sni;0:~11:tls_version;0:~11:cipher_list;0:]6:cipher;0:~11:alpn_offers;0:]4:alpn;0:~16:certificate_list;0:]3:tls;5:false!5:error;0:~18:transport_protocol;3:tcp;2:id;36:0d4b5407-f87c-4a56-b1be-0eb43eef8855;8:sockname;20:9:127.0.0.1;5:47522#]8:peername;19:9:127.0.0.1;4:8082#]}11:client_conn;423:10:proxy_mode;7:regular;8:mitmcert;0:~19:timestamp_tls_setup;0:~13:timestamp_end;18:1704284722.2085643^15:timestamp_start;18:1704284722.2027926^3:sni;0:~11:tls_version;0:~11:cipher_list;0:]6:cipher;0:~11:alpn_offers;0:]4:alpn;0:~16:certificate_list;0:]3:tls;5:false!5:error;0:~18:transport_protocol;3:tcp;2:id;36:d298f071-2025-4766-b64d-6a868949b78c;8:sockname;21:3:::1;4:8080#1:0#1:0#]8:peername;22:3:::1;5:60670#1:0#1:0#]}5:error;0:~2:id;36:acc4fde4-062c-4fbd-b7a8-5a46d3a0bcfc;4:type;4:http;7:version;2:20#}

testdata/msgpack_testclient.py

@@ -0,0 +1,25 @@
+# -*- coding: utf-8 -*-
+
+import msgpack
+import requests  # type: ignore
+
+# Sample MessagePack data
+msgpack_data = {"field1": "value1", "field2": "value2"}
+
+
+url = "http://localhost:8082"
+headers = {"Content-Type": "application/msgpack"}
+response = requests.post(
+    url,
+    data=msgpack.packb(msgpack_data, use_bin_type=True),
+    headers=headers,
+    proxies={"http": "http://localhost:8080", "https": "http://localhost:8080"},
+)
+
+# Print the response
+print(response.status_code)
+print(response.headers)
+
+# convert the response data from MessagePack to JSON
+data = msgpack.unpackb(response.content, raw=False)
+print(data)

testdata/msgpack_testserver.py

@@ -0,0 +1,45 @@
+# -*- coding: utf-8 -*-
+import http.server
+import socketserver
+
+import msgpack
+
+
+class MessagePackHandler(http.server.BaseHTTPRequestHandler):
+    def do_POST(self):
+        content_length = int(self.headers["Content-Length"])
+        raw_data = self.rfile.read(content_length)
+
+        try:
+            # Decode received MessagePack data
+
+            print(raw_data)
+            data = msgpack.unpackb(raw_data, raw=False)
+
+            # Add a new field to the data
+            data["new_field"] = "Added Field"
+
+            # Encode the modified data as MessagePack
+            modified_data = msgpack.packb(data, use_bin_type=True)
+
+            # Send the response
+            self.send_response(200)
+            self.send_header("Content-type", "application/msgpack")
+            self.send_header("Content-length", len(modified_data))
+            self.end_headers()
+            self.wfile.write(modified_data)
+
+        except Exception as e:
+            print(f"Error processing request: {str(e)}")
+            self.send_response(500)
+            self.send_header("Content-type", "text/plain")
+            self.end_headers()
+            self.wfile.write(f"Error processing request: {str(e)}".encode())
+
+
+if __name__ == "__main__":
+    PORT = 8082
+
+    with socketserver.TCPServer(("", PORT), MessagePackHandler) as httpd:
+        print(f"Serving on port {PORT}")
+        httpd.serve_forever()