Fix or skip tests that fail with auth emulator

Author: muru

firebase_admin/_auth_client.py

@@ -126,7 +126,7 @@ def verify_id_token(self, id_token, check_revoked=False):
                 raise _auth_utils.TenantIdMismatchError(
                     'Invalid tenant ID: {0}'.format(token_tenant_id))
 
-        if not self.emulated and check_revoked:
+        if check_revoked:
             self._check_jwt_revoked(verified_claims, _token_gen.RevokedIdTokenError, 'ID token')
         return verified_claims
 

tests/test_token_gen.py

@@ -76,13 +76,19 @@ def _merge_jwt_claims(defaults, overrides):
 
 def verify_custom_token(custom_token, expected_claims, tenant_id=None):
     assert isinstance(custom_token, bytes)
-    token = google.oauth2.id_token.verify_token(
-        custom_token,
-        testutils.MockRequest(200, MOCK_PUBLIC_CERTS),
-        _token_gen.FIREBASE_AUDIENCE)
+    if _is_emulated():
+        token = jwt.decode(custom_token, verify=False)
+    else:
+        token = google.oauth2.id_token.verify_token(
+            custom_token,
+            testutils.MockRequest(200, MOCK_PUBLIC_CERTS),
+            _token_gen.FIREBASE_AUDIENCE)
     assert token['uid'] == MOCK_UID
-    assert token['iss'] == MOCK_SERVICE_ACCOUNT_EMAIL
-    assert token['sub'] == MOCK_SERVICE_ACCOUNT_EMAIL
+    expected_email = MOCK_SERVICE_ACCOUNT_EMAIL
+    if _is_emulated():
+        expected_email = _token_gen.AUTH_EMULATOR_EMAIL
+    assert token['iss'] == expected_email
+    assert token['sub'] == expected_email
     if tenant_id is None:
         assert 'tenant_id' not in token
     else:
@@ -141,7 +147,13 @@ def _overwrite_iam_request(app, request):
     client = auth._get_client(app)
     client._token_generator.request = request
 
-@pytest.fixture(scope='module', params=[{'emulated': False}, {'emulated': True}])
+
+def _is_emulated():
+    emulator_host = os.getenv(EMULATOR_HOST_ENV_VAR, '')
+    return emulator_host and '//' not in emulator_host
+
+
+@pytest.fixture(scope='function', params=[{'emulated': False}, {'emulated': True}])
 def auth_app(request):
     """Returns an App initialized with a mock service account credential.
 
@@ -217,6 +229,12 @@ class TestCreateCustomToken:
         'MultipleReservedClaims': (MOCK_UID, {'sub':'1234', 'aud':'foo'}, ValueError),
     }
 
+    def _check_emulated_token(self, uid, app):
+        # Should work fine with the emulator, so do a condensed version of
+        # test_sign_with_iam above.
+        custom_token = auth.create_custom_token(uid, app=app).decode()
+        self._verify_signer(custom_token, _token_gen.AUTH_EMULATOR_EMAIL)
+
     @pytest.mark.parametrize('values', valid_args.values(), ids=list(valid_args))
     def test_valid_params(self, auth_app, values):
         user, claims = values
@@ -230,20 +248,27 @@ def test_invalid_params(self, auth_app, values):
             auth.create_custom_token(user, claims, app=auth_app)
 
     def test_noncert_credential(self, user_mgt_app):
+        if _is_emulated():
+            self._check_emulated_token(MOCK_UID, user_mgt_app)
+            return
         with pytest.raises(ValueError):
             auth.create_custom_token(MOCK_UID, app=user_mgt_app)
 
     def test_sign_with_iam(self):
-        options = {'serviceAccountId': 'test-service-account', 'projectId': 'mock-project-id'}
+        signer = _token_gen.AUTH_EMULATOR_EMAIL if _is_emulated() else 'test-service-account'
+        options = {'serviceAccountId': signer, 'projectId': 'mock-project-id'}
         app = firebase_admin.initialize_app(
             testutils.MockCredential(), name='iam-signer-app', options=options)
         try:
             signature = base64.b64encode(b'test').decode()
             iam_resp = '{{"signature": "{0}"}}'.format(signature)
             _overwrite_iam_request(app, testutils.MockRequest(200, iam_resp))
             custom_token = auth.create_custom_token(MOCK_UID, app=app).decode()
-            assert custom_token.endswith('.' + signature.rstrip('='))
-            self._verify_signer(custom_token, 'test-service-account')
+            if _is_emulated():
+                assert custom_token.endswith('.')
+            else:
+                assert custom_token.endswith('.' + signature.rstrip('='))
+            self._verify_signer(custom_token, signer)
         finally:
             firebase_admin.delete_app(app)
 
@@ -254,7 +279,10 @@ def test_sign_with_iam_error(self):
         try:
             iam_resp = '{"error": {"code": 403, "message": "test error"}}'
             _overwrite_iam_request(app, testutils.MockRequest(403, iam_resp))
-            with pytest.raises(auth.TokenSignError) as excinfo:
+            if _is_emulated():
+                self._check_emulated_token(MOCK_UID, app)
+                return
+            with pytest.raises((ValueError, auth.TokenSignError)) as excinfo:
                 auth.create_custom_token(MOCK_UID, app=app)
             error = excinfo.value
             assert error.code == exceptions.UNKNOWN
@@ -264,7 +292,8 @@ def test_sign_with_iam_error(self):
             firebase_admin.delete_app(app)
 
     def test_sign_with_discovered_service_account(self):
-        request = testutils.MockRequest(200, 'discovered-service-account')
+        signer = _token_gen.AUTH_EMULATOR_EMAIL if _is_emulated() else 'discovered-service-account'
+        request = testutils.MockRequest(200, signer)
         options = {'projectId': 'mock-project-id'}
         app = firebase_admin.initialize_app(testutils.MockCredential(), name='iam-signer-app',
                                             options=options)
@@ -279,10 +308,17 @@ def test_sign_with_discovered_service_account(self):
             request.response = testutils.MockResponse(
                 200, '{{"signature": "{0}"}}'.format(signature))
             custom_token = auth.create_custom_token(MOCK_UID, app=app).decode()
-            assert custom_token.endswith('.' + signature.rstrip('='))
-            self._verify_signer(custom_token, 'discovered-service-account')
-            assert len(request.log) == 2
-            assert request.log[0][1]['headers'] == {'Metadata-Flavor': 'Google'}
+            if _is_emulated():
+                # No signature from the emulator
+                assert custom_token.endswith('.')
+            else:
+                assert custom_token.endswith('.' + signature.rstrip('='))
+            if _is_emulated():
+                # No requests will be made with the emulator
+                assert len(request.log) == 0
+            else:
+                assert len(request.log) == 2
+                assert request.log[0][1]['headers'] == {'Metadata-Flavor': 'Google'}
         finally:
             firebase_admin.delete_app(app)
 
@@ -293,6 +329,9 @@ def test_sign_with_discovery_failure(self):
                                             options=options)
         try:
             _overwrite_iam_request(app, request)
+            if _is_emulated():
+                self._check_emulated_token(MOCK_UID, app)
+                return
             with pytest.raises(ValueError) as excinfo:
                 auth.create_custom_token(MOCK_UID, app=app)
             assert str(excinfo.value).startswith('Failed to determine service account: test error')
@@ -304,6 +343,14 @@ def test_sign_with_discovery_failure(self):
     def _verify_signer(self, token, signer):
         segments = token.split('.')
         assert len(segments) == 3
+        # See https://github.com/googleapis/google-auth-library-python/pull/324
+        # and also RFC 7515 which is the basis of that PR:
+        # JWT segments don't have padding and base64 decoding might fail.
+        #
+        # Workaround from https://stackoverflow.com/a/9807138/2072269
+        missing_padding = len(segments[1]) % 4
+        if missing_padding:
+            segments[1] += '=' * (4 - missing_padding)
         body = json.loads(base64.b64decode(segments[1]).decode())
         assert body['iss'] == signer
         assert body['sub'] == signer
@@ -406,6 +453,12 @@ class TestVerifyIdToken:
         'BadFormatToken': 'foobar'
     }
 
+    tokens_not_invalid_in_emulator = [
+        'WrongKid',
+        'FutureToken',
+        'ExpiredToken'
+    ]
+
     @pytest.mark.parametrize('id_token', valid_tokens.values(), ids=list(valid_tokens))
     def test_valid_token(self, user_mgt_app, id_token):
         _overwrite_cert_request(user_mgt_app, MOCK_REQUEST)
@@ -458,15 +511,25 @@ def test_invalid_arg(self, user_mgt_app, id_token):
             auth.verify_id_token(id_token, app=user_mgt_app)
         assert 'Illegal ID token provided' in str(excinfo.value)
 
-    @pytest.mark.parametrize('id_token', invalid_tokens.values(), ids=list(invalid_tokens))
-    def test_invalid_token(self, user_mgt_app, id_token):
+    @pytest.mark.parametrize('id_token_key', list(invalid_tokens))
+    def test_invalid_token(self, user_mgt_app, id_token_key):
+        id_token = self.invalid_tokens[id_token_key]
+        if _is_emulated() and id_token_key in self.tokens_not_invalid_in_emulator:
+            # These tokens won't be invalid when using the emulator
+            claims = auth.verify_id_token(id_token, app=user_mgt_app)
+            assert claims['admin'] is True
+            assert claims['uid'] == claims['sub']
+            return
         _overwrite_cert_request(user_mgt_app, MOCK_REQUEST)
         with pytest.raises(auth.InvalidIdTokenError) as excinfo:
             auth.verify_id_token(id_token, app=user_mgt_app)
         assert isinstance(excinfo.value, exceptions.InvalidArgumentError)
         assert excinfo.value.http_response is None
 
     def test_expired_token(self, user_mgt_app):
+        if _is_emulated():
+            pytest.skip('Not supported in emulator mode: Token expiration')
+        _overwrite_cert_request(user_mgt_app, MOCK_REQUEST)
         _overwrite_cert_request(user_mgt_app, MOCK_REQUEST)
         id_token = self.invalid_tokens['ExpiredToken']
         with pytest.raises(auth.ExpiredIdTokenError) as excinfo:
@@ -506,6 +569,10 @@ def test_custom_token(self, auth_app):
 
     def test_certificate_request_failure(self, user_mgt_app):
         _overwrite_cert_request(user_mgt_app, testutils.MockRequest(404, 'not found'))
+        if _is_emulated():
+            # Shouldn't fetch certificates in emulator mode.
+            auth.verify_id_token(TEST_ID_TOKEN, app=user_mgt_app)
+            return
         with pytest.raises(auth.CertificateFetchError) as excinfo:
             auth.verify_id_token(TEST_ID_TOKEN, app=user_mgt_app)
         assert 'Could not fetch certificates' in str(excinfo.value)
@@ -540,6 +607,12 @@ class TestVerifySessionCookie:
         'IDToken': TEST_ID_TOKEN,
     }
 
+    cookies_not_invalid_in_emulator = [
+        'WrongKid',
+        'FutureCookie',
+        'ExpiredCookie'
+    ]
+
     @pytest.mark.parametrize('cookie', valid_cookies.values(), ids=list(valid_cookies))
     def test_valid_cookie(self, user_mgt_app, cookie):
         _overwrite_cert_request(user_mgt_app, MOCK_REQUEST)
@@ -578,15 +651,24 @@ def test_invalid_args(self, user_mgt_app, cookie):
             auth.verify_session_cookie(cookie, app=user_mgt_app)
         assert 'Illegal session cookie provided' in str(excinfo.value)
 
-    @pytest.mark.parametrize('cookie', invalid_cookies.values(), ids=list(invalid_cookies))
-    def test_invalid_cookie(self, user_mgt_app, cookie):
+    @pytest.mark.parametrize('cookie_key', list(invalid_cookies))
+    def test_invalid_cookie(self, user_mgt_app, cookie_key):
+        cookie = self.invalid_cookies[cookie_key]
+        if _is_emulated() and cookie_key in self.cookies_not_invalid_in_emulator:
+            # These cookies won't be invalid when using the emulator
+            claims = auth.verify_session_cookie(cookie, app=user_mgt_app)
+            assert claims['admin'] is True
+            assert claims['uid'] == claims['sub']
+            return
         _overwrite_cert_request(user_mgt_app, MOCK_REQUEST)
         with pytest.raises(auth.InvalidSessionCookieError) as excinfo:
             auth.verify_session_cookie(cookie, app=user_mgt_app)
         assert isinstance(excinfo.value, exceptions.InvalidArgumentError)
         assert excinfo.value.http_response is None
 
     def test_expired_cookie(self, user_mgt_app):
+        if _is_emulated():
+            pytest.skip('Not supported in emulator mode: Cookie expiration')
         _overwrite_cert_request(user_mgt_app, MOCK_REQUEST)
         cookie = self.invalid_cookies['ExpiredCookie']
         with pytest.raises(auth.ExpiredSessionCookieError) as excinfo:
@@ -621,6 +703,10 @@ def test_custom_token(self, auth_app):
 
     def test_certificate_request_failure(self, user_mgt_app):
         _overwrite_cert_request(user_mgt_app, testutils.MockRequest(404, 'not found'))
+        if _is_emulated():
+            # Shouldn't fetch certificates in emulator mode.
+            auth.verify_session_cookie(TEST_SESSION_COOKIE, app=user_mgt_app)
+            return
         with pytest.raises(auth.CertificateFetchError) as excinfo:
             auth.verify_session_cookie(TEST_SESSION_COOKIE, app=user_mgt_app)
         assert 'Could not fetch certificates' in str(excinfo.value)
@@ -637,9 +723,11 @@ def test_certificate_caching(self, user_mgt_app, httpserver):
         verifier.cookie_verifier.cert_url = httpserver.url
         verifier.id_token_verifier.cert_url = httpserver.url
         verifier.verify_session_cookie(TEST_SESSION_COOKIE)
-        assert len(httpserver.requests) == 1
+        # No requests should be made in emulated mode
+        request_count = 0 if _is_emulated() else 1
+        assert len(httpserver.requests) == request_count
         # Subsequent requests should not fetch certs from the server
         verifier.verify_session_cookie(TEST_SESSION_COOKIE)
-        assert len(httpserver.requests) == 1
+        assert len(httpserver.requests) == request_count
         verifier.verify_id_token(TEST_ID_TOKEN)
-        assert len(httpserver.requests) == 1
+        assert len(httpserver.requests) == request_count