CVE-2019-1020015 (Other, published 2019-07-29T13:15Z)
Description: value: graphql-engine (aka Hasura GraphQL Engine) before 1.0.0-beta.3 mishandles the audience check while verifying JWT.
Severity: MEDIUM (CVSS v3 base score: 7.5)
Weakness: NVD-CWE-noinfo \u2014 Insufficient Information
There is insufficient information about the issue to classify it; details are unkown or unspecified.
Repository: https://github.com/hasura/graphql-engine
Files changed:
.circleci/test-server.sh
docs/graphql/manual/auth/authentication/jwt.rst
server/src-lib/Hasura/Server/Auth.hs
server/src-lib/Hasura/Server/Auth/JWT.hs
server/tests-py/test_jwt.py
Commit message: Merge pull request from GHSA-2j98-fw5g-j43v

* fix bug in audience check while verifying JWT

  - previously the check was converting the audience type into a string
  and then comparing with the conf value. all audience types (as it is a
  string or URI) will convert to plain strings
  - use the Audience type from the jose library for comparing

* add docs for audience

* add issuer check as well

* docs minor syntax fix

* skip audience check if not given in conf

* minor docs update

* qualify import jose library


Vulnerable code:

     ...
     ...
     ...
  return $ JWTCtx jwkRef (jcClaimNs conf) (jcAudience conf) claimsFmt
  , JWKSet (..)
import           Crypto.JWT
  , jcKeyOrUrl     :: !(Either JWK URI)
  , jcAudience     :: !(Maybe T.Text)
  -- , jcIssuer   :: !(Maybe T.Text)
  { jcxKey          :: !(IORef JWKSet)
  , jcxAudience     :: !(Maybe T.Text)
  show (JWTCtx _ nsM audM cf) =
    show ["<IORef JWKSet>", show nsM, show audM, show cf]
  -> IORef JWKSet
  -> IORef JWKSet
      expTimeM = fmap (\(NumericDate t) -> t) $ claims ^. claimExp
  let mHasuraClaims = Map.lookup claimsNs $ claims ^. unregisteredClaims
  :: ( MonadError JWTError m
  -> m ClaimsSet
  jwt <- decodeCompact rawJWT
  verifyClaimsAt config key t jwt
    audCheck aud = maybe True (== (T.pack . show) aud) $ jcxAudience ctx
    config = defaultJWTValidationSettings audCheck
  toJSON (JWTConfig ty keyOrUrl claimNs aud claimsFmt) =
        return $ JWTConfig keyType (Left key) claimNs aud isStrngfd
        return $ JWTConfig keyType (Right url) claimNs aud isStrngfd
    pytest.skip('--hge-jwt-key-file is missing, skipping JWT basic tests', allow_module_level=True)
    pytest.skip('--hge-jwt-key-conf is missing, skipping JWT basic tests', allow_module_level=True)

Fixed code:
echo -e "\n$(time_elapsed): <########## TEST GRAPHQL-ENGINE WITH ADMIN SECRET AND JWT (with audience check - string) #####################################>\n"
TEST_TYPE="jwt-audience-check-single-string"


export HASURA_GRAPHQL_JWT_SECRET="$(jq -n --arg key "$(cat $OUTPUT_FOLDER/ssl/jwt_public.key)" '{ type: "RS512", key: $key , audience: "myapp-1234"}')"

run_hge_with_args serve
wait_for_port 8080

pytest -n 1 -vv --hge-urls "$HGE_URL" --pg-urls "$HASURA_GRAPHQL_DATABASE_URL" --hge-key="$HASURA_GRAPHQL_ADMIN_SECRET" --hge-jwt-key-file="$OUTPUT_FOLDER/ssl/jwt_private.key" --hge-jwt-conf="$HASURA_GRAPHQL_JWT_SECRET" test_jwt.py

kill_hge_servers

unset HASURA_GRAPHQL_JWT_SECRET

echo -e "\n$(time_elapsed): <########## TEST GRAPHQL-ENGINE WITH ADMIN SECRET AND JWT (with audience check - list of strings) #################################>\n"
TEST_TYPE="jwt-audience-check-list-string"

export HASURA_GRAPHQL_JWT_SECRET="$(jq -n --arg key "$(cat $OUTPUT_FOLDER/ssl/jwt_public.key)" '{ type: "RS512", key: $key , audience: ["myapp-1234", "myapp-9876"]}')"

run_hge_with_args serve
wait_for_port 8080

pytest -n 1 -vv --hge-urls "$HGE_URL" --pg-urls "$HASURA_GRAPHQL_DATABASE_URL" --hge-key="$HASURA_GRAPHQL_ADMIN_SECRET" --hge-jwt-key-file="$OUTPUT_FOLDER/ssl/jwt_private.key" --hge-jwt-conf="$HASURA_GRAPHQL_JWT_SECRET" test_jwt.py

kill_hge_servers

unset HASURA_GRAPHQL_JWT_SECRET

echo -e "\n$(time_elapsed): <########## TEST GRAPHQL-ENGINE WITH ADMIN SECRET AND JWT (with issuer check) #####################################>\n"
TEST_TYPE="jwt-issuer-check"

export HASURA_GRAPHQL_JWT_SECRET="$(jq -n --arg key "$(cat $OUTPUT_FOLDER/ssl/jwt_public.key)" '{ type: "RS512", key: $key , issuer: "https://hasura.com"}')"

run_hge_with_args serve
wait_for_port 8080

pytest -n 1 -vv --hge-urls "$HGE_URL" --pg-urls "$HASURA_GRAPHQL_DATABASE_URL" --hge-key="$HASURA_GRAPHQL_ADMIN_SECRET" --hge-jwt-key-file="$OUTPUT_FOLDER/ssl/jwt_private.key" --hge-jwt-conf="$HASURA_GRAPHQL_JWT_SECRET" test_jwt.py

kill_hge_servers

unset HASURA_GRAPHQL_JWT_SECRET



  return $ JWTCtx jwkRef (jcClaimNs conf) (jcAudience conf) claimsFmt (jcIssuer conf)
  , Jose.JWKSet (..)
import qualified Crypto.JWT                      as Jose
  , jcKeyOrUrl     :: !(Either Jose.JWK URI)
  , jcAudience     :: !(Maybe Jose.Audience)
  , jcIssuer       :: !(Maybe Jose.StringOrURI)
  { jcxKey          :: !(IORef Jose.JWKSet)
  , jcxAudience     :: !(Maybe Jose.Audience)
  , jcxIssuer       :: !(Maybe Jose.StringOrURI)
  show (JWTCtx _ nsM audM cf iss) =
    show ["<IORef JWKSet>", show nsM, show audM, show cf, show iss]
  -> IORef Jose.JWKSet
  -> IORef Jose.JWKSet
      expTimeM = fmap (\(Jose.NumericDate t) -> t) $ claims ^. Jose.claimExp
  let mHasuraClaims = Map.lookup claimsNs $ claims ^. Jose.unregisteredClaims
  :: ( MonadError Jose.JWTError m
  -> m Jose.ClaimsSet
  jwt <- Jose.decodeCompact rawJWT
  Jose.verifyClaimsAt config key t jwt
    config = case jcxIssuer ctx of
      Nothing  -> Jose.defaultJWTValidationSettings audCheck
      Just iss -> Jose.defaultJWTValidationSettings audCheck
                  & set Jose.issuerPredicate (== iss)
    audCheck audience =
      -- dont perform the check if there are no audiences in the conf
      case jcxAudience ctx of
        Nothing                        -> True
        Just (Jose.Audience audiences) -> audience `elem` audiences
  toJSON (JWTConfig ty keyOrUrl claimNs aud claimsFmt iss) =
                            , "issuer" A..= iss
    iss     <- o A..:? "issuer"
        return $ JWTConfig keyType (Left key) claimNs aud isStrngfd iss
        return $ JWTConfig keyType (Right url) claimNs aud isStrngfd iss
    pytest.skip('--hge-jwt-key-file is missing, skipping JWT tests', allow_module_level=True)
    pytest.skip('--hge-jwt-key-conf is missing, skipping JWT tests', allow_module_level=True)
    def test_jwt_no_audience_in_conf(self, hge_ctx, endpoint):
        self.claims['https://hasura.io/jwt/claims'] = mk_claims(hge_ctx.hge_jwt_conf, {
            'x-hasura-user-id': '1',
            'x-hasura-default-role': 'user',
            'x-hasura-allowed-roles': ['user'],
        })

        self.claims['aud'] = 'hasura-test-suite'
        token = jwt.encode(self.claims, hge_ctx.hge_jwt_key, algorithm='RS512').decode('utf-8')
        self.conf['headers']['Authorization'] = 'Bearer ' + token
        self.conf['url'] = endpoint
        check_query(hge_ctx, self.conf, add_auth=False)

    def test_jwt_no_issuer_in_conf(self, hge_ctx, endpoint):
        self.claims['https://hasura.io/jwt/claims'] = mk_claims(hge_ctx.hge_jwt_conf, {
            'x-hasura-user-id': '1',
            'x-hasura-default-role': 'user',
            'x-hasura-allowed-roles': ['user'],
        })

        self.claims['iss'] = 'rubbish-issuer'
        token = jwt.encode(self.claims, hge_ctx.hge_jwt_key, algorithm='RS512').decode('utf-8')
        self.conf['headers']['Authorization'] = 'Bearer ' + token
        self.conf['url'] = endpoint
        check_query(hge_ctx, self.conf, add_auth=False)



@pytest.mark.parametrize('endpoint', ['/v1/graphql', '/v1alpha1/graphql'])
class TestJwtAudienceCheck():
    def test_jwt_valid_audience(self, hge_ctx, endpoint):
        jwt_conf = json.loads(hge_ctx.hge_jwt_conf)
        if 'audience' not in jwt_conf:
            pytest.skip('audience not present in conf, skipping testing audience')

        audience = jwt_conf['audience']
        audience = audience if isinstance(audience, str) else audience[0]
        self.claims['https://hasura.io/jwt/claims'] = mk_claims(hge_ctx.hge_jwt_conf, {
            'x-hasura-user-id': '1',
            'x-hasura-default-role': 'user',
            'x-hasura-allowed-roles': ['user'],
        })
        self.claims['aud'] = audience

        token = jwt.encode(self.claims, hge_ctx.hge_jwt_key, algorithm='RS512').decode('utf-8')
        self.conf['headers']['Authorization'] = 'Bearer ' + token
        check_query(hge_ctx, self.conf, add_auth=False)

    def test_jwt_invalid_audience(self, hge_ctx, endpoint):
        jwt_conf = json.loads(hge_ctx.hge_jwt_conf)
        if 'audience' not in jwt_conf:
            pytest.skip('audience not present in conf, skipping testing audience')

        self.claims['https://hasura.io/jwt/claims'] = mk_claims(hge_ctx.hge_jwt_conf, {
            'x-hasura-user-id': '1',
            'x-hasura-default-role': 'user',
            'x-hasura-allowed-roles': ['user'],
        })
        self.claims['aud'] = 'rubbish_audience'

        token = jwt.encode(self.claims, hge_ctx.hge_jwt_key, algorithm='RS512').decode('utf-8')
        self.conf['headers']['Authorization'] = 'Bearer ' + token
        self.conf['response'] = {
            'errors': [{
                'extensions': {
                    'code': 'invalid-jwt',
                    'path': '$'
                },
                'message': 'Could not verify JWT: JWTNotInAudience'
            }]
        }
        self.conf['url'] = endpoint
        if endpoint == '/v1/graphql':
            self.conf['status'] = 200
        if endpoint == '/v1alpha1/graphql':
            self.conf['status'] = 400
        check_query(hge_ctx, self.conf, add_auth=False)

    @pytest.fixture(autouse=True)
    def transact(self, setup):
        self.dir = 'queries/graphql_query/permissions'
        with open(self.dir + '/user_select_query_unpublished_articles.yaml') as c:
            self.conf = yaml.safe_load(c)
        curr_time = datetime.now()
        exp_time = curr_time + timedelta(hours=1)
        self.claims = {
            'sub': '1234567890',
            'name': 'John Doe',
            'iat': math.floor(curr_time.timestamp()),
            'exp': math.floor(exp_time.timestamp())
        }

    @pytest.fixture(scope='class')
    def setup(self, request, hge_ctx):
        self.dir = 'queries/graphql_query/permissions'
        st_code, resp = hge_ctx.v1q_f(self.dir + '/setup.yaml')
        assert st_code == 200, resp
        yield
        st_code, resp = hge_ctx.v1q_f(self.dir + '/teardown.yaml')
        assert st_code == 200, resp

@pytest.mark.parametrize('endpoint', ['/v1/graphql', '/v1alpha1/graphql'])
class TestJwtIssuerCheck():
    def test_jwt_valid_issuer(self, hge_ctx, endpoint):
        jwt_conf = json.loads(hge_ctx.hge_jwt_conf)
        if 'issuer' not in jwt_conf:
            pytest.skip('issuer not present in conf, skipping testing issuer')

        issuer = jwt_conf['issuer']
        self.claims['https://hasura.io/jwt/claims'] = mk_claims(hge_ctx.hge_jwt_conf, {
            'x-hasura-user-id': '1',
            'x-hasura-default-role': 'user',
            'x-hasura-allowed-roles': ['user'],
        })
        self.claims['iss'] = issuer

        token = jwt.encode(self.claims, hge_ctx.hge_jwt_key, algorithm='RS512').decode('utf-8')
        self.conf['headers']['Authorization'] = 'Bearer ' + token
        check_query(hge_ctx, self.conf, add_auth=False)

    def test_jwt_invalid_issuer(self, hge_ctx, endpoint):
        jwt_conf = json.loads(hge_ctx.hge_jwt_conf)
        if 'issuer' not in jwt_conf:
            pytest.skip('issuer not present in conf, skipping testing issuer')

        self.claims['https://hasura.io/jwt/claims'] = mk_claims(hge_ctx.hge_jwt_conf, {
            'x-hasura-user-id': '1',
            'x-hasura-default-role': 'user',
            'x-hasura-allowed-roles': ['user'],
        })
        self.claims['iss'] = 'rubbish_issuer'

        token = jwt.encode(self.claims, hge_ctx.hge_jwt_key, algorithm='RS512').decode('utf-8')
        self.conf['headers']['Authorization'] = 'Bearer ' + token
        self.conf['response'] = {
            'errors': [{
                'extensions': {
                    'code': 'invalid-jwt',
                    'path': '$'
                },
                'message': 'Could not verify JWT: JWTNotInIssuer'
            }]
        }
        self.conf['url'] = endpoint
        if endpoint == '/v1/graphql':
            self.conf['status'] = 200
        if endpoint == '/v1alpha1/graphql':
            self.conf['status'] = 400
        check_query(hge_ctx, self.conf, add_auth=False)

    @pytest.fixture(autouse=True)
    def transact(self, setup):
        self.dir = 'queries/graphql_query/permissions'
        with open(self.dir + '/user_select_query_unpublished_articles.yaml') as c:
            self.conf = yaml.safe_load(c)
        curr_time = datetime.now()
        exp_time = curr_time + timedelta(hours=1)
        self.claims = {
            'sub': '1234567890',
            'name': 'John Doe',
            'iat': math.floor(curr_time.timestamp()),
            'exp': math.floor(exp_time.timestamp())
        }

    @pytest.fixture(scope='class')
    def setup(self, request, hge_ctx):
        self.dir = 'queries/graphql_query/permissions'
        st_code, resp = hge_ctx.v1q_f(self.dir + '/setup.yaml')
        assert st_code == 200, resp
        yield
        st_code, resp = hge_ctx.v1q_f(self.dir + '/teardown.yaml')
        assert st_code == 200, resp