graphql-engine/server/tests-py/test_allowlist_queries.py

#!/usr/bin/env python3

import pytest
from validate import check_query, check_query_f
from context import PytestConf

usefixtures = pytest.mark.usefixtures

if not PytestConf.config.getoption("--test-allowlist-queries"):
    pytest.skip("flag --test-allowlist-queries is not set. Cannot run tests for allowlist queries", allow_module_level=True)

@pytest.mark.parametrize("transport", ['http', 'websocket'])
@usefixtures('per_class_tests_db_state')
class TestAllowlistQueries:

    def test_query_user(self, hge_ctx, transport):
        check_query_f(hge_ctx, self.dir() + '/query_user.yaml', transport)

    def test_query_user_by_pk(self, hge_ctx, transport):
        check_query_f(hge_ctx, self.dir() + '/query_user_by_pk.yaml', transport)

    def test_query_user_with_typename(self, hge_ctx, transport):
        check_query_f(hge_ctx, self.dir() + '/query_user_with_typename.yaml', transport)

    def test_query_non_allowlist(self, hge_ctx, transport):
        check_query_f(hge_ctx, self.dir() + '/query_non_allowlist.yaml', transport)
    
    def test_query_user_fragment(self, hge_ctx, transport):
        check_query_f(hge_ctx, self.dir() + '/query_user_fragment.yaml', transport)

    def test_query_as_admin(self, hge_ctx, transport):
        check_query_f(hge_ctx, self.dir() + '/query_as_admin.yaml', transport)

    def test_update_query(self, hge_ctx, transport):
        # test only for http
        if transport != 'http':
            pytest.skip("http-only")
        check_query_f(hge_ctx, self.dir() + '/update_query.yaml', transport)
        check_query_f(hge_ctx, self.dir() + '/add_duplicate_query.yaml', transport)

    @classmethod
    def dir(cls):
        return 'queries/graphql_query/allowlist'


def export_metadata(hge_ctx):
    st, resp = hge_ctx.v1metadataq({
        'type': 'export_metadata',
        'version': 2,
        'args': {}
    })
    assert st == 200, resp
    return resp["metadata"]

def assert_allowlist_metadata(hge_ctx, expected):
    metadata = export_metadata(hge_ctx)
    assert metadata.get("allowlist") == expected

def replace_allowlist_metadata_unchecked(hge_ctx, base_metadata, allowlist):
    metadata = dict(base_metadata)
    metadata["allowlist"] = allowlist
    st, resp = hge_ctx.v1metadataq({
        'type': 'replace_metadata',
        'version': 2,
        'args': {
            'metadata': metadata
        }
    })
    return st, resp

def replace_allowlist_metadata(hge_ctx, base_metadata, allowlist):
    st, resp = replace_allowlist_metadata_unchecked(hge_ctx, base_metadata, allowlist)
    assert st == 200, resp

def fail_replace_allowlist_metadata(hge_ctx, base_metadata, allowlist, expected_error):
    st, resp = replace_allowlist_metadata_unchecked(hge_ctx, base_metadata, allowlist)
    assert st == 400
    assert resp == expected_error

def add_collection_to_allowlist_unchecked(hge_ctx, args):
    payload = {
        "type": "add_collection_to_allowlist",
        "args": args,
    }
    return hge_ctx.v1metadataq(payload)
    assert st == 200, resp

def add_collection_to_allowlist(hge_ctx, args):
    st, resp = add_collection_to_allowlist_unchecked(hge_ctx, args)
    assert st == 200, resp

def fail_add_collection_to_allowlist(hge_ctx, args, expected_error):
    st, resp = add_collection_to_allowlist_unchecked(hge_ctx, args)
    assert st == 400
    assert resp == expected_error

def update_scope_of_collection_in_allowlist(hge_ctx, args):
    payload = {
        "type": "update_scope_of_collection_in_allowlist",
        "args": args,
    }
    st, resp = hge_ctx.v1metadataq(payload)
    assert st == 200, resp

def update_scope_of_collection_in_allowlist_unchecked(hge_ctx, args):
    payload = {
        "type": "update_scope_of_collection_in_allowlist",
        "args": args,
    }
    return hge_ctx.v1metadataq(payload)

