Revoke refresh token

CHANGELOG.md

@@ -1,11 +1,18 @@
 # Unreleased
 - 
+### Token revokaction
+- [added] The ['verify_id_token(...)'](https://firebase.google.com/docs/reference/admin/python/firebase_admin.auth#verify_id_token)
+  method can now take an additional parameter `check_revoked`, when `True`, 
+  a further check will be performed to see if the token has been revoked. 
+- [added] A new method ['auth.revoke_refresh_tokens(uid)'](https://firebase.google.com/docs/reference/admin/python/firebase_admin.auth#revoke_refresh_tokens)
+  has been added to invalidate all tokens issued before the current second.
+- [added] A new property `tokens_valid_after_time` has been added to the ['UserRecord'](https://firebase.google.com/docs/reference/admin/python/firebase_admin.auth#userrecord)
 
 # v2.8.0
 
 ### Initialization
 
-- [added] The [`initialize_app()`](https://firebase.google.com/docs/reference/admin/node/admin#.initializeApp)
+- [added] The [`initialize_app()`](https://firebase.google.com/docs/reference/admin/python/firebase_admin#initialize_app)
   method can now be invoked without any arguments. This initializes an app
   using Google Application Default Credentials, and other
   options loaded from the `FIREBASE_CONFIG` environment variable.

firebase_admin/_user_mgt.py

@@ -113,6 +113,20 @@ def validate_photo_url(cls, photo_url):
         except Exception:
             raise ValueError('Malformed photo URL: "{0}".'.format(photo_url))
 
+    @classmethod
+    def validate_tokens_valid_after_time(cls, valid_since):
+        if not isinstance(valid_since, six.string_types) or not valid_since:
+            raise ValueError(
+                'Invalid time string for: "{0}". Tokens Valid After Time must be a non-empty '
+                'string.'.format(valid_since))
+        try:
+            int(valid_since)
+        except ValueError:
+            raise ValueError('Malformed Tokens Valid After Time: "{0}" '
+                             'must be an interger.'.format(valid_since))
+        except Exception:
+            raise ValueError('Malformed Tokens Valid After Time: "{0}".'.format(valid_since))
+
     @classmethod
     def validate_disabled(cls, disabled):
         if not isinstance(disabled, bool):
@@ -184,6 +198,7 @@ class UserManager(object):
         'password' : _Validator.validate_password,
         'phoneNumber' : _Validator.validate_phone,
         'photoUrl' : _Validator.validate_photo_url,
+        'validSince' : _Validator.validate_tokens_valid_after_time,
     }
 
     _CREATE_USER_FIELDS = {
@@ -206,6 +221,7 @@ class UserManager(object):
         'password' : 'password',
         'disabled' : 'disableUser',
         'custom_claims' : 'customAttributes',
+        'tokens_valid_after_time' : 'validSince',
     }
 
     _REMOVABLE_FIELDS = {

firebase_admin/auth.py

@@ -76,7 +76,7 @@ def create_custom_token(uid, developer_claims=None, app=None):
     return token_generator.create_custom_token(uid, developer_claims)
 
 
-def verify_id_token(id_token, app=None):
+def verify_id_token(id_token, app=None, check_revoked=False):
     """Verifies the signature and data for the provided JWT.
 
     Accepts a signed token string, verifies that it is current, and issued
@@ -85,17 +85,27 @@ def verify_id_token(id_token, app=None):
     Args:
       id_token: A string of the encoded JWT.
       app: An App instance (optional).
+      check_revoked: check whether the token was revoked(optional).
 
     Returns:
       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 was not
           initialized with a credentials.Certificate.
+      AuthError: If check_token is requested and the token was revoked.
     """
     token_generator = _get_auth_service(app).token_generator
-    return token_generator.verify_id_token(id_token)
-
+    verified_claims = token_generator.verify_id_token(id_token)
+    if check_revoked:
+        user = get_user(verified_claims.get('uid'), app)
+        if  verified_claims.get('iat') < user.tokens_valid_after_time:
+            raise AuthError('ID_TOKEN_REVOKED', 'The Firebase ID token has been revoked.')
+    return verified_claims
+
+def revoke_refresh_tokens(uid, app=None):
+    user_manager = _get_auth_service(app).user_manager
+    user_manager.update_user(uid, tokens_valid_after_time=str(int(time.time())))
 
 def get_user(uid, app=None):
     """Gets the user data corresponding to the specified user ID.
@@ -430,6 +440,16 @@ def disabled(self):
         """
         return bool(self._data.get('disabled'))
 
+    @property
+    def tokens_valid_after_time(self):
+        """Returns the time before which tokens are invalid.
+
+        Returns:
+            int: Timestap in seconds since the epoch. All tokens issued before that time
+                 are considered revoked.
+        """
+        return int(self._data.get('validSince', 0))
+
     @property
     def user_metadata(self):
         """Returns additional metadata associated with this user.

integration/test_auth.py

@@ -14,6 +14,7 @@
 
 """Integration tests for firebase_admin.auth module."""
 import random
+import time
 import uuid
 
 import pytest
@@ -238,3 +239,35 @@ def test_delete_user():
     with pytest.raises(auth.AuthError) as excinfo:
         auth.get_user(user.uid)
     assert excinfo.value.code == 'USER_NOT_FOUND_ERROR'
+
+def test_revoke_refresh_tokens(new_user):
+    user = auth.get_user(new_user.uid)
+    old_valid_since = user.tokens_valid_after_time
+    time.sleep(1)
+    auth.revoke_refresh_tokens(new_user.uid)
+    user = auth.get_user(new_user.uid)
+    new_valid_since = user.tokens_valid_after_time
+    assert new_valid_since > old_valid_since
+
+def test_verify_id_token_revoked(new_user, api_key):
+    custom_token = auth.create_custom_token(new_user.uid)
+    id_token = _sign_in(custom_token, api_key)
+    claims = auth.verify_id_token(id_token)
+    assert claims['iat'] >= new_user.tokens_valid_after_time
+
+    time.sleep(1)
+    auth.revoke_refresh_tokens(new_user.uid)
+    claims = auth.verify_id_token(id_token, check_revoked=False)
+    user = auth.get_user(new_user.uid)
+    # verify_id_token succeeded because it didn't check revoked.
+    assert claims['iat'] < user.tokens_valid_after_time
+
+    with pytest.raises(auth.AuthError) as excinfo:
+        claims = auth.verify_id_token(id_token, check_revoked=True)
+    assert excinfo.value.code == 'ID_TOKEN_REVOKED'
+    assert str(excinfo.value) == 'The Firebase ID token has been revoked.'
+
+    # Sign in again, verify works.
+    id_token = _sign_in(custom_token, api_key)
+    claims = auth.verify_id_token(id_token, check_revoked=True)
+    assert claims['iat'] >= new_user.tokens_valid_after_time

tests/test_auth.py

@@ -48,6 +48,9 @@
 
 MOCK_GET_USER_RESPONSE = testutils.resource('get_user.json')
 MOCK_LIST_USERS_RESPONSE = testutils.resource('list_users.json')
+_REVOKED_TOKENS_USER = json.loads(MOCK_GET_USER_RESPONSE)
+_REVOKED_TOKENS_USER['users'][0]['validSince'] = str(int(time.time())+1)
+MOCK_GET_USER_REVOKED_TOKENS_RESPONSE = json.dumps(_REVOKED_TOKENS_USER)
 
 
 class AuthFixture(object):
@@ -60,14 +63,12 @@ def __init__(self, name=None):
     def create_custom_token(self, *args):
         if self.app:
             return auth.create_custom_token(*args, app=self.app)
-        else:
-            return auth.create_custom_token(*args)
+        return auth.create_custom_token(*args)
 
-    def verify_id_token(self, *args):
+    def verify_id_token(self, *args, **kwargs):
         if self.app:
-            return auth.verify_id_token(*args, app=self.app)
-        else:
-            return auth.verify_id_token(*args)
+            return auth.verify_id_token(*args, app=self.app, **kwargs)
+        return auth.verify_id_token(*args, **kwargs)
 
 def setup_module():
     firebase_admin.initialize_app(MOCK_CREDENTIAL)
@@ -160,7 +161,6 @@ def get_id_token(payload_overrides=None, header_overrides=None):
 
 TEST_ID_TOKEN = get_id_token()
 
-
 class TestCreateCustomToken(object):
 
     valid_args = {
@@ -244,8 +244,47 @@ def test_valid_token(self, authtest, id_token):
         assert claims['admin'] is True
         assert claims['uid'] == claims['sub']
 
-    @pytest.mark.parametrize('id_token', invalid_tokens.values(),
-                             ids=list(invalid_tokens))
+    @pytest.mark.parametrize('id_token', valid_tokens.values(), ids=list(valid_tokens))
+    def test_valid_token_check_revoked(self, user_mgt_app_for_verify_token, id_token):
+        _instrument_user_manager(user_mgt_app_for_verify_token, 200, MOCK_GET_USER_RESPONSE)
+        claims = auth.verify_id_token(id_token, app=user_mgt_app_for_verify_token,
+                                      check_revoked=True)
+        assert claims['admin'] is True
+        assert claims['uid'] == claims['sub']
+
+    @pytest.mark.parametrize('id_token', valid_tokens.values(), ids=list(valid_tokens))
+    def test_revoked_token_check_revoked(self, user_mgt_app_for_verify_token, id_token):
+        _instrument_user_manager(user_mgt_app_for_verify_token, 200,
+                                 MOCK_GET_USER_REVOKED_TOKENS_RESPONSE)
+
+        with pytest.raises(auth.AuthError) as excinfo:
+            auth.verify_id_token(id_token, app=user_mgt_app_for_verify_token,
+                                 check_revoked=True)
+
+        assert excinfo.value.code == 'ID_TOKEN_REVOKED'
+        assert str(excinfo.value) == 'The Firebase ID token has been revoked.'
+
+    @pytest.mark.parametrize('id_token', valid_tokens.values(), ids=list(valid_tokens))
+    def test_revoked_token_do_not_check_revoked(self, user_mgt_app_for_verify_token, id_token):
+        _instrument_user_manager(user_mgt_app_for_verify_token, 200,
+                                 MOCK_GET_USER_REVOKED_TOKENS_RESPONSE)
+        claims = auth.verify_id_token(id_token, app=user_mgt_app_for_verify_token,
+                                      check_revoked=False)
+        assert claims['admin'] is True
+        assert claims['uid'] == claims['sub']
+
+    def test_revoke_refresh_token(self, user_mgt_app):
+        _, recorder = _instrument_user_manager(user_mgt_app, 200, '{"localId":"testuser"}')
+        before_time = time.time()
+        auth.revoke_refresh_tokens('testuser', app=user_mgt_app)
+        after_time = time.time()
+
+        request = json.loads(recorder[0].body.decode())
+        assert request['localId'] == 'testuser'
+        assert int(request['validSince']) >= int(before_time)
+        assert int(request['validSince']) <= int(after_time)
+
+    @pytest.mark.parametrize('id_token', invalid_tokens.values(), ids=list(invalid_tokens))
     def test_invalid_token(self, authtest, id_token):
         with pytest.raises(ValueError):
             authtest.verify_id_token(id_token)
@@ -287,6 +326,13 @@ def user_mgt_app():
     yield app
     firebase_admin.delete_app(app)
 
+@pytest.fixture(scope='module')
+def user_mgt_app_for_verify_token():
+    app = firebase_admin.initialize_app(testutils.MockCredential(), name='userMgtVer',
+                                        options={'projectId': 'mock-project-id'})
+    yield app
+    firebase_admin.delete_app(app)
+
 def _instrument_user_manager(app, status, payload):
     auth_service = auth._get_auth_service(app)
     user_manager = auth_service.user_manager