WIP hack integration tests for auth emulator

Author: muru

.github/workflows/ci.yml

@@ -30,6 +30,8 @@ jobs:
       run: |
         npm install -g firebase-tools
         firebase emulators:exec --only database --project fake-project-id 'pytest integration/test_db.py'
+        echo mock-api-key > apikey.txt
+        firebase emulators:exec --only auth --project mock-project-id 'pytest integration/test_auth.py --cert tests/data/service_account.json --apikey apikey.txt'
 
   lint:
     runs-on: ubuntu-latest

firebase_admin/_auth_client.py

@@ -14,7 +14,6 @@
 
 """Firebase auth client sub module."""
 
-import os
 import time
 
 import firebase_admin
@@ -27,9 +26,6 @@
 from firebase_admin import _user_mgt
 from firebase_admin import _utils
 
-_EMULATOR_HOST_ENV_VAR = 'FIREBASE_AUTH_EMULATOR_HOST'
-_DEFAULT_AUTH_URL = 'https://identitytoolkit.googleapis.com'
-
 class Client:
     """Firebase Authentication client scoped to a specific tenant."""
 
@@ -44,19 +40,17 @@ def __init__(self, app, tenant_id=None):
         version_header = 'Python/Admin/{0}'.format(firebase_admin.__version__)
         # Non-default endpoint URLs for emulator support are set in this dict later.
         endpoint_urls = {}
+        self.emulated = False
 
         # If an emulator is present, check that the given value matches the expected format and set
         # endpoint URLs to use the emulator. Additionally, use a fake credential.
-        emulator_host = os.environ.get(_EMULATOR_HOST_ENV_VAR)
+        emulator_host = _auth_utils.get_emulator_host()
         if emulator_host:
-            if '//' in emulator_host:
-                raise ValueError(
-                    'Invalid {0}: "{1}". It must follow format "host:port".'.format(
-                        _EMULATOR_HOST_ENV_VAR, emulator_host))
             base_url = 'http://{0}/identitytoolkit.googleapis.com'.format(emulator_host)
             endpoint_urls['v1'] = base_url + '/v1'
             endpoint_urls['v2beta1'] = base_url + '/v2beta1'
             credential = _utils.EmulatorAdminCredentials()
+            self.emulated = True
         else:
             # Use credentials if provided
             credential = app.credential.get_credential()
@@ -132,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 check_revoked:
+        if not self.emulated and check_revoked:
             self._check_jwt_revoked(verified_claims, _token_gen.RevokedIdTokenError, 'ID token')
         return verified_claims
 

firebase_admin/_auth_utils.py

@@ -15,13 +15,14 @@
 """Firebase auth utils."""
 
 import json
+import os
 import re
 from urllib import parse
 
 from firebase_admin import exceptions
 from firebase_admin import _utils
 