def update_scope_of_collection_in_allowlist(hge_ctx, args):
    st, resp = update_scope_of_collection_in_allowlist_unchecked(hge_ctx, args)
    assert st == 200, resp

def fail_update_scope_of_collection_in_allowlist(hge_ctx, args, expected_error):
    st, resp = update_scope_of_collection_in_allowlist_unchecked(hge_ctx, args)
    assert st == 400
    assert resp == expected_error

def drop_collection_from_allowlist_unchecked(hge_ctx, args):
    payload = {
        "type": "drop_collection_from_allowlist",
        "args": args
    }
    return hge_ctx.v1metadataq(payload)

def drop_collection_from_allowlist(hge_ctx, args):
    st, resp = drop_collection_from_allowlist_unchecked(hge_ctx, args)
    assert st == 200, resp

def fail_drop_collection_from_allowlist(hge_ctx, args, expected_error):
    st, resp = drop_collection_from_allowlist_unchecked(hge_ctx, args)
    assert st == 400
    assert resp == expected_error

@pytest.fixture(scope="function")
def clean_allowlist(hge_ctx):
    yield
    drop_collection_from_allowlist_unchecked(hge_ctx, {"collection": "collection_1"})
    drop_collection_from_allowlist_unchecked(hge_ctx, {"collection": "collection_2"})

