Merge pull request #148 from timvahlbrock/master

feature: support generic keys

mitmproxy2swagger/swagger_util.py

@@ -1,5 +1,6 @@
 # -*- coding: utf-8 -*-
 import urllib
+import uuid
 from typing import Any, List
 
 VERBS = [
@@ -127,6 +128,15 @@ def value_to_schema(value):
         return {"type": "array", "items": value_to_schema(value[0])}
     # check if value is a dict
     elif isinstance(value, dict):
+        all_keys_are_numeric = all(is_numeric_string(key) for key in value)
+        all_keys_are_uuid = all(is_uuid(key) for key in value)
+        keys_are_generic = all_keys_are_numeric or all_keys_are_uuid
+
+        if keys_are_generic and len(value) > 0:
+            return {
+                "type": "object",
+                "additionalProperties": value_to_schema(list(value.values())[0]),
+            }
         return {
             "type": "object",
             "properties": {key: value_to_schema(value[key]) for key in value},
@@ -136,6 +146,22 @@ def value_to_schema(value):
         return {"type": "object"}
 
 
+def is_uuid(key):
+    return isinstance(key, str) and is_valid_uuid(key)
+
+
+def is_numeric_string(key):
+    return isinstance(key, str) and key.isnumeric()
+
+
+def is_valid_uuid(val):
+    try:
+        uuid.UUID(str(val))
+        return True
+    except ValueError:
+        return False
+
+
 MAX_EXAMPLE_ARRAY_ELEMENTS = 10
 MAX_EXAMPLE_OBJECT_PROPERTIES = 150
 

mitmproxy2swagger/test_mitmproxy2swagger.py

@@ -53,6 +53,62 @@ def test_mitmproxy2swagger_generates_swagger_from_mitmproxy_flow_file_with_form_
     )
 
 
+def test_mitmproxy2swagger_generates_swagger_from_mitmproxy_flow_file_with_generic_keys():
+    data = mitmproxy2swagger_e2e_test(
+        "testdata/generic_keys_flows",
+        "http://localhost:8082/",
+        [
+            "--format",
+            "flow",
+        ],
+    )
+    assert data is not None
+
+    assert (
+        get_nested_key(
+            data,
+            "paths./.post.responses.200.content.application/json.schema.properties.numeric.properties",
+        )
+        is None
+    )
+    assert (
+        get_nested_key(
+            data,
+            "paths./.post.responses.200.content.application/json.schema.properties.uuid.properties",
+        )
+        is None
+    )
+    assert (
+        get_nested_key(
+            data,
+            "paths./.post.responses.200.content.application/json.schema.properties.numeric.additionalProperties",
+        )
+        is not None
+    )
+    assert (
+        get_nested_key(
+            data,
+            "paths./.post.responses.200.content.application/json.schema.properties.numeric.additionalProperties",
+        )
+        is not None
+    )
+
+    assert (
+        get_nested_key(
+            data,
+            "paths./.post.responses.200.content.application/json.schema.properties.mixed.properties",
+        )
+        is not None
+    )
+    assert (
+        get_nested_key(
+            data,
+            "paths./.post.responses.200.content.application/json.schema.properties.mixed.additionalProperties",
+        )
+        is None
+    )
+
+
 def test_mitmproxy2swagger_generates_headers_for_flow_files():
     data = mitmproxy2swagger_e2e_test(
         "testdata/form_data_flows",

mitmproxy2swagger/test_openapi_compliance.py

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

testdata/generic_keys_flows

@@ -0,0 +1 @@
+3299:9:websocket;0:~8:response;1105:6:reason;2:OK,11:status_code;3:200#13:timestamp_end;18:1709848351.4360309^15:timestamp_start;17:1709848351.435767^8:trailers;0:~7:content;765:{"numeric": {"1234": {"lorem": "ipsum", "dolor": "sit", "amet": "consectetur"}, "5678": {"lorem": "ipsum", "dolor": "sit", "amet": "consectetur"}, "0000": {"lorem": "ipsum", "dolor": "sit", "amet": "consectetur"}}, "uuid": {"123e4567-e89b-12d3-a456-426614174000": {"lorem": "ipsum", "dolor": "sit", "amet": "consectetur"}, "123e4567-e89b-12d3-a456-426614174001": {"lorem": "ipsum", "dolor": "sit", "amet": "consectetur"}, "123e4567-e89b-12d3-a456-426614174002": {"lorem": "ipsum", "dolor": "sit", "amet": "consectetur"}}, "mixed": {"1234": {"lorem": "ipsum", "dolor": "sit", "amet": "consectetur"}, "123e4567-e89b-12d3-a456-426614174000": {"lorem": "ipsum", "dolor": "sit", "amet": "consectetur"}, "0000": {"lorem": "ipsum", "dolor": "sit", "amet": "consectetur"}}},7:headers;155:39:6:Server,26:BaseHTTP/0.6 Python/3.11.7,]40:4:Date,29:Thu, 07 Mar 2024 21:52:31 GMT,]36:12:Content-type,16:application/json,]24:14:Content-length,3:765,]]12:http_version;8:HTTP/1.0,}7:request;1003: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;17:1709848351.433835^15:timestamp_start;16:1709848351.43342^8:trailers;0:~7:content;532:{"numeric": {"1234": {"lorem": "ipsum", "dolor": "sit", "amet": "consectetur"}, "5678": {"lorem": "ipsum", "dolor": "sit", "amet": "consectetur"}}, "uuid": {"123e4567-e89b-12d3-a456-426614174000": {"lorem": "ipsum", "dolor": "sit", "amet": "consectetur"}, "123e4567-e89b-12d3-a456-426614174001": {"lorem": "ipsum", "dolor": "sit", "amet": "consectetur"}}, "mixed": {"1234": {"lorem": "ipsum", "dolor": "sit", "amet": "consectetur"}, "123e4567-e89b-12d3-a456-426614174000": {"lorem": "ipsum", "dolor": "sit", "amet": "consectetur"}}},7:headers;232:25:4:Host,14:localhost:8082,]40:10:User-Agent,22:python-requests/2.31.0,]36:15:Accept-Encoding,13:gzip, deflate,]15:6:Accept,3:*/*,]28:10:Connection,10:keep-alive,]36:12:Content-Type,16:application/json,]24:14:Content-Length,3:532,]]12:http_version;8:HTTP/1.1,}6:backup;0:~17:timestamp_created;17:1709848351.433495^7:comment;0:;8:metadata;0:}6:marked;0:;9:is_replay;0:~11:intercepted;5:false!11:server_conn;466:3:via;0:~19:timestamp_tcp_setup;18:1709848351.4350889^7:address;19:9:localhost;4:8082#]19:timestamp_tls_setup;0:~13:timestamp_end;17:1709848351.436529^15:timestamp_start;17:1709848351.434129^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:48b3aee0-862d-4ed5-b160-594334eaaa19;8:sockname;20:9:127.0.0.1;5:61640#]8:peername;19:9:127.0.0.1;4:8082#]}11:client_conn;421:10:proxy_mode;7:regular;8:mitmcert;0:~19:timestamp_tls_setup;0:~13:timestamp_end;17:1709848351.436451^15:timestamp_start;17:1709848351.432699^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:50acece3-e5e3-460f-b761-be89aa458880;8:sockname;21:3:::1;4:8080#1:0#1:0#]8:peername;22:3:::1;5:61638#1:0#1:0#]}5:error;0:~2:id;36:14ae0961-3df8-41b0-8ddb-510082083178;4:type;4:http;7:version;2:20#}

testdata/generic_keys_testclient.py

@@ -0,0 +1,51 @@
+# -*- coding: utf-8 -*-
+
+import json
+
+from testclient import testclient
+
+# Sample data
+data = {
+    "numeric": {
+        "1234": {
+            "lorem": "ipsum",
+            "dolor": "sit",
+            "amet": "consectetur",
+        },
+        "5678": {
+            "lorem": "ipsum",
+            "dolor": "sit",
+            "amet": "consectetur",
+        },
+    },
+    "uuid": {
+        "123e4567-e89b-12d3-a456-426614174000": {
+            "lorem": "ipsum",
+            "dolor": "sit",
+            "amet": "consectetur",
+        },
+        "123e4567-e89b-12d3-a456-426614174001": {
+            "lorem": "ipsum",
+            "dolor": "sit",
+            "amet": "consectetur",
+        },
+    },
+    "mixed": {
+        "1234": {
+            "lorem": "ipsum",
+            "dolor": "sit",
+            "amet": "consectetur",
+        },
+        "123e4567-e89b-12d3-a456-426614174000": {
+            "lorem": "ipsum",
+            "dolor": "sit",
+            "amet": "consectetur",
+        },
+    },
+}
+
+testclient(
+    "application/json",
+    lambda: json.dumps(data),
+    lambda content: json.loads(content),
+)

testdata/generic_keys_testserver.py

@@ -0,0 +1,32 @@
+# -*- coding: utf-8 -*-
+import json
+
+from testserver import TestServerHandler, launchServerWith
+
+
+class GenericKeysHandler(TestServerHandler):
+    def transform_data(self, raw_data):
+        data = json.loads(raw_data)
+
+        data["numeric"]["0000"] = {
+            "lorem": "ipsum",
+            "dolor": "sit",
+            "amet": "consectetur",
+        }
+        data["uuid"]["123e4567-e89b-12d3-a456-426614174002"] = {
+            "lorem": "ipsum",
+            "dolor": "sit",
+            "amet": "consectetur",
+        }
+        data["mixed"]["0000"] = {
+            "lorem": "ipsum",
+            "dolor": "sit",
+            "amet": "consectetur",
+        }
+
+        # Encode the modified data
+        return bytes(json.dumps(data), "utf-8")
+
+
+if __name__ == "__main__":
+    launchServerWith(GenericKeysHandler)

testdata/msgpack_testclient.py

@@ -1,25 +1,13 @@
 # -*- coding: utf-8 -*-
 
 import msgpack
-import requests  # type: ignore
+from testclient import testclient
 
 # Sample MessagePack data
 msgpack_data = {"field1": "value1", "field2": "value2"}
 
-
+testclient(
-url = "http://localhost:8082"
+    "application/msgpack",
-headers = {"Content-Type": "application/msgpack"}
+    lambda: msgpack.packb(msgpack_data),
-response = requests.post(
+    lambda content: msgpack.unpackb(content),
-    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

@@ -1,45 +1,18 @@
 # -*- coding: utf-8 -*-
-import http.server
-import socketserver
-
 import msgpack
+from testserver import TestServerHandler, launchServerWith
 
 
-class MessagePackHandler(http.server.BaseHTTPRequestHandler):
+class MessagePackHandler(TestServerHandler):
-    def do_POST(self):
+    def transform_data(self, raw_data):
-        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)
+        return 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
+    launchServerWith(MessagePackHandler)
-
-    with socketserver.TCPServer(("", PORT), MessagePackHandler) as httpd:
-        print(f"Serving on port {PORT}")
-        httpd.serve_forever()

testdata/testclient.py

@@ -0,0 +1,27 @@
+# -*- coding: utf-8 -*-
+from typing import Any, Callable
+
+import requests  # type: ignore
+
+
+def testclient(
+    contentType: str,
+    getData: Callable[[], Any],
+    decodeData: Callable[[Any], Any],
+) -> None:
+    url = "http://localhost:8082"
+    headers = {"Content-Type": contentType}
+    response = requests.post(
+        url,
+        data=getData(),
+        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 = decodeData(response.content)
+    print(data)

testdata/testserver.py

@@ -0,0 +1,40 @@
+# -*- coding: utf-8 -*-
+import http.server
+import socketserver
+from typing import Type
+
+
+class TestServerHandler(http.server.BaseHTTPRequestHandler):
+    def do_POST(self):
+        content_length = int(self.headers["Content-Length"])
+        raw_data = self.rfile.read(content_length)
+
+        try:
+            # Decode received data
+
+            print(raw_data)
+            modified_data = self.transform_data(raw_data)
+
+            # Send the response
+            self.send_response(200)
+            self.send_header("Content-type", self.headers["Content-type"])
+            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())
+
+    def transform_data(self, raw_data):
+        raise NotImplementedError("Subclasses must implement this method")
+
+
+def launchServerWith(handler: Type[TestServerHandler]):
+    PORT = 8082
+    with socketserver.TCPServer(("", PORT), handler) as httpd:
+        print(f"Serving on port {PORT}")
+        httpd.serve_forever()