GraphQL content-view (#4396)

* Add GraphQL contentview

* Simpler graphql formatting

* Do not prompt for exit

* Give graphql priority over json

* Revert "Do not prompt for exit"

This reverts commit 386b38f9eb4601ae41365ace5b9689c30761a412.

* Handle batched queries

* Fix flake issues

* typing fix

* Typing fixes

* Add graphql content view test

* Fix reference

* Fixed tests

* flake

* Implement GraphQLView.render_priority

* Show query key stub value

* Code coverage

* Import PARSE_ERROR

* Test for invalid json

mitmproxy/contentviews/__init__.py

@@ -20,7 +20,7 @@ from mitmproxy.net import http
 from mitmproxy.utils import strutils
 from . import (
     auto, raw, hex, json, xml_html, wbxml, javascript, css,
-    urlencoded, multipart, image, query, protobuf, msgpack
+    urlencoded, multipart, image, query, protobuf, msgpack, graphql
 )
 from .base import View, KEY_MAX, format_text, format_dict, TViewResult
 from ..http import HTTPFlow
@@ -175,6 +175,7 @@ def get_content_view(
 add(auto.ViewAuto())
 add(raw.ViewRaw())
 add(hex.ViewHex())
+add(graphql.ViewGraphQL())
 add(json.ViewJSON())
 add(xml_html.ViewXmlHtml())
 add(wbxml.ViewWBXML())

mitmproxy/contentviews/graphql.py

@@ -0,0 +1,57 @@
+import json
+
+import typing
+
+from mitmproxy.contentviews import base
+from mitmproxy.contentviews.json import parse_json, PARSE_ERROR
+
+
+def format_graphql(data):
+    query = data["query"]
+    header_data = data.copy()
+    header_data["query"] = "..."
+    return """{header}
+---
+{query}
+""".format(header=json.dumps(header_data, indent=2), query = query)
+
+
+def format_query_list(data: typing.List[typing.Any]):
+    num_queries = len(data) - 1
+    result = ""
+    for i, op in enumerate(data):
+        result += "--- {i}/{num_queries}\n".format(i=i, num_queries=num_queries)
+        result += format_graphql(op)
+    return result
+
+
+def is_graphql_query(data):
+    return isinstance(data, dict) and "query" in data and "\n" in data["query"]
+
+
+def is_graphql_batch_query(data):
+    return isinstance(data, list) and isinstance(data[0], dict) and "query" in data[0]
+
+
+class ViewGraphQL(base.View):
+    name = "GraphQL"
+
+    def __call__(self, data, **metadata):
+        data = parse_json(data)
+        if data is not PARSE_ERROR:
+            if is_graphql_query(data):
+                return "GraphQL", base.format_text(format_graphql(data))
+            elif is_graphql_batch_query(data):
+                return "GraphQL", base.format_text(format_query_list(data))
+
+    def render_priority(self, data: bytes, *, content_type: typing.Optional[str] = None, **metadata) -> float:
+        if content_type != "application/json":
+            return 0
+
+        data = parse_json(data)
+
+        if data is not PARSE_ERROR:
+            if is_graphql_query(data) or is_graphql_batch_query(data):
+                return 2
+
+        return 0

mitmproxy/contentviews/json.py

@@ -1,5 +1,6 @@
 import re
 import json
+from functools import lru_cache
 
 import typing
 
@@ -8,6 +9,7 @@ from mitmproxy.contentviews import base
 PARSE_ERROR = object()
 
 
+@lru_cache(1)
 def parse_json(s: bytes) -> typing.Any:
     try:
         return json.loads(s.decode('utf-8'))

test/mitmproxy/contentviews/test_graphql.py

@@ -0,0 +1,34 @@
+from hypothesis import given
+from hypothesis.strategies import binary
+
+from mitmproxy.contentviews import graphql
+from . import full_eval
+
+
+def test_render_priority():
+    v = graphql.ViewGraphQL()
+    assert 2 == v.render_priority(b"""{"query": "query P { \\n }"}""", content_type="application/json")
+    assert 2 == v.render_priority(b"""[{"query": "query P { \\n }"}]""", content_type="application/json")
+    assert 0 == v.render_priority(b"""[{"query": "query P { \\n }"}]""", content_type="text/html")
+    assert 0 == v.render_priority(b"""[{"xquery": "query P { \\n }"}]""", content_type="application/json")
+    assert 0 == v.render_priority(b"}", content_type="application/json")
+
+
+def test_format_graphql():
+    assert graphql.format_graphql({"query": "query P { \\n }"})
+
+
+def test_format_query_list():
+    assert graphql.format_query_list([{"query": "query P { \\n }"}])
+
+
+def test_view_graphql():
+    v = graphql.ViewGraphQL()
+    assert v(b"""{"query": "query P { \\n }"}""", content_type="application/json")
+    assert v(b"""[{"query": "query P { \\n }"}]""", content_type="application/json")
+
+
+@given(binary())
+def test_view_graphql_doesnt_crash(data):
+    v = full_eval(graphql.ViewGraphQL())
+    v(data)