@usefixtures('clean_allowlist', 'per_class_tests_db_state')
class TestAllowlistMetadata:
    @classmethod
    def dir(cls):
        return 'queries/graphql_query/allowlist_role_based'

    def test_add_update_drop(self, hge_ctx):
        # Cycle through add_collection_to_allowlist,
        # update_scope_of_collection_in_allowlist and
        # drop_collection_from_allowlist, verifying
        # state of metadata after each step.

        assert_allowlist_metadata(hge_ctx, None)

        add_collection_to_allowlist(hge_ctx, {
            "collection": "collection_1",
            "scope": {
                "global": True
            }
        })

        assert_allowlist_metadata(hge_ctx, [{
            "collection": "collection_1",
            "scope": {
                "global": True
            }
        }])

        add_collection_to_allowlist(hge_ctx, {
            "collection": "collection_2",
            "scope": {
                "global": False,
                "roles": ["foo", "bar"]
            }
        })

        assert_allowlist_metadata(hge_ctx, [{
            "collection": "collection_1",
            "scope": {
                "global": True
            }
        }, {
            "collection": "collection_2",
            "scope": {
                "global": False,
                "roles": ["foo", "bar"]
            }
        }])

        update_scope_of_collection_in_allowlist(hge_ctx, {
            "collection": "collection_1",
            "scope": {
                "global": False,
                "roles": ["baz"]
            }
        })
        update_scope_of_collection_in_allowlist(hge_ctx, {
            "collection": "collection_2",
            "scope": {
                "global": True
            }
        })

        assert_allowlist_metadata(hge_ctx, [{
            "collection": "collection_1",
            "scope": {
                "global": False,
                "roles": ["baz"]
            }
        }, {
            "collection": "collection_2",
            "scope": {
                "global": True
            }
        }])

        drop_collection_from_allowlist(hge_ctx, {"collection": "collection_1"})

        assert_allowlist_metadata(hge_ctx, [{
            "collection": "collection_2",
            "scope": {
                "global": True
            }
        }])

        drop_collection_from_allowlist(hge_ctx, {"collection": "collection_2"})

        assert_allowlist_metadata(hge_ctx, None)

    def test_add_validation(self, hge_ctx, request):
        # collection required
        fail_add_collection_to_allowlist(hge_ctx, {
            "scope": {
                "global": True,
            }
        }, {
            "path": "$.args",
            "error": 'key "collection" not found',
            "code": "parse-failed",
        })

        # when global is false, roles needs to be present
        fail_add_collection_to_allowlist(hge_ctx, {
            "collection": "collection_1",
            "scope": {
                "global": False,
            }
        }, {
            "path": "$.args.scope",
            "error": 'key "roles" not found',
            "code": "parse-failed",
        })

        # when global is false, roles needs to be non-empty
        fail_add_collection_to_allowlist(hge_ctx, {
            "collection": "collection_1",
            "scope": {
                "global": False,
                "roles": []
            }
        }, {
            "path": "$.args.scope.roles",
            "error": "parsing NonEmpty failed, unexpected empty list",
            "code": "parse-failed",
        })

        # when global is true, there can't be roles
        fail_add_collection_to_allowlist(hge_ctx, {
            "collection": "collection_1",
            "scope": {
                "global": True,
                "roles": ["foo", "bar"],
            }
        }, {
            "path": "$.args.scope",
            "error": "roles are not allowed when global is true",
            "code": "parse-failed",
        })

        # if scope is given, global is required
        fail_add_collection_to_allowlist(hge_ctx, {
            "collection": "collection_1",
            "scope": {}
        }, {
            "path": "$.args.scope",
            "error": 'key "global" not found',
            "code": "parse-failed",
        })

    def test_update_validation(self, hge_ctx):
        # collection required
        fail_update_scope_of_collection_in_allowlist(hge_ctx, {
            "scope": {
                "global": True,
            }
        }, {
            "path": "$.args",
            "error": 'key "collection" not found',
            "code": "parse-failed",
        })

        # scope required
        fail_update_scope_of_collection_in_allowlist(hge_ctx, {
            "collection": "collection_1"
        }, {
            "path": "$.args",
            "error": 'key "scope" not found',
            "code": "parse-failed",
        })

        # we could cover most of the failure cases from test_add_validation

    def test_drop_validation(self, hge_ctx):
        # collection required
        fail_drop_collection_from_allowlist(hge_ctx, {
        }, {
            "path": "$.args",
            "error": "the key 'collection' was not present",
            "code": "parse-failed",
        })

    def test_replace_validation(self, hge_ctx):
        base_metadata = export_metadata(hge_ctx)

        # duplicate collections aren't allowed
        fail_replace_allowlist_metadata(hge_ctx, base_metadata, [{
            "collection": "collection_1",
        }, {
            "collection": "collection_1",
            "scope": {
                "global": True
            }
        }], {
            "path": "$.args.metadata",
            "error": 'multiple declarations exist for the following allowlist entries: collection_1',
            "code": "parse-failed",
        })

    def test_add_scope_defaults_to_global(self, hge_ctx):
        add_collection_to_allowlist(hge_ctx, {
            "collection": "collection_1",
        })

        assert_allowlist_metadata(hge_ctx, [{
            "collection": "collection_1",
            "scope": {
                "global": True
            }
        }])

    def test_drop_fail_when_missing(self, hge_ctx):
        fail_drop_collection_from_allowlist(hge_ctx, {
            "collection": "collection_1"
        }, {
            'path': '$.args',
            'error': 'collection "collection_1" doesn\'t exist in the allowlist',
            'code': 'not-found'
        })

    def test_update_fail_if_missing(self, hge_ctx):
        fail_update_scope_of_collection_in_allowlist(hge_ctx, {
            "collection": "collection_1",
            "scope": {
                "global": True
            }
        }, {
            'path': '$.args',
            'error': 'collection "collection_1" doesn\'t exist in the allowlist',
            'code': 'not-found'
        })

    def test_add_again(self, hge_ctx):
        # re-adding doesn't fail (compare RFC), instead points at change scope
        orig = {
            "collection": "collection_1",
            "scope": {
                "global": True
            }
        }
        add_collection_to_allowlist(hge_ctx, orig)
        assert_allowlist_metadata(hge_ctx, [orig])

        st, resp = add_collection_to_allowlist_unchecked(hge_ctx, {
            "collection": "collection_1",
            "scope": {
                "global": False,
                "roles": ["foo"]
            }
        })
        assert st == 200, resp
        assert resp == {
            "message": 'collection "collection_1" already exists in the allowlist, scope ignored; to change scope, use update_scope_of_collection_in_allowlist'
        }
        assert_allowlist_metadata(hge_ctx, [orig])

        orig2 = {
            "collection": "collection_2",
            "scope": {
                "global": False,
                "roles": ["foo"]
            }
        }
        add_collection_to_allowlist(hge_ctx, orig2)
        assert_allowlist_metadata(hge_ctx, [orig, orig2])

        st, resp = add_collection_to_allowlist_unchecked(hge_ctx, {
            "collection": "collection_2",
            "scope": {
                "global": True
            }
        })
        assert st == 200, resp
        assert resp == {
            "message": 'collection "collection_2" already exists in the allowlist, scope ignored; to change scope, use update_scope_of_collection_in_allowlist'
        }
        assert_allowlist_metadata(hge_ctx, [orig, orig2])


