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,7 +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:
@@ -44,19 +42,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 +128,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,15 @@ def __iter__(self):
         return self
 
 
+def get_emulator_host():
+    emulator_host = os.getenv(EMULATOR_HOST_ENV_VAR)
+    if '//' 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 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
 
 
+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 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 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 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 IS_EMULATED:
+            unverified_claims = jwt.decode(token, verify=False)
+            unverified_claims['uid'] = unverified_claims['user_id']
+            return unverified_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

@@ -431,6 +431,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: