Accomodate auth emulator behaviour in tests.

Where possible, tests are modified to account for the current behaviour
in emulator mode (e.g., invalid or expired tokens or cookies still
work).

In one case, signer verification didn't account for padding being
stripped in JWT tokens, and was fixed to do so.

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,15 @@ 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
+
+
+# These fixtures are set to the function scope as the emulator environment variable bleeds over when
+# in module scope.
+@pytest.fixture(scope='function', params=[{'emulated': False}, {'emulated': True}])
 def auth_app(request):
     """Returns an App initialized with a mock service account credential.
 
@@ -157,7 +171,7 @@ def auth_app(request):
     firebase_admin.delete_app(app)
     monkeypatch.undo()
 
-@pytest.fixture(scope='module', params=[{'emulated': False}, {'emulated': True}])
+@pytest.fixture(scope='function', params=[{'emulated': False}, {'emulated': True}])
 def user_mgt_app(request):
     monkeypatch = testutils.new_monkeypatch()
     if request.param['emulated']:
@@ -230,20 +244,30 @@ 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():
+            # Should work fine with the emulator, so do a condensed version of
+            # test_sign_with_iam below.
+            custom_token = auth.create_custom_token(MOCK_UID, app=user_mgt_app).decode()
+            self._verify_signer(custom_token, _token_gen.AUTH_EMULATOR_EMAIL)
+            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,6 +278,12 @@ 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))
+            if _is_emulated():
+                # Should work fine with the emulator, so do a condensed version of
+                # test_sign_with_iam above.
+                custom_token = auth.create_custom_token(MOCK_UID, app=app).decode()
+                self._verify_signer(custom_token, _token_gen.AUTH_EMULATOR_EMAIL)
+                return
             with pytest.raises(auth.TokenSignError) as excinfo:
                 auth.create_custom_token(MOCK_UID, app=app)
             error = excinfo.value
@@ -264,7 +294,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 +310,16 @@ 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('.')
+                # No requests will be made with the emulator
+                assert len(request.log) == 0
+            else:
+                assert custom_token.endswith('.' + signature.rstrip('='))
+                assert len(request.log) == 2
+                assert request.log[0][1]['headers'] == {'Metadata-Flavor': 'Google'}
+            self._verify_signer(custom_token, signer)
         finally:
             firebase_admin.delete_app(app)
 
@@ -293,6 +330,12 @@ def test_sign_with_discovery_failure(self):
                                             options=options)
         try:
             _overwrite_iam_request(app, request)
+            if _is_emulated():
+                # Should work fine with the emulator, so do a condensed version of
+                # test_sign_with_iam above.
+                custom_token = auth.create_custom_token(MOCK_UID, app=app).decode()
+                self._verify_signer(custom_token, _token_gen.AUTH_EMULATOR_EMAIL)
+                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 +347,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 +457,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,17 +515,31 @@ 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, check them like valid tokens.
+            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):
+        _overwrite_cert_request(user_mgt_app, MOCK_REQUEST)
         _overwrite_cert_request(user_mgt_app, MOCK_REQUEST)
         id_token = self.invalid_tokens['ExpiredToken']
+        if _is_emulated():
+            # This token won't be invalid when using the emulator, check it like a valid token.
+            claims = auth.verify_id_token(id_token, app=user_mgt_app)
+            assert claims['admin'] is True
+            assert claims['uid'] == claims['sub']
+            return
         with pytest.raises(auth.ExpiredIdTokenError) as excinfo:
             auth.verify_id_token(id_token, app=user_mgt_app)
         assert isinstance(excinfo.value, auth.InvalidIdTokenError)
@@ -506,6 +577,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 +615,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,8 +659,15 @@ 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, check them like valid cookies.
+            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)
@@ -589,6 +677,12 @@ def test_invalid_cookie(self, user_mgt_app, cookie):
     def test_expired_cookie(self, user_mgt_app):
         _overwrite_cert_request(user_mgt_app, MOCK_REQUEST)
         cookie = self.invalid_cookies['ExpiredCookie']
+        if _is_emulated():
+            # This cookie won't be invalid when using the emulator, check it like a valid cookie.
+            claims = auth.verify_session_cookie(cookie, app=user_mgt_app)
+            assert claims['admin'] is True
+            assert claims['uid'] == claims['sub']
+            return
         with pytest.raises(auth.ExpiredSessionCookieError) as excinfo:
             auth.verify_session_cookie(cookie, app=user_mgt_app)
         assert isinstance(excinfo.value, auth.InvalidSessionCookieError)
@@ -621,6 +715,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 +735,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