# query_0 is in no collection
query_0 = """
  query {
    author {
      first_name
    }
  }
"""
response_0 = {
    "data": {
        "author": [
            {"first_name": "George"},
            {"first_name": "Paulo"},
        ],
    },
}

# query_1 is only in collection_1
query_1 = """
  query {
    author {
      first_name
      last_name
    }
  }
"""
response_1 = {
    "data": {
        "author": [
            {"first_name": "George", "last_name": "Martin"},
            {"first_name": "Paulo", "last_name": "Coelho"},
        ],
    },
}

# query_2 is in both collections
query_2 = """
  query {
    author {
      __typename
      last_name
    }
  }
"""
response_2 = {
    "data": {
        "author": [
            {"__typename": "author", "last_name": "Martin"},
            {"__typename": "author", "last_name": "Coelho"}
        ]
    }
}

# query_3 is only in collection_2
query_3 = """
  query {
    author {
      id
    }
  }
"""
response_3 = {
    "data": {
        "author": [
            {"id": 1},
            {"id": 2}
        ],
    },
}

response_not_allowed = {
    "errors": [
        {
            "extensions": {
                "path": "$",
                "code": "validation-failed",
            },
            "message": "query is not allowed",
        },
    ],
}

def assert_query_allowed(hge_ctx, role, query, response):
    check_query(hge_ctx, {
        "url": "/v1/graphql",
        "status": 200,
        "headers": {"x-hasura-role": role},
        "response": response,
        "query": {"query": query},
    })

def assert_query_not_allowed(hge_ctx, role, query):
    check_query(hge_ctx, {
        "url": "/v1/graphql",
        "status": 200,
        "headers": {"x-hasura-role": role},
        "response": response_not_allowed,
        "query": {"query": query},
    })

@usefixtures('clean_allowlist', 'per_class_tests_db_state')
class TestRoleBasedAllowlistQueries:
    @classmethod
    def dir(cls):
        return 'queries/graphql_query/allowlist_role_based'

    def test_baseline(self, hge_ctx):
        for role in ["user", "guest"]:
            assert_query_not_allowed(hge_ctx, role, query_0)
            assert_query_not_allowed(hge_ctx, role, query_1)
            assert_query_not_allowed(hge_ctx, role, query_2)
            assert_query_not_allowed(hge_ctx, role, query_3)
        assert_query_allowed(hge_ctx, "admin", query_0, response_0)
        assert_query_allowed(hge_ctx, "admin", query_1, response_1)
        assert_query_allowed(hge_ctx, "admin", query_2, response_2)
        assert_query_allowed(hge_ctx, "admin", query_3, response_3)

    def test_allowlists(self, hge_ctx):
        add_collection_to_allowlist(hge_ctx, {
            "collection": "collection_1",
            "scope": {
                "global": True,
            }
        })
        add_collection_to_allowlist(hge_ctx, {
            "collection": "collection_2",
            "scope": {
                "global": False,
                "roles": [
                    "user",
                    "editor"
                ]
            }
        })

        for role in ["user", "guest"]:
          assert_query_not_allowed(hge_ctx, role, query_0)
          assert_query_allowed(hge_ctx, role, query_1, response_1)
          assert_query_allowed(hge_ctx, role, query_2, response_2)

        assert_query_not_allowed(hge_ctx, "guest", query_3)

        if hge_ctx.pro_tests:
            assert_query_allowed(hge_ctx, "user", query_3, response_3)
        else:
            assert_query_not_allowed(hge_ctx, "user", query_3)