-
+EMULATOR_HOST_ENV_VAR = 'FIREBASE_AUTH_EMULATOR_HOST'
 MAX_CLAIMS_PAYLOAD_SIZE = 1000
 RESERVED_CLAIMS = set([
     'acr', 'amr', 'at_hash', 'aud', 'auth_time', 'azp', 'cnf', 'c_hash', 'exp', 'iat',
@@ -66,6 +67,19 @@ def __iter__(self):
         return self
 
 
+def get_emulator_host():
+    emulator_host = os.getenv(EMULATOR_HOST_ENV_VAR)
+    if emulator_host and '//' in emulator_host:
+        raise ValueError(
+            'Invalid {0}: "{1}". It must follow format "host:port".'.format(
+                EMULATOR_HOST_ENV_VAR, emulator_host))
+    return emulator_host
+
+
+def is_emulated():
+    return get_emulator_host() not in [None, '']
+
+
 def validate_uid(uid, required=False):
     if uid is None and not required:
         return None

firebase_admin/_token_gen.py

@@ -31,6 +31,7 @@
 from firebase_admin import _auth_utils
 
 
+#_auth_utils.is_emulated() = _auth_utils.get_emulator_host() != ''
 # ID token constants
 ID_TOKEN_ISSUER_PREFIX = 'https://securetoken.google.com/'
 ID_TOKEN_CERT_URI = ('https://www.googleapis.com/robot/v1/metadata/x509/'
@@ -54,6 +55,16 @@
                         'service-accounts/default/email')
 
 
+class _EmulatedSigner(google.auth.crypt.Signer):
+    key_id = None
+
+    def __init__(self):
+        pass
+
+    def sign(self, message):
+        return b''
+
+
 class _SigningProvider:
     """Stores a reference to a google.auth.crypto.Signer."""
 
@@ -78,6 +89,10 @@ def from_iam(cls, request, google_cred, service_account):
         signer = iam.Signer(request, google_cred, service_account)
         return _SigningProvider(signer, service_account)
 
+    @classmethod
+    def for_emulator(cls):
+        return _SigningProvider(_EmulatedSigner(), '[email protected]')
+
 
 class TokenGenerator:
     """Generates custom tokens and session cookies."""
@@ -94,6 +109,8 @@ def __init__(self, app, http_client, url_override=None):
 
     def _init_signing_provider(self):
         """Initializes a signing provider by following the go/firebase-admin-sign protocol."""
+        if _auth_utils.is_emulated():
+            return _SigningProvider.for_emulator()
         # If the SDK was initialized with a service account, use it to sign bytes.
         google_cred = self.app.credential.get_credential()
         if isinstance(google_cred, google.oauth2.service_account.Credentials):
@@ -291,15 +308,15 @@ def verify(self, token, request):
             error_message = (
                 '{0} expects {1}, but was given a custom '
                 'token.'.format(self.operation, self.articled_short_name))
-        elif not header.get('kid'):
+        elif not _auth_utils.is_emulated() and not header.get('kid'):
             if header.get('alg') == 'HS256' and payload.get(
                     'v') == 0 and 'uid' in payload.get('d', {}):
                 error_message = (
                     '{0} expects {1}, but was given a legacy custom '
                     'token.'.format(self.operation, self.articled_short_name))
             else:
                 error_message = 'Firebase {0} has no "kid" claim.'.format(self.short_name)
-        elif header.get('alg') != 'RS256':
+        elif not _auth_utils.is_emulated() and header.get('alg') != 'RS256':
             error_message = (
                 'Firebase {0} has incorrect algorithm. Expected "RS256" but got '
                 '"{1}". {2}'.format(self.short_name, header.get('alg'), verify_id_token_msg))
@@ -329,6 +346,10 @@ def verify(self, token, request):
         if error_message:
             raise self._invalid_token_error(error_message)
 
+        if _auth_utils.is_emulated():
+            claims = jwt.decode(token, verify=False)
+            claims['uid'] = claims['sub']
+            return claims
         try:
             verified_claims = google.oauth2.id_token.verify_token(
                 token,
@@ -342,7 +363,7 @@ def verify(self, token, request):
         except ValueError as error:
             if 'Token expired' in str(error):
                 raise self._expired_token_error(str(error), cause=error)
-            raise self._invalid_token_error(str(error), cause=error)
+            raise self._invalid_token_error(str(error) + "FOO", cause=error)
 
     def _decode_unverified(self, token):
         try:

integration/test_auth.py

@@ -15,6 +15,7 @@
 """Integration tests for firebase_admin.auth module."""
 import base64
 import datetime
+import os
 import random
 import string
 import time
@@ -32,11 +33,18 @@
 from firebase_admin import credentials
 
 
-_verify_token_url = 'https://www.googleapis.com/identitytoolkit/v3/relyingparty/verifyCustomToken'
-_verify_password_url = 'https://www.googleapis.com/identitytoolkit/v3/relyingparty/verifyPassword'
-_password_reset_url = 'https://www.googleapis.com/identitytoolkit/v3/relyingparty/resetPassword'
-_verify_email_url = 'https://www.googleapis.com/identitytoolkit/v3/relyingparty/setAccountInfo'
-_email_sign_in_url = 'https://www.googleapis.com/identitytoolkit/v3/relyingparty/emailLinkSignin'
+_EMULATOR_HOST_ENV_VAR = 'FIREBASE_AUTH_EMULATOR_HOST'
+URL_PREFIX = 'https://www.googleapis.com/identitytoolkit'
+
+emulator_host = os.getenv(_EMULATOR_HOST_ENV_VAR)
+if emulator_host:
+    URL_PREFIX = 'http://{0}/www.googleapis.com/identitytoolkit'.format(emulator_host)
+
+_verify_token_url = '{0}/v3/relyingparty/verifyCustomToken'.format(URL_PREFIX)
+_verify_password_url = '{0}/v3/relyingparty/verifyPassword'.format(URL_PREFIX)
+_password_reset_url = '{0}/v3/relyingparty/resetPassword'.format(URL_PREFIX)
+_verify_email_url = '{0}/v3/relyingparty/setAccountInfo'.format(URL_PREFIX)
+_email_sign_in_url = '{0}/v3/relyingparty/emailLinkSignin'.format(URL_PREFIX)
 
 ACTION_LINK_CONTINUE_URL = 'http://localhost?a=1&b=5#f=1'
 
@@ -560,6 +568,9 @@ def test_verify_id_token_revoked(new_user, api_key):
     # verify_id_token succeeded because it didn't check revoked.
     assert claims['iat'] * 1000 < user.tokens_valid_after_timestamp
 
+    if emulator_host:
+        pytest.skip("Not supported with auth emulator")
+
     with pytest.raises(auth.RevokedIdTokenError) as excinfo:
         claims = auth.verify_id_token(id_token, check_revoked=True)
     assert str(excinfo.value) == 'The Firebase ID token has been revoked.'

tests/test_token_gen.py

@@ -75,6 +75,8 @@ def _merge_jwt_claims(defaults, overrides):
     return defaults
 
 def verify_custom_token(custom_token, expected_claims, tenant_id=None):
+    if os.getenv(EMULATOR_HOST_ENV_VAR):
+        pytest.skip("Not supported with auth emulator")
     assert isinstance(custom_token, bytes)
     token = google.oauth2.id_token.verify_token(
         custom_token,
@@ -230,10 +232,14 @@ 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 os.getenv(EMULATOR_HOST_ENV_VAR):
+            pytest.skip("Not supported with auth emulator")
         with pytest.raises(ValueError):
             auth.create_custom_token(MOCK_UID, app=user_mgt_app)
 
     def test_sign_with_iam(self):
+        if os.getenv(EMULATOR_HOST_ENV_VAR):
+            pytest.skip("Not supported with auth emulator")
         options = {'serviceAccountId': 'test-service-account', 'projectId': 'mock-project-id'}
         app = firebase_admin.initialize_app(
             testutils.MockCredential(), name='iam-signer-app', options=options)
@@ -248,6 +254,8 @@ def test_sign_with_iam(self):
             firebase_admin.delete_app(app)
 
     def test_sign_with_iam_error(self):
+        if os.getenv(EMULATOR_HOST_ENV_VAR):
+            pytest.skip("Not supported with auth emulator")
         options = {'serviceAccountId': 'test-service-account', 'projectId': 'mock-project-id'}
         app = firebase_admin.initialize_app(
             testutils.MockCredential(), name='iam-signer-app', options=options)
@@ -264,6 +272,8 @@ def test_sign_with_iam_error(self):
             firebase_admin.delete_app(app)
 
     def test_sign_with_discovered_service_account(self):
+        if os.getenv(EMULATOR_HOST_ENV_VAR):
+            pytest.skip("Not supported with auth emulator")
         request = testutils.MockRequest(200, 'discovered-service-account')
         options = {'projectId': 'mock-project-id'}
         app = firebase_admin.initialize_app(testutils.MockCredential(), name='iam-signer-app',
@@ -287,6 +297,8 @@ def test_sign_with_discovered_service_account(self):
             firebase_admin.delete_app(app)
 
     def test_sign_with_discovery_failure(self):
+        if os.getenv(EMULATOR_HOST_ENV_VAR):
+            pytest.skip("Not supported with auth emulator")
         request = testutils.MockFailedRequest(Exception('test error'))
         options = {'projectId': 'mock-project-id'}
         app = firebase_admin.initialize_app(testutils.MockCredential(), name='iam-signer-app',
@@ -431,6 +443,8 @@ def test_valid_token_check_revoked(self, user_mgt_app, id_token):
 
     @pytest.mark.parametrize('id_token', valid_tokens.values(), ids=list(valid_tokens))
     def test_revoked_token_check_revoked(self, user_mgt_app, revoked_tokens, id_token):
+        if os.getenv(EMULATOR_HOST_ENV_VAR):
+            pytest.skip("Not supported with auth emulator")
         _overwrite_cert_request(user_mgt_app, MOCK_REQUEST)
         _instrument_user_manager(user_mgt_app, 200, revoked_tokens)
         with pytest.raises(auth.RevokedIdTokenError) as excinfo:
@@ -460,13 +474,18 @@ def test_invalid_arg(self, user_mgt_app, id_token):
 
     @pytest.mark.parametrize('id_token', invalid_tokens.values(), ids=list(invalid_tokens))
     def test_invalid_token(self, user_mgt_app, id_token):
+        if os.getenv(EMULATOR_HOST_ENV_VAR):
+            pytest.skip("Not supported with auth emulator")
         _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 os.getenv(EMULATOR_HOST_ENV_VAR):
+            pytest.skip("Not supported with auth emulator")
+        _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:
@@ -505,6 +524,8 @@ def test_custom_token(self, auth_app):
         assert str(excinfo.value) == message
 
     def test_certificate_request_failure(self, user_mgt_app):
+        if os.getenv(EMULATOR_HOST_ENV_VAR):
+            pytest.skip("Not supported with auth emulator")
         _overwrite_cert_request(user_mgt_app, testutils.MockRequest(404, 'not found'))
         with pytest.raises(auth.CertificateFetchError) as excinfo:
             auth.verify_id_token(TEST_ID_TOKEN, app=user_mgt_app)
@@ -580,13 +601,17 @@ def test_invalid_args(self, user_mgt_app, cookie):
 
     @pytest.mark.parametrize('cookie', invalid_cookies.values(), ids=list(invalid_cookies))
     def test_invalid_cookie(self, user_mgt_app, cookie):
+        if os.getenv(EMULATOR_HOST_ENV_VAR):
+            pytest.skip("Not supported with auth emulator")
         _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 os.getenv(EMULATOR_HOST_ENV_VAR):
+            pytest.skip("Not supported with auth emulator")
         _overwrite_cert_request(user_mgt_app, MOCK_REQUEST)
         cookie = self.invalid_cookies['ExpiredCookie']
         with pytest.raises(auth.ExpiredSessionCookieError) as excinfo:
@@ -620,6 +645,8 @@ def test_custom_token(self, auth_app):
             auth.verify_session_cookie(custom_token, app=auth_app)
 
     def test_certificate_request_failure(self, user_mgt_app):
+        if os.getenv(EMULATOR_HOST_ENV_VAR):
+            pytest.skip("Not supported with auth emulator")
         _overwrite_cert_request(user_mgt_app, testutils.MockRequest(404, 'not found'))
         with pytest.raises(auth.CertificateFetchError) as excinfo:
             auth.verify_session_cookie(TEST_SESSION_COOKIE, app=user_mgt_app)
@@ -632,6 +659,8 @@ def test_certificate_request_failure(self, user_mgt_app):
 class TestCertificateCaching:
 
     def test_certificate_caching(self, user_mgt_app, httpserver):
+        if os.getenv(EMULATOR_HOST_ENV_VAR):
+            pytest.skip("Not supported with auth emulator")
         httpserver.serve_content(MOCK_PUBLIC_CERTS, 200, headers={'Cache-Control': 'max-age=3600'})
         verifier = _token_gen.TokenVerifier(user_mgt_app)
         verifier.cookie_verifier.cert_url = httpserver.url