Migrated token verification APIs to new exception types

firebase_admin/_auth_utils.py

@@ -216,7 +216,7 @@ class InvalidIdTokenError(exceptions.InvalidArgumentError):
 
     default_message = 'The provided ID token is invalid'
 
-    def __init__(self, message, cause, http_response=None):
+    def __init__(self, message, cause=None, http_response=None):
         exceptions.InvalidArgumentError.__init__(self, message, cause, http_response)
 
 

firebase_admin/_token_gen.py

@@ -217,12 +217,18 @@ def __init__(self, app):
             project_id=app.project_id, short_name='ID token',
             operation='verify_id_token()',
             doc_url='https://firebase.google.com/docs/auth/admin/verify-id-tokens',
-            cert_url=ID_TOKEN_CERT_URI, issuer=ID_TOKEN_ISSUER_PREFIX)
+            cert_url=ID_TOKEN_CERT_URI,
+            issuer=ID_TOKEN_ISSUER_PREFIX,
+            invalid_token_error=_auth_utils.InvalidIdTokenError,
+            expired_token_error=ExpiredIdTokenError)
         self.cookie_verifier = _JWTVerifier(
             project_id=app.project_id, short_name='session cookie',
             operation='verify_session_cookie()',
             doc_url='https://firebase.google.com/docs/auth/admin/verify-id-tokens',
-            cert_url=COOKIE_CERT_URI, issuer=COOKIE_ISSUER_PREFIX)
+            cert_url=COOKIE_CERT_URI,
+            issuer=COOKIE_ISSUER_PREFIX,
+            invalid_token_error=InvalidSessionCookieError,
+            expired_token_error=ExpiredSessionCookieError)
 
     def verify_id_token(self, id_token):
         return self.id_token_verifier.verify(id_token, self.request)
@@ -245,6 +251,8 @@ def __init__(self, **kwargs):
             self.articled_short_name = 'an {0}'.format(self.short_name)
         else:
             self.articled_short_name = 'a {0}'.format(self.short_name)
+        self._invalid_token_error = kwargs.pop('invalid_token_error')
+        self._expired_token_error = kwargs.pop('expired_token_error')
 
     def verify(self, token, request):
         """Verifies the signature and data for the provided JWT."""
@@ -261,8 +269,7 @@ def verify(self, token, request):
                 'or set your Firebase project ID as an app option. Alternatively set the '
                 'GOOGLE_CLOUD_PROJECT environment variable.'.format(self.operation))
 
-        header = jwt.decode_header(token)
-        payload = jwt.decode(token, verify=False)
+        header, payload = self._decode_unverified(token)
         issuer = payload.get('iss')
         audience = payload.get('aud')
         subject = payload.get('sub')
@@ -275,12 +282,12 @@ def verify(self, token, request):
             'See {0} for details on how to retrieve {1}.'.format(self.url, self.short_name))
 
         error_message = None
-        if not header.get('kid'):
-            if audience == FIREBASE_AUDIENCE:
-                error_message = (
-                    '{0} expects {1}, but was given a custom '
-                    'token.'.format(self.operation, self.articled_short_name))
-            elif header.get('alg') == 'HS256' and payload.get(
+        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'):
+            if header.get('alg') == 'HS256' and payload.get(
                     'v') is 0 and 'uid' in payload.get('d', {}):
                 error_message = (
                     '{0} expects {1}, but was given a legacy custom '
@@ -315,19 +322,76 @@ def verify(self, token, request):
                 '{1}'.format(self.short_name, verify_id_token_msg))
 
         if error_message:
-            raise ValueError(error_message)
+            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)
+            verified_claims['uid'] = verified_claims['sub']
+            return verified_claims
+        except google.auth.exceptions.TransportError as error:
+            raise CertificateFetchError(str(error), cause=error)
+        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)
 
-        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
+    def _decode_unverified(self, token):
+        try:
+            header = jwt.decode_header(token)
+            payload = jwt.decode(token, verify=False)
+            return header, payload
+        except ValueError as error:
+            raise self._invalid_token_error(str(error), cause=error)
 
 
 class TokenSignError(exceptions.UnknownError):
     """Unexpected error while signing a Firebase custom token."""
 
     def __init__(self, message, cause):
         exceptions.UnknownError.__init__(self, message, cause)
+
+
+class CertificateFetchError(exceptions.UnknownError):
+    """Failed to fetch some public key certificates required to verify a token."""
+
+    def __init__(self, message, cause):
+        exceptions.UnknownError.__init__(self, message, cause)
+
+
+class ExpiredIdTokenError(_auth_utils.InvalidIdTokenError):
+    """The provided ID token is expired."""
+
+    def __init__(self, message, cause):
+        _auth_utils.InvalidIdTokenError.__init__(self, message, cause)
+
+
+class RevokedIdTokenError(_auth_utils.InvalidIdTokenError):
+    """The provided ID token has been revoked."""
+
+    def __init__(self, message):
+        _auth_utils.InvalidIdTokenError.__init__(self, message)
+
+
+class InvalidSessionCookieError(exceptions.InvalidArgumentError):
+    """The provided string is not a valid Firebase session cookie."""
+
+    def __init__(self, message, cause=None):
+        exceptions.InvalidArgumentError.__init__(self, message, cause)
+
+
+class ExpiredSessionCookieError(InvalidSessionCookieError):
+    """The provided session cookie is expired."""
+
+    def __init__(self, message, cause):
+        InvalidSessionCookieError.__init__(self, message, cause)
+
+
+class RevokedSessionCookieError(InvalidSessionCookieError):
+    """The provided session cookie has been revoked."""
+
+    def __init__(self, message):
+        InvalidSessionCookieError.__init__(self, message)

firebase_admin/auth.py

@@ -31,20 +31,23 @@
 
 
 _AUTH_ATTRIBUTE = '_auth'
-_ID_TOKEN_REVOKED = 'ID_TOKEN_REVOKED'
-_SESSION_COOKIE_REVOKED = 'SESSION_COOKIE_REVOKED'
 
 
 __all__ = [
     'ActionCodeSettings',
-    'AuthError',
+    'CertificateFetchError',
     'DELETE_ATTRIBUTE',
     'ErrorInfo',
+    'ExpiredIdTokenError',
+    'ExpiredSessionCookieError',
     'ExportedUserRecord',
     'ImportUserRecord',
     'InvalidDynamicLinkDomainError',
     'InvalidIdTokenError',
+    'InvalidSessionCookieError',
     'ListUsersPage',
+    'RevokedIdTokenError',
+    'RevokedSessionCookieError',
     'TokenSignError',
     'UidAlreadyExistsError',
     'UnexpectedResponseError',
@@ -76,17 +79,23 @@
 ]
 
 ActionCodeSettings = _user_mgt.ActionCodeSettings
+CertificateFetchError = _token_gen.CertificateFetchError
 DELETE_ATTRIBUTE = _user_mgt.DELETE_ATTRIBUTE
 ErrorInfo = _user_import.ErrorInfo
+ExpiredIdTokenError = _token_gen.ExpiredIdTokenError
+ExpiredSessionCookieError = _token_gen.ExpiredSessionCookieError
 ExportedUserRecord = _user_mgt.ExportedUserRecord
-ListUsersPage = _user_mgt.ListUsersPage
-UserImportHash = _user_import.UserImportHash
 ImportUserRecord = _user_import.ImportUserRecord
 InvalidDynamicLinkDomainError = _auth_utils.InvalidDynamicLinkDomainError
 InvalidIdTokenError = _auth_utils.InvalidIdTokenError
+InvalidSessionCookieError = _token_gen.InvalidSessionCookieError
+ListUsersPage = _user_mgt.ListUsersPage
+RevokedIdTokenError = _token_gen.RevokedIdTokenError
+RevokedSessionCookieError = _token_gen.RevokedSessionCookieError
 TokenSignError = _token_gen.TokenSignError
 UidAlreadyExistsError = _auth_utils.UidAlreadyExistsError
 UnexpectedResponseError = _auth_utils.UnexpectedResponseError
+UserImportHash = _user_import.UserImportHash
 UserImportResult = _user_import.UserImportResult
 UserInfo = _user_mgt.UserInfo
 UserMetadata = _user_mgt.UserMetadata
@@ -149,9 +158,12 @@ def verify_id_token(id_token, app=None, check_revoked=False):
         dict: A dictionary of key-value pairs parsed from the decoded JWT.
 
     Raises:
-        ValueError: If the JWT was found to be invalid, or if the App's project ID cannot
-            be determined.
-        AuthError: If ``check_revoked`` is requested and the token was revoked.
+        ValueError: If ``id_token`` is a not a string or is empty.
+        InvalidIdTokenError: If ``id_token`` is not a valid Firebase ID token.
+        ExpiredIdTokenError: If the specified ID token has expired.
+        RevokedIdTokenError: If ``check_revoked`` is ``True`` and the ID token has been revoked.
+        CertificateFetchError: If an error occurs while fetching the public key certificates
+            required to verify the ID token.
     """
     if not isinstance(check_revoked, bool):
         # guard against accidental wrong assignment.
@@ -160,7 +172,7 @@ def verify_id_token(id_token, app=None, check_revoked=False):
     token_verifier = _get_auth_service(app).token_verifier
     verified_claims = token_verifier.verify_id_token(id_token)
     if check_revoked:
-        _check_jwt_revoked(verified_claims, _ID_TOKEN_REVOKED, 'ID token', app)
+        _check_jwt_revoked(verified_claims, RevokedIdTokenError, 'ID token', app)
     return verified_claims
 
 
@@ -201,14 +213,17 @@ def verify_session_cookie(session_cookie, check_revoked=False, app=None):
         dict: A dictionary of key-value pairs parsed from the decoded JWT.
 
     Raises:
-        ValueError: If the cookie was found to be invalid, or if the App's project ID cannot
-            be determined.
-        AuthError: If ``check_revoked`` is requested and the cookie was revoked.
+        ValueError: If ``session_cookie`` is a not a string or is empty.
+        InvalidSessionCookieError: If ``session_cookie`` is not a valid Firebase session cookie.
+        ExpiredSessionCookieError: If the specified session cookie has expired.
+        RevokedSessionCookieError: If ``check_revoked`` is ``True`` and the cookie has been revoked.
+        CertificateFetchError: If an error occurs while fetching the public key certificates
+            required to verify the session cookie.
     """
     token_verifier = _get_auth_service(app).token_verifier
     verified_claims = token_verifier.verify_session_cookie(session_cookie)
     if check_revoked:
-        _check_jwt_revoked(verified_claims, _SESSION_COOKIE_REVOKED, 'session cookie', app)
+        _check_jwt_revoked(verified_claims, RevokedSessionCookieError, 'session cookie', app)
     return verified_claims
 
 
@@ -513,19 +528,10 @@ def generate_sign_in_with_email_link(email, action_code_settings, app=None):
         'EMAIL_SIGNIN', email, action_code_settings=action_code_settings)
 
 
-def _check_jwt_revoked(verified_claims, error_code, label, app):
+def _check_jwt_revoked(verified_claims, exc_type, label, app):
     user = get_user(verified_claims.get('uid'), app=app)
     if verified_claims.get('iat') * 1000 < user.tokens_valid_after_timestamp:
-        raise AuthError(error_code, 'The Firebase {0} has been revoked.'.format(label))
-
-
-class AuthError(Exception):
-    """Represents an Exception encountered while invoking the Firebase auth API."""
-
-    def __init__(self, code, message, error=None):
-        Exception.__init__(self, message)
-        self.code = code
-        self.detail = error
+        raise exc_type('The Firebase {0} has been revoked.'.format(label))
 
 
 class _AuthService(object):

integration/test_auth.py

@@ -351,9 +351,8 @@ 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
 
-    with pytest.raises(auth.AuthError) as excinfo:
+    with pytest.raises(auth.RevokedIdTokenError) as excinfo:
         claims = auth.verify_id_token(id_token, check_revoked=True)
-    assert excinfo.value.code == auth._ID_TOKEN_REVOKED
     assert str(excinfo.value) == 'The Firebase ID token has been revoked.'
 
     # Sign in again, verify works.
@@ -373,9 +372,8 @@ def test_verify_session_cookie_revoked(new_user, api_key):
     # verify_session_cookie succeeded because it didn't check revoked.
     assert claims['iat'] * 1000 < user.tokens_valid_after_timestamp
 
-    with pytest.raises(auth.AuthError) as excinfo:
+    with pytest.raises(auth.RevokedSessionCookieError) as excinfo:
         claims = auth.verify_session_cookie(session_cookie, check_revoked=True)
-    assert excinfo.value.code == auth._SESSION_COOKIE_REVOKED
     assert str(excinfo.value) == 'The Firebase session cookie has been revoked.'
 
     # Sign in again, verify works.