Token verification for the auth emulator

Author: muru

firebase_admin/._auth_utils.py.swp

@@ -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_client.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/_auth_utils.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: