server/tests-py: Parallelize JWT tests.

This rewrites the JWT tests to generate and specify the secrets per test class, and to provide the server configuration to the HGE fixture.

It covers the tests in:

  - *test_jwt.py*
  - *test_jwt_claims_map.py*
  - *test_config_api.py*
  - *test_graphql_queries.py* (just a couple here)

This does reduce the number of code paths exercised with JWT, as we were previously running *all* tests with JWT tokens. However, this seems excessive; we don't need to tread every code path, just enough to ensure we handle the tokens appropriately. I believe that the test coverage in *test_jwt.py* does this well enough (though I'd prefer if we moved the coverage lower down in the stack as unit tests).

These tests were configured in multiple different ways by *test-server.sh*; this configuration is now moved to test subclasses within the various files. This results in a bit of duplication.

Unfortunately, the tests would ideally use parameterization rather than subclassing, but that doesn't work because of `hge_fixture_env`, which creates a "soft" dependency between the environment variables and `hge_server`. Parameterizing the former *should* force the latter to be recreated for each new set of environment variables, but `hge_server` isn't actually aware there's a dependency.

It currently looks like this adds lines of code; we'll more than make up for it when we delete the relevant lines from *test-server.sh*. I am not doing that here because I plan on deleting the whole file in a subsequent changeset.

[NDAT-538]: https://hasurahq.atlassian.net/browse/NDAT-538?atlOrigin=eyJpIjoiNWRkNTljNzYxNjVmNDY3MDlhMDU5Y2ZhYzA5YTRkZjUiLCJwIjoiZ2l0aHViLWNvbS1KU1cifQ

PR-URL: https://github.com/hasura/graphql-engine-mono/pull/8803
GitOrigin-RevId: f7f2caa62de0b0a45e42964b69a8ae73d1575fe8

.circleci/server-test-names.txt

@@ -18,18 +18,18 @@
 # admin-secret
 # admin-secret-unauthorized-role
 # read-only-db
-jwt-rs512
-jwt-ed25519
-jwt-stringified
-jwt-audience-check-single-string
-jwt-audience-check-list-string
-jwt-issuer-check
-jwt-with-claims-namespace-path
-jwt-claims-map-with-json-path-values
-jwt-claims-map-with-literal-values
-jwt-with-expiry-time-leeway
-jwt-cookie
-jwt-cookie-unauthorized-role
+# jwt-rs512
+# jwt-ed25519
+# jwt-stringified
+# jwt-audience-check-single-string
+# jwt-audience-check-list-string
+# jwt-issuer-check
+# jwt-with-claims-namespace-path
+# jwt-claims-map-with-json-path-values
+# jwt-claims-map-with-literal-values
+# jwt-with-expiry-time-leeway
+# jwt-cookie
+# jwt-cookie-unauthorized-role
 # cors-domains
 # auth-webhook-cookie
 # ws-init-cookie-read-cors-enabled

server/tests-py/conftest.py

@@ -1,5 +1,6 @@
 import collections
 import http.server
+import json
 import inspect
 import os
 import pytest
@@ -16,6 +17,7 @@ import uuid
 import auth_webhook_server
 from context import ActionsWebhookServer, EvtsWebhookServer, GQLWsClient, GraphQLWSClient, HGECtx, HGECtxGQLServer, HGECtxWebhook, PytestConf
 import fixtures.hge
+import fixtures.jwt
 import fixtures.postgres
 import fixtures.tls
 import jwk_server
@@ -47,13 +49,6 @@ def pytest_addoption(parser):
     parser.addoption('--tls-ca-cert', help='The CA certificate used for helper services', required=False)
     parser.addoption('--tls-ca-key', help='The CA key used for helper services', required=False)
 
-    parser.addoption(
-        "--hge-jwt-key-file", metavar="HGE_JWT_KEY_FILE", help="File containing the private key used to encode jwt tokens using RS512 algorithm", required=False
-    )
-    parser.addoption(
-        "--hge-jwt-conf", metavar="HGE_JWT_CONF", help="The JWT conf", required=False
-    )
-
     parser.addoption(
         "--test-hge-scale-url",
         metavar="<url>",
@@ -115,18 +110,6 @@ This option may result in test failures if the schema has to change between the
         help="When used along with collect-only, it will write the list of upgrade tests into the file specified"
     )
 
-    parser.addoption(
-        "--test-unauthorized-role",
-        action="store_true",
-        help="Run testcases for unauthorized role",
-    )
-
-    parser.addoption(
-        "--test-no-cookie-and-unauth-role",
-        action="store_true",
-        help="Run testcases for no unauthorized role and no cookie jwt header set (cookie auth is set as part of jwt config upon engine startup)",
-    )
-
     parser.addoption(
         "--redis-url",
         metavar="REDIS_URL",
@@ -710,6 +693,35 @@ def ws_client_graphql_ws(hge_ctx):
     yield client
     client.teardown()
 
+@pytest.fixture(scope='class')
+@pytest.mark.early
+def jwt_configuration(
+    request: pytest.FixtureRequest,
+    tmp_path_factory: pytest.TempPathFactory,
+    hge_fixture_env: dict[str, str],
+) -> Optional[fixtures.jwt.JWTConfiguration]:
+    marker = request.node.get_closest_marker('jwt')
+    if not marker:
+        raise Exception('JWT configuration is required.')
+
+    algorithm = marker.args[0]
+    try:
+        configuration = marker.args[1]
+    except IndexError:
+        configuration = {}
+
+    tmp_path = tmp_path_factory.mktemp('jwt')
+    match algorithm:
+        case 'rsa':
+            configuration = fixtures.jwt.init_rsa(tmp_path, configuration)
+        case 'ed25519':
+            configuration = fixtures.jwt.init_ed25519(tmp_path, configuration)
+        case _:
+            raise Exception(f'Unsupported JWT configuration: {marker.args!r}')
+
+    hge_fixture_env['HASURA_GRAPHQL_JWT_SECRET'] = json.dumps(configuration.server_configuration)
+    return configuration
+
 @pytest.fixture(scope='class')
 @pytest.mark.early
 def jwk_server_url(request: pytest.FixtureRequest, hge_fixture_env: dict[str, str]):

server/tests-py/context.py

@@ -838,18 +838,6 @@ class HGECtx:
         self.metadata_schema_url = metadata_schema_url
         self.hge_key = hge_key
         self.webhook = webhook
-        hge_jwt_key_file = config.getoption('--hge-jwt-key-file')
-        if hge_jwt_key_file is None:
-            self.hge_jwt_key = None
-        else:
-            with open(hge_jwt_key_file) as f:
-                self.hge_jwt_key = f.read()
-        self.hge_jwt_conf = config.getoption('--hge-jwt-conf')
-        if self.hge_jwt_conf is not None:
-            self.hge_jwt_conf_dict = json.loads(self.hge_jwt_conf)
-            self.hge_jwt_algo = self.hge_jwt_conf_dict["type"]
-            if self.hge_jwt_algo == "Ed25519":
-                self.hge_jwt_algo = "EdDSA"
         self.may_skip_test_teardown = False
 
         # This will be GC'd, but we also explicitly dispose() in teardown()

server/tests-py/fixtures/jwt.py

@@ -0,0 +1,60 @@
+import pathlib
+from typing import Any, NamedTuple
+import subprocess
+
+
+class JWTConfiguration(NamedTuple):
+    private_key_file: pathlib.Path
+    public_key_file: pathlib.Path
+    private_key: str
+    public_key: str
+    algorithm: str
+    server_configuration: dict
+
+
+def init_rsa(tmp_path: pathlib.Path, configuration: Any) -> JWTConfiguration:
+    private_key_file = tmp_path / 'private.key'
+    public_key_file = tmp_path / 'public.key'
+    subprocess.run(['openssl', 'genrsa', '-out', private_key_file, '2048'], check=True, capture_output=True)
+    subprocess.run(['openssl', 'rsa', '-pubout', '-in', private_key_file, '-out', public_key_file], check=True, capture_output=True)
+    with open(private_key_file) as f:
+        private_key = f.read()
+    with open(public_key_file) as f:
+        public_key = f.read()
+    server_configuration = {
+        'type': 'RS512',
+        'key': public_key,
+        **configuration,
+    }
+    return JWTConfiguration(
+        private_key_file = private_key_file,
+        public_key_file = public_key_file,
+        private_key = private_key,
+        public_key = public_key,
+        algorithm = 'RS512',
+        server_configuration = server_configuration,
+    )
+
+
+def init_ed25519(tmp_path: pathlib.Path, configuration: Any) -> JWTConfiguration:
+    private_key_file = tmp_path / 'private.key'
+    public_key_file = tmp_path / 'public.key'
+    subprocess.run(['openssl', 'genpkey', '-algorithm', 'ed25519', '-outform', 'PEM', '-out', private_key_file], check=True, capture_output=True)
+    subprocess.run(['openssl', 'pkey', '-pubout', '-in', private_key_file, '-out', public_key_file], check=True, capture_output=True)
+    with open(private_key_file) as f:
+        private_key = f.read()
+    with open(public_key_file) as f:
+        public_key = f.read()
+    server_configuration = {
+        'type': 'Ed25519',
+        'key': public_key,
+        **configuration,
+    }
+    return JWTConfiguration(
+        private_key_file = private_key_file,
+        public_key_file = public_key_file,
+        private_key = private_key,
+        public_key = public_key,
+        algorithm = 'EdDSA',
+        server_configuration = server_configuration,
+    )

server/tests-py/pytest.ini

@@ -8,8 +8,10 @@ markers =
   backend: The backends supported by the test case
   admin_secret: Generate and use an admin secret
   no_admin_secret: Skip if an admin secret is provided (legacy)
+  requires_an_admin_secret: Skip if no admin secret is provided
   hge_env: Pass additional environment variables to the GraphQL Engine
   jwk_path: When running a JWK server, the URL path that HGE should use
+  jwt: JWT configuration for the test
   tls_webhook_server: Only run the webhook server with TLS enabled
   no_tls_webhook_server: Only run the webhook server with TLS disabled
   tls_insecure_certificate: Create an insecure (self-signed) certificate for the webhook server

server/tests-py/test_config_api.py

@@ -1,53 +1,183 @@
-import json
+import pytest
 
-class TestConfigAPI:
-    def test_config_api_user_role_error(self, hge_ctx, hge_key):
-        headers = {'x-hasura-role': 'user'}
-        if hge_key is not None:
-            headers['x-hasura-admin-secret'] = hge_key
 
+class TestConfigApiWithAnInsecureServer:
+    def test_responds_correctly(self, hge_ctx):
+        headers = {
+            'x-hasura-role': 'admin',
+        }
         resp = hge_ctx.http.get(hge_ctx.hge_url + '/v1alpha1/config', headers=headers)
-
-        assert resp.status_code == 400, resp
-
-    def test_config_api(self, hge_ctx, hge_key):
-        jwt_conf = hge_ctx.hge_jwt_conf
-        if jwt_conf is not None:
-            jwt_conf_dict = json.loads(hge_ctx.hge_jwt_conf)
-        else:
-            jwt_conf_dict = None
-
-        headers = {'x-hasura-role': 'admin'}
-        if hge_key is not None:
-            headers['x-hasura-admin-secret'] = hge_key
-
-        resp = hge_ctx.http.get(hge_ctx.hge_url + '/v1alpha1/config', headers=headers)
-
         assert resp.status_code == 200, resp
 
         body = resp.json()
-        assert body['is_admin_secret_set'] == (hge_key is not None)
-        assert body['is_auth_hook_set'] == (hge_ctx.webhook is not None)
-        assert body['is_jwt_set'] == (jwt_conf is not None)
+        print('Body:', body)
+        assert body['is_admin_secret_set'] == False
+        assert body['is_auth_hook_set'] == False
+        assert body['is_jwt_set'] == False
+        assert body['jwt'] == []
 
-        if jwt_conf_dict:
-            claims_format = "json"
-            if 'claims_namespace_path' in jwt_conf_dict:
-                assert body['jwt']['claims_namespace_path'] == jwt_conf_dict['claims_namespace_path']
-                assert body['jwt']['claims_format'] == claims_format
-            else:
-                claims_namespace = "https://hasura.io/jwt/claims"
-                if 'claims_namespace' in jwt_conf_dict:
-                    claims_namespace = jwt_conf_dict['claims_namespace']
-                if 'claims_format' in jwt_conf_dict:
-                    claims_format = jwt_conf_dict['claims_format']
-                    assert body['jwt']['claims_namespace'] == claims_namespace
-                    assert body['jwt']['claims_format'] == claims_format
-        else:
-            assert body['jwt'] == []
+    def test_rejects_an_invalid_role(self, hge_ctx):
+        headers = {
+            'x-hasura-role': 'user',
+        }
+        resp = hge_ctx.http.get(hge_ctx.hge_url + '/v1alpha1/config', headers=headers)
+        assert resp.status_code == 400, resp
 
-        # test if the request fails without auth headers if admin secret is set
-        if hge_key is not None:
-            resp = hge_ctx.http.get(hge_ctx.hge_url + '/v1alpha1/config')
-            body = resp.json()
-            assert ((resp.status_code == 401) or (resp.status_code == 400))
+
+@pytest.mark.admin_secret
+class TestConfigApiWithAdminSecret:
+    def test_responds_correctly(self, hge_ctx, hge_key):
+        headers = {
+            'x-hasura-role': 'admin',
+            'x-hasura-admin-secret': hge_key,
+        }
+        resp = hge_ctx.http.get(hge_ctx.hge_url + '/v1alpha1/config', headers=headers)
+        assert resp.status_code == 200, resp
+
+        body = resp.json()
+        print('Body:', body)
+        assert body['is_admin_secret_set'] == True
+        assert body['is_auth_hook_set'] == False
+        assert body['is_jwt_set'] == False
+        assert body['jwt'] == []
+
+    def test_rejects_an_invalid_role(self, hge_ctx, hge_key):
+        headers = {
+            'x-hasura-role': 'user',
+            'x-hasura-admin-secret': hge_key,
+        }
+        resp = hge_ctx.http.get(hge_ctx.hge_url + '/v1alpha1/config', headers=headers)
+        assert resp.status_code == 400, resp
+
+    def test_request_fails_without_auth_headers(self, hge_ctx):
+        resp = hge_ctx.http.get(hge_ctx.hge_url + '/v1alpha1/config')
+        assert (resp.status_code == 401) or (resp.status_code == 400)
+
+
+@pytest.mark.admin_secret
+@pytest.mark.usefixtures('auth_hook')
+class TestConfigApiWithAuthHook:
+    def test_responds_correctly(self, hge_ctx, hge_key):
+        headers = {
+            'x-hasura-role': 'admin',
+            'x-hasura-admin-secret': hge_key,
+        }
+        resp = hge_ctx.http.get(hge_ctx.hge_url + '/v1alpha1/config', headers=headers)
+        assert resp.status_code == 200, resp
+
+        body = resp.json()
+        print('Body:', body)
+        assert body['is_admin_secret_set'] == True
+        assert body['is_auth_hook_set'] == True
+        assert body['is_jwt_set'] == False
+        assert body['jwt'] == []
+
+    def test_rejects_an_invalid_role(self, hge_ctx, hge_key):
+        headers = {
+            'x-hasura-role': 'user',
+            'x-hasura-admin-secret': hge_key,
+        }
+        resp = hge_ctx.http.get(hge_ctx.hge_url + '/v1alpha1/config', headers=headers)
+        assert resp.status_code == 400, resp
+
+    def test_request_fails_without_auth_headers(self, hge_ctx):
+        resp = hge_ctx.http.get(hge_ctx.hge_url + '/v1alpha1/config')
+        assert (resp.status_code == 401) or (resp.status_code == 400)
+
+
+@pytest.mark.admin_secret
+@pytest.mark.usefixtures('jwt_configuration')
+@pytest.mark.jwt('ed25519')
+class TestConfigApiWithJwtAndNoClaims:
+    def test_rejects_an_invalid_role(self, hge_ctx, hge_key):
+        headers = {
+            'x-hasura-role': 'user',
+            'x-hasura-admin-secret': hge_key,
+        }
+        resp = hge_ctx.http.get(hge_ctx.hge_url + '/v1alpha1/config', headers=headers)
+        assert resp.status_code == 400, resp
+
+    def test_responds_correctly(self, hge_ctx, hge_key):
+        headers = {
+            'x-hasura-role': 'admin',
+            'x-hasura-admin-secret': hge_key,
+        }
+        resp = hge_ctx.http.get(hge_ctx.hge_url + '/v1alpha1/config', headers=headers)
+        assert resp.status_code == 200, resp
+
+        body = resp.json()
+        print('Body:', body)
+        assert body['is_admin_secret_set'] == True
+        assert body['is_auth_hook_set'] == False
+        assert body['is_jwt_set'] == True
+
+        assert body['jwt'] == [
+            {
+                'claims_format': 'json',
+                'claims_map': None,
+                'claims_namespace': 'https://hasura.io/jwt/claims',
+            }
+        ]
+
+    def test_request_fails_without_auth_headers(self, hge_ctx):
+        resp = hge_ctx.http.get(hge_ctx.hge_url + '/v1alpha1/config')
+        assert (resp.status_code == 401) or (resp.status_code == 400)
+
+
+@pytest.mark.admin_secret
+@pytest.mark.usefixtures('jwt_configuration')
+@pytest.mark.jwt('ed25519', {
+    'claims_format': 'stringified_json',
+})
+class TestConfigApiWithJwtAndClaimsFormat:
+    def test_responds_correctly(self, hge_ctx, hge_key):
+        headers = {
+            'x-hasura-role': 'admin',
+            'x-hasura-admin-secret': hge_key,
+        }
+        resp = hge_ctx.http.get(hge_ctx.hge_url + '/v1alpha1/config', headers=headers)
+        assert resp.status_code == 200, resp
+
+        body = resp.json()
+        print('Body:', body)
+        assert body['is_admin_secret_set'] == True
+        assert body['is_auth_hook_set'] == False
+        assert body['is_jwt_set'] == True
+
+        assert body['jwt'] == [
+            {
+                'claims_format': 'stringified_json',
+                'claims_map': None,
+                'claims_namespace': 'https://hasura.io/jwt/claims',
+            }
+        ]
+
+
+@pytest.mark.admin_secret
+@pytest.mark.usefixtures('jwt_configuration')
+@pytest.mark.jwt('ed25519', {
+    'claims_namespace': 'https://example.org/jwt/claims',
+    'claims_format': 'stringified_json',
+})
+class TestConfigApiWithJwtAndClaimsNamespace:
+    def test_responds_correctly(self, hge_ctx, hge_key):
+        headers = {
+            'x-hasura-role': 'admin',
+            'x-hasura-admin-secret': hge_key,
+        }
+        resp = hge_ctx.http.get(hge_ctx.hge_url + '/v1alpha1/config', headers=headers)
+        assert resp.status_code == 200, resp
+
+        body = resp.json()
+        print('Body:', body)
+        assert body['is_admin_secret_set'] == True
+        assert body['is_auth_hook_set'] == False
+        assert body['is_jwt_set'] == True
+
+        assert body['jwt'] == [
+            {
+                'claims_format': 'stringified_json',
+                'claims_map': None,
+                'claims_namespace': 'https://example.org/jwt/claims',
+            }
+        ]

server/tests-py/test_graphql_queries.py

@@ -927,7 +927,6 @@ class TestUnauthorizedRolePermission:
     def test_unauth_role(self, hge_ctx, transport):
         check_query_f(hge_ctx, self.dir() + '/unauthorized_role.yaml', transport, False)
 
-@pytest.mark.parametrize('transport', ['http'])
 @usefixtures('per_class_tests_db_state')
 @pytest.mark.admin_secret
 @pytest.mark.hge_env('HASURA_GRAPHQL_UNAUTHORIZED_ROLE', 'anonymous')
@@ -936,22 +935,31 @@ class TestFallbackUnauthorizedRoleCookie:
     def dir(cls):
         return 'queries/unauthorized_role'
 
-    def test_fallback_unauth_role_jwt_cookie_not_set(self, hge_ctx, transport):
-        check_query_f(hge_ctx, self.dir() + '/cookie_header_absent_unauth_role_set.yaml', transport, add_auth=False)
+    def test_fallback_unauth_role_jwt_cookie_not_set(self, hge_ctx):
+        check_query_f(hge_ctx, self.dir() + '/cookie_header_absent_unauth_role_set.yaml', add_auth=False)
 
-@pytest.mark.skipif(
-    not PytestConf.config.getoption("--test-no-cookie-and-unauth-role"),
-    reason="--test-no-cookie-and-unauth-role missing"
-)
-@pytest.mark.parametrize('transport', ['http'])
-@usefixtures('per_class_tests_db_state')
+@usefixtures('per_class_tests_db_state', 'jwt_configuration')
+@pytest.mark.admin_secret
+@pytest.mark.jwt('rsa')
+@pytest.mark.hge_env('HASURA_GRAPHQL_UNAUTHORIZED_ROLE', 'anonymous')
+class TestFallbackUnauthorizedRoleCookieWithJwt:
+    @classmethod
+    def dir(cls):
+        return 'queries/unauthorized_role'
+
+    def test_fallback_unauth_role_jwt_cookie_not_set(self, hge_ctx):
+        check_query_f(hge_ctx, self.dir() + '/cookie_header_absent_unauth_role_set.yaml', add_auth=False)
+
+@usefixtures('per_class_tests_db_state', 'jwt_configuration')
+@pytest.mark.admin_secret
+@pytest.mark.jwt('rsa')
 class TestMissingUnauthorizedRoleAndCookie:
     @classmethod
     def dir(cls):
         return 'queries/unauthorized_role'
 
-    def test_error_unauth_role_not_set_jwt_cookie_not_set(self, hge_ctx, transport):
-        check_query_f(hge_ctx, self.dir() + '/cookie_header_absent_unauth_role_not_set.yaml', transport, add_auth=False)
+    def test_error_unauth_role_not_set_jwt_cookie_not_set(self, hge_ctx):
+        check_query_f(hge_ctx, self.dir() + '/cookie_header_absent_unauth_role_not_set.yaml', add_auth=False)
 
 @usefixtures('per_class_tests_db_state')
 class TestGraphQLExplainPostgresMSSQL:

server/tests-py/test_jwt_claims_map.py

@@ -1,34 +1,49 @@
-import pytest
-import jwt
-import math
-import json
-
-from validate import check_query
 from datetime import datetime, timedelta
-from context import PytestConf
+import math
+
+import jwt
+import pytest
 from ruamel.yaml import YAML
 
-yaml=YAML(typ='safe', pure=True)
+from validate import check_query
 
-if not PytestConf.config.getoption('--hge-jwt-key-file'):
-    pytest.skip('--hge-jwt-key-file is missing, skipping JWT tests', allow_module_level=True)
+yaml = YAML(typ='safe', pure=True)
 
-hge_jwt_conf = PytestConf.config.getoption('--hge-jwt-conf')
+basic_claims_map = {
+    'x-hasura-user-id': {
+        'path': "$.['https://myapp.com/jwt/claims'].user.id"
+    },
+    'x-hasura-allowed-roles': {
+        'path': "$.['https://myapp.com/jwt/claims'].role.allowed"
+    },
+    'x-hasura-default-role': {
+        'path': "$.['https://myapp.com/jwt/claims'].role.default"
+    }
+}
 
-if not hge_jwt_conf:
-    pytest.skip('--hge-jwt-key-conf is missing, skipping JWT tests', allow_module_level=True)
+basic_claims_map_with_default_values = {
+    'x-hasura-user-id': {
+        'path': "$.['https://myapp.com/jwt/claims'].user.id",
+        'default': '1'
+    },
+    'x-hasura-allowed-roles': {
+        'path': "$.['https://myapp.com/jwt/claims'].role.allowed",
+        'default': ['user', 'editor']
+    },
+    'x-hasura-default-role': {
+        'path': "$.['https://myapp.com/jwt/claims'].role.default",
+        'default': 'user'
+    }
+}
 
-if 'claims_map' not in hge_jwt_conf:
-    pytest.skip('cliams_map missing in jwt config, skipping JWT Claims Map tests', allow_module_level=True)
-
-# The following claims_map is assumed to be set
-# {
-#     "claims_map": {
-#         "x-hasura-user-id": {"path":"$.["https://myapp.com/jwt/claims"].user.id"}
-#         "x-hasura-allowed-roles": {"$.["https://myapp.com/jwt/claims"].role.allowed","default":["user","editor"]}
-#         "x-hasura-default-role": {"$.["https://myapp.com/jwt/claims"].role.default","default":"user"}
-#     }
-# }
+static_claims_map = {
+    'x-hasura-user-id': {
+        'path': "$.['https://myapp.com/jwt/claims'].user.id"
+    },
+    'x-hasura-allowed-roles': ['user','editor'],
+    'x-hasura-default-role': 'user',
+    'x-hasura-custom-header': 'custom-value'
+}
 
 def clean_null_terms(d):
    clean = {}
@@ -46,8 +61,9 @@ def clean_null_terms(d):
 # default values here is referred to the default value that's being
 # used when a value is not found while looking up the JWT token using
 # the JSON Path provided
+@pytest.mark.admin_secret
 @pytest.mark.parametrize('endpoint', ['/v1/graphql', '/v1alpha1/graphql'])
-class TestJWTClaimsMapBasic():
+class AbstractTestJWTClaimsMapBasic:
     def mk_claims(self, user_id=None, allowed_roles=None, default_role=None):
         self.claims['https://myapp.com/jwt/claims'] = clean_null_terms({
             'user': {
@@ -59,17 +75,17 @@ class TestJWTClaimsMapBasic():
             }
         })
 
-    def test_jwt_claims_map_valid_claims_success(self, hge_ctx, endpoint):
+    def test_jwt_claims_map_valid_claims_success(self, hge_ctx, jwt_configuration, endpoint):
         self.mk_claims('1', ['user', 'editor'], 'user')
-        token = jwt.encode(self.claims, hge_ctx.hge_jwt_key, algorithm=hge_ctx.hge_jwt_algo)
+        token = jwt.encode(self.claims, jwt_configuration.private_key, algorithm=jwt_configuration.algorithm)
         self.conf['headers']['Authorization'] = 'Bearer ' + token
         self.conf['url'] = endpoint
         self.conf['status'] = 200
         check_query(hge_ctx, self.conf, add_auth=False)
 
-    def test_jwt_claims_map_invalid_role_in_request_header(self, hge_ctx, endpoint):
+    def test_jwt_claims_map_invalid_role_in_request_header(self, hge_ctx, jwt_configuration, endpoint):
         self.mk_claims('1', ['contractor', 'editor'], 'contractor')
-        token = jwt.encode(self.claims, hge_ctx.hge_jwt_key, algorithm=hge_ctx.hge_jwt_algo)
+        token = jwt.encode(self.claims, jwt_configuration.private_key, algorithm=jwt_configuration.algorithm)
         self.conf['headers']['Authorization'] = 'Bearer ' + token
         self.conf['response'] = {
             'errors': [{
@@ -87,10 +103,10 @@ class TestJWTClaimsMapBasic():
             self.conf['status'] = 400
         check_query(hge_ctx, self.conf, add_auth=False)
 
-    def test_jwt_claims_map_no_allowed_roles_in_claim(self, hge_ctx, endpoint):
+    def test_jwt_claims_map_no_allowed_roles_in_claim(self, hge_ctx, jwt_configuration, endpoint):
         self.mk_claims('1', None, 'user')
-        default_allowed_roles = hge_ctx.hge_jwt_conf_dict['claims_map']['x-hasura-allowed-roles'].get('default')
-        token = jwt.encode(self.claims, hge_ctx.hge_jwt_key, algorithm=hge_ctx.hge_jwt_algo)
+        default_allowed_roles = jwt_configuration.server_configuration['claims_map']['x-hasura-allowed-roles'].get('default')
+        token = jwt.encode(self.claims, jwt_configuration.private_key, algorithm=jwt_configuration.algorithm)
         self.conf['headers']['Authorization'] = 'Bearer ' + token
         if default_allowed_roles is None:
             self.conf['response'] = {
@@ -112,9 +128,9 @@ class TestJWTClaimsMapBasic():
         self.conf['url'] = endpoint
         check_query(hge_ctx, self.conf, add_auth=False)
 
-    def test_jwt_claims_map_invalid_allowed_roles_in_claim(self, hge_ctx, endpoint):
+    def test_jwt_claims_map_invalid_allowed_roles_in_claim(self, hge_ctx, jwt_configuration, endpoint):
         self.mk_claims('1', 'user', 'user')
-        token = jwt.encode(self.claims, hge_ctx.hge_jwt_key, algorithm=hge_ctx.hge_jwt_algo)
+        token = jwt.encode(self.claims, jwt_configuration.private_key, algorithm=jwt_configuration.algorithm)
         self.conf['headers']['Authorization'] = 'Bearer ' + token
         self.conf['response'] = {
             'errors': [{
@@ -132,13 +148,13 @@ class TestJWTClaimsMapBasic():
             self.conf['status'] = 400
         check_query(hge_ctx, self.conf, add_auth=False)
 
-    def test_jwt_claims_map_no_default_role(self, hge_ctx, endpoint):
+    def test_jwt_claims_map_no_default_role(self, hge_ctx, jwt_configuration, endpoint):
         # default_default_role is the default default role set in the JWT config
         # when the lookup with the JSONPath fails, this is the value that will
         # be used for the `x-hasura-default-role` claim
-        default_default_role = hge_ctx.hge_jwt_conf_dict['claims_map']['x-hasura-default-role'].get('default')
+        default_default_role = jwt_configuration.server_configuration['claims_map']['x-hasura-default-role'].get('default')
         self.mk_claims('1', ['user'])
-        token = jwt.encode(self.claims, hge_ctx.hge_jwt_key, algorithm=hge_ctx.hge_jwt_algo)
+        token = jwt.encode(self.claims, jwt_configuration.private_key, algorithm=jwt_configuration.algorithm)
         self.conf['headers']['Authorization'] = 'Bearer ' + token
         if default_default_role is None:
             self.conf['response'] = {
@@ -159,10 +175,10 @@ class TestJWTClaimsMapBasic():
         self.conf['url'] = endpoint
         check_query(hge_ctx, self.conf, add_auth=False)
 
-    def test_jwt_claims_map_claim_not_found(self, hge_ctx, endpoint):
-        default_user_id = hge_ctx.hge_jwt_conf_dict['claims_map']['x-hasura-user-id'].get('default')
+    def test_jwt_claims_map_claim_not_found(self, hge_ctx, jwt_configuration, endpoint):
+        default_user_id = jwt_configuration.server_configuration['claims_map']['x-hasura-user-id'].get('default')
         self.mk_claims(None, ['user', 'editor'], 'user')
-        token = jwt.encode(self.claims, hge_ctx.hge_jwt_key, algorithm=hge_ctx.hge_jwt_algo)
+        token = jwt.encode(self.claims, jwt_configuration.private_key, algorithm=jwt_configuration.algorithm)
         self.conf['headers']['Authorization'] = 'Bearer ' + token
         if default_user_id is None:
             self.conf['response'] = {
@@ -204,10 +220,32 @@ class TestJWTClaimsMapBasic():
         yield
         hge_ctx.v1q_f(self.dir + '/teardown.yaml')
 
+
+@pytest.mark.jwt('rsa', { 'claims_map': basic_claims_map })
+class TestJWTClaimsMapBasicWithRSA(AbstractTestJWTClaimsMapBasic):
+    pass
+
+
+@pytest.mark.jwt('ed25519', { 'claims_map': basic_claims_map })
+class TestJWTClaimsMapBasicWithEd25519(AbstractTestJWTClaimsMapBasic):
+    pass
+
+
+@pytest.mark.jwt('rsa', { 'claims_map': basic_claims_map_with_default_values })
+class TestJWTClaimsMapBasicWithRSAAndDefaultValues(AbstractTestJWTClaimsMapBasic):
+    pass
+
+
+@pytest.mark.jwt('ed25519', { 'claims_map': basic_claims_map_with_default_values })
+class TestJWTClaimsMapBasicWithEd25519AndDefaultValues(AbstractTestJWTClaimsMapBasic):
+    pass
+
+
 # The values of 'x-hasura-allowed-roles' and 'x-hasura-default-role' has
 # been set in the JWT config
+@pytest.mark.admin_secret
 @pytest.mark.parametrize('endpoint', ['/v1/graphql', '/v1alpha1/graphql'])
-class TestJWTClaimsMapWithStaticHasuraClaimsMapValues():
+class AbstractTestJWTClaimsMapWithStaticHasuraClaimsMapValues:
     def mk_claims(self, user_id=None):
         self.claims['https://myapp.com/jwt/claims'] = clean_null_terms({
             'user': {
@@ -215,18 +253,18 @@ class TestJWTClaimsMapWithStaticHasuraClaimsMapValues():
             }
         })
 
-    def test_jwt_claims_map_valid_claims_success(self, hge_ctx, endpoint):
+    def test_jwt_claims_map_valid_claims_success(self, hge_ctx, jwt_configuration, endpoint):
         self.mk_claims('1')
-        token = jwt.encode(self.claims, hge_ctx.hge_jwt_key, algorithm=hge_ctx.hge_jwt_algo)
+        token = jwt.encode(self.claims, jwt_configuration.private_key, algorithm=jwt_configuration.algorithm)
         self.conf['headers']['Authorization'] = 'Bearer ' + token
         self.conf['headers']['x-hasura-custom-header'] = 'custom-value'
         self.conf['url'] = endpoint
         self.conf['status'] = 200
         check_query(hge_ctx, self.conf, add_auth=False)
 
-    def test_jwt_claims_map_invalid_role_in_request_header(self, hge_ctx, endpoint):
+    def test_jwt_claims_map_invalid_role_in_request_header(self, hge_ctx, jwt_configuration, endpoint):
         self.mk_claims('1')
-        token = jwt.encode(self.claims, hge_ctx.hge_jwt_key, algorithm=hge_ctx.hge_jwt_algo)
+        token = jwt.encode(self.claims, jwt_configuration.private_key, algorithm=jwt_configuration.algorithm)
         self.conf['headers']['Authorization'] = 'Bearer ' + token
         self.conf['headers']['X-Hasura-Role'] = 'random_string'
         self.conf['response'] = {
@@ -245,9 +283,9 @@ class TestJWTClaimsMapWithStaticHasuraClaimsMapValues():
             self.conf['status'] = 400
         check_query(hge_ctx, self.conf, add_auth=False)
 
-    def test_jwt_claims_map_claim_not_found(self, hge_ctx, endpoint):
+    def test_jwt_claims_map_claim_not_found(self, hge_ctx, jwt_configuration, endpoint):
         self.mk_claims(None)
-        token = jwt.encode(self.claims, hge_ctx.hge_jwt_key, algorithm=hge_ctx.hge_jwt_algo)
+        token = jwt.encode(self.claims, jwt_configuration.private_key, algorithm=jwt_configuration.algorithm)
         self.conf['headers']['Authorization'] = 'Bearer ' + token
         self.conf['response'] = {
             'errors': [{
@@ -285,3 +323,13 @@ class TestJWTClaimsMapWithStaticHasuraClaimsMapValues():
         hge_ctx.v1q_f(self.dir + '/setup.yaml')
         yield
         hge_ctx.v1q_f(self.dir + '/teardown.yaml')
+
+
+@pytest.mark.jwt('rsa', { 'claims_map': static_claims_map })
+class TestJWTClaimsMapWithStaticHasuraClaimsMapValuesWithRSA(AbstractTestJWTClaimsMapWithStaticHasuraClaimsMapValues):
+    pass
+
+
+@pytest.mark.jwt('ed25519', { 'claims_map': static_claims_map })
+class TestJWTClaimsMapWithStaticHasuraClaimsMapValuesWithEd25519(AbstractTestJWTClaimsMapWithStaticHasuraClaimsMapValues):
+    pass

server/tests-py/validate.py

@@ -5,7 +5,6 @@ import copy
 import graphql
 import json
 import jsondiff
-import jwt
 import os
 import pytest
 import queue
@@ -220,26 +219,9 @@ def test_forbidden_webhook(hge_ctx, conf):
         'request id': resp_hdrs.get('x-request-id')
     })
 
-def mk_claims_with_namespace_path(claims,hasura_claims,namespace_path):
-    if namespace_path is None:
-        claims['https://hasura.io/jwt/claims'] = hasura_claims
-    elif namespace_path == "$":
-        claims.update(hasura_claims)
-    elif namespace_path == "$.hasura_claims":
-        claims['hasura_claims'] = hasura_claims
-    elif namespace_path == "$.hasura['claims%']":
-        claims['hasura'] = {}
-        claims['hasura']['claims%'] = hasura_claims
-    else:
-        raise Exception(
-                '''claims_namespace_path should not be anything
-                other than $.hasura_claims, $.hasura['claims%'] or $ for testing. The
-                value of claims_namespace_path was {}'''.format(namespace_path))
-    return claims
-
 # Returns the response received and a bool indicating whether the test passed
 # or not (this will always be True unless we are `--accepting`)
-def check_query(hge_ctx: HGECtx, conf, transport='http', add_auth=True, claims_namespace_path=None, gqlws=False):
+def check_query(hge_ctx: HGECtx, conf, transport='http', add_auth=True, gqlws=False):
     headers = {}
     if 'headers' in conf:
         # Convert header values to strings, as the YAML parser might give us an internal class.
@@ -253,21 +235,6 @@ def check_query(hge_ctx: HGECtx, conf, transport='http', add_auth=True, claims_n
         headers['X-Hasura-Role'] = 'admin'
 
     if add_auth:
-        # Use the hasura role specified in the test case, and create a JWT token
-        if hge_ctx.hge_jwt_key is not None and len(headers) > 0 and 'X-Hasura-Role' in headers:
-            hClaims = dict()
-            hClaims['X-Hasura-Allowed-Roles'] = [headers['X-Hasura-Role']]
-            hClaims['X-Hasura-Default-Role'] = headers['X-Hasura-Role']
-            for key in headers:
-                if key != 'X-Hasura-Role':
-                    hClaims[key] = headers[key]
-            claim = {
-                "sub": "foo",
-                "name": "bar",
-            }
-            claim = mk_claims_with_namespace_path(claim, hClaims, claims_namespace_path)
-            headers['Authorization'] = 'Bearer ' + jwt.encode(claim, hge_ctx.hge_jwt_key, algorithm=hge_ctx.hge_jwt_algo)
-
         # Use the hasura role specified in the test case, and create an authorization token which will be verified by webhook
         if hge_ctx.webhook and len(headers) > 0:
             if hge_ctx.webhook.tls_trust != TLSTrust.INSECURE:
@@ -275,14 +242,14 @@ def check_query(hge_ctx: HGECtx, conf, transport='http', add_auth=True, claims_n
                 test_forbidden_webhook(hge_ctx, conf)
             headers = authorize_for_webhook(headers)
 
-        # The case as admin with admin-secret and jwt/webhook
-        elif (hge_ctx.webhook or hge_ctx.hge_jwt_key is not None) \
+        # The case as admin with admin-secret and webhook
+        elif hge_ctx.webhook \
              and hge_ctx.hge_key is not None \
              and len(headers) == 0:
             headers['X-Hasura-Admin-Secret'] = hge_ctx.hge_key
 
         # The case as admin with only admin-secret
-        elif hge_ctx.hge_key is not None and not hge_ctx.webhook and hge_ctx.hge_jwt_key is None:
+        elif hge_ctx.hge_key is not None and not hge_ctx.webhook:
             # Test whether it is forbidden when incorrect/no admin_secret is specified
             test_forbidden_when_admin_secret_reqd(hge_ctx, conf)
             headers['X-Hasura-Admin-Secret'] = hge_ctx.hge_key
@@ -570,7 +537,7 @@ def check_query_f(hge_ctx, f, transport='http', add_auth=True, gqlws = False):
               if PytestConf.config.getoption("--port-to-haskell"):
                   add_spec(f + " [" + str(ix) + "]", sconf)
               else:
-                  actual_resp, matched = check_query(hge_ctx, sconf, transport, add_auth, None, gqlws)
+                  actual_resp, matched = check_query(hge_ctx, sconf, transport, add_auth, gqlws)
                   if PytestConf.config.getoption("--accept") and not matched:
                       conf[ix]['response'] = actual_resp
                       should_write_back = True
@@ -580,7 +547,7 @@ def check_query_f(hge_ctx, f, transport='http', add_auth=True, gqlws = False):
             else:
                 if conf['status'] != 200:
                     hge_ctx.may_skip_test_teardown = True
-                actual_resp, matched = check_query(hge_ctx, conf, transport, add_auth, None, gqlws)
+                actual_resp, matched = check_query(hge_ctx, conf, transport, add_auth, gqlws)
                 # If using `--accept` write the file back out with the new expected
                 # response set to the actual response we got:
                 if PytestConf.config.getoption("--accept") and not matched: