Token verification for the auth emulator

Author: muru

firebase_admin/_token_gen.py

@@ -53,6 +53,19 @@
 METADATA_SERVICE_URL = ('http://metadata.google.internal/computeMetadata/v1/instance/'
                         'service-accounts/default/email')
 
+# Emulator fake account
+AUTH_EMULATOR_EMAIL = '[email protected]'
+
+
+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 +91,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(), AUTH_EMULATOR_EMAIL)
+
 
 class TokenGenerator:
     """Generates custom tokens and session cookies."""
@@ -94,6 +111,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):
@@ -286,20 +305,22 @@ def verify(self, token, request):
         verify_id_token_msg = (
             'See {0} for details on how to retrieve {1}.'.format(self.url, self.short_name))
 
+        emulated = _auth_utils.is_emulated()
+
         error_message = None
         if audience == FIREBASE_AUDIENCE:
             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 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 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))
@@ -330,11 +351,14 @@ def verify(self, token, request):
             raise self._invalid_token_error(error_message)
 
         try:
-            verified_claims = google.oauth2.id_token.verify_token(
-                token,
-                request=request,
-                audience=self.project_id,
-                certs_url=self.cert_url)
+            if emulated:
+                verified_claims = payload
+            else:
+                verified_claims = google.oauth2.id_token.verify_token(
+                    token,
+                    request=request,
+                    audience=self.project_id,
+                    certs_url=self.cert_url)
             verified_claims['uid'] = verified_claims['sub']
             return verified_claims
         except google.auth.exceptions.TransportError as error: