feat: add clockSkewInSeconds

per feedback in #625 (comment)

adds unit and integration tests as well. unit tests and lint pass.

Author: stillmatic

firebase_admin/__init__.py

@@ -29,8 +29,8 @@
 
 _DEFAULT_APP_NAME = '[DEFAULT]'
 _FIREBASE_CONFIG_ENV_VAR = 'FIREBASE_CONFIG'
-_CONFIG_VALID_KEYS = ['databaseAuthVariableOverride', 'databaseURL', 'httpTimeout', 'projectId',
-                      'storageBucket']
+_CONFIG_VALID_KEYS = ['clockSkewInSeconds', 'databaseAuthVariableOverride', 'databaseURL',
+                      'httpTimeout', 'projectId', 'storageBucket']
 
 def initialize_app(credential=None, options=None, name=_DEFAULT_APP_NAME):
     """Initializes and returns a new App instance.
@@ -49,9 +49,11 @@ def initialize_app(credential=None, options=None, name=_DEFAULT_APP_NAME):
       credential: A credential object used to initialize the SDK (optional). If none is provided,
           Google Application Default Credentials are used.
       options: A dictionary of configuration options (optional). Supported options include
-          ``databaseURL``, ``storageBucket``, ``projectId``, ``databaseAuthVariableOverride``,
-          ``serviceAccountId`` and ``httpTimeout``. If ``httpTimeout`` is not set, the SDK
-          uses a default timeout of 120 seconds.
+          ``clockSkewInSeconds``, ``databaseURL``, ``storageBucket``, ``projectId``,
+          ``databaseAuthVariableOverride``, ``serviceAccountId`` and ``httpTimeout``. If
+          ``httpTimeout`` is not set, the SDK uses a default timeout of 120 seconds. If
+          ``clockSkewInSeconds`` is not set, 0 is used when verifying a token or cookie.
+
       name: Name of the app (optional).
     Returns:
       App: A newly initialized instance of App.

firebase_admin/_auth_client.py

@@ -92,7 +92,7 @@ def create_custom_token(self, uid, developer_claims=None):
         return self._token_generator.create_custom_token(
             uid, developer_claims, tenant_id=self.tenant_id)
 
-    def verify_id_token(self, id_token, check_revoked=False):
+    def verify_id_token(self, id_token, check_revoked=False, clock_skew_in_seconds=0):
         """Verifies the signature and data for the provided JWT.
 
         Accepts a signed token string, verifies that it is current, was issued
@@ -102,6 +102,7 @@ def verify_id_token(self, id_token, check_revoked=False):
             id_token: A string of the encoded JWT.
             check_revoked: Boolean, If true, checks whether the token has been revoked or
                 the user disabled (optional).
+            clock_skew_in_seconds: The number of seconds to tolerate when checking the token
 
         Returns:
             dict: A dictionary of key-value pairs parsed from the decoded JWT.
@@ -124,7 +125,7 @@ def verify_id_token(self, id_token, check_revoked=False):
             raise ValueError('Illegal check_revoked argument. Argument must be of type '
                              ' bool, but given "{0}".'.format(type(check_revoked)))
 
-        verified_claims = self._token_verifier.verify_id_token(id_token)
+        verified_claims = self._token_verifier.verify_id_token(id_token, clock_skew_in_seconds)
         if self.tenant_id:
             token_tenant_id = verified_claims.get('firebase', {}).get('tenant')
             if self.tenant_id != token_tenant_id:

firebase_admin/_token_gen.py

@@ -289,11 +289,11 @@ def __init__(self, app):
             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)
+    def verify_id_token(self, id_token, clock_skew_in_seconds=0):
+        return self.id_token_verifier.verify(id_token, self.request, clock_skew_in_seconds)
 
-    def verify_session_cookie(self, cookie):
-        return self.cookie_verifier.verify(cookie, self.request)
+    def verify_session_cookie(self, cookie, clock_skew_in_seconds=0):
+        return self.cookie_verifier.verify(cookie, self.request, clock_skew_in_seconds)
 
 
 class _JWTVerifier:
@@ -313,7 +313,7 @@ def __init__(self, **kwargs):
         self._invalid_token_error = kwargs.pop('invalid_token_error')
         self._expired_token_error = kwargs.pop('expired_token_error')
 
-    def verify(self, token, request):
+    def verify(self, token, request, clock_skew_in_seconds=0):
         """Verifies the signature and data for the provided JWT."""
         token = token.encode('utf-8') if isinstance(token, str) else token
         if not isinstance(token, bytes) or not token:
@@ -393,7 +393,8 @@ def verify(self, token, request):
                     token,
                     request=request,
                     audience=self.project_id,
-                    certs_url=self.cert_url)
+                    certs_url=self.cert_url,
+                    clock_skew_in_seconds=clock_skew_in_seconds)
             verified_claims['uid'] = verified_claims['sub']
             return verified_claims
         except google.auth.exceptions.TransportError as error:

firebase_admin/auth.py

@@ -191,7 +191,7 @@ def create_custom_token(uid, developer_claims=None, app=None):
     return client.create_custom_token(uid, developer_claims)
 
 
-def verify_id_token(id_token, app=None, check_revoked=False):
+def verify_id_token(id_token, app=None, check_revoked=False, clock_skew_in_seconds=0):
     """Verifies the signature and data for the provided JWT.
 
     Accepts a signed token string, verifies that it is current, and issued
@@ -202,6 +202,7 @@ def verify_id_token(id_token, app=None, check_revoked=False):
         app: An App instance (optional).
         check_revoked: Boolean, If true, checks whether the token has been revoked or
             the user disabled (optional).
+        clock_skew_in_seconds: The number of seconds to tolerate when checking the token.
 
     Returns:
         dict: A dictionary of key-value pairs parsed from the decoded JWT.
@@ -217,7 +218,8 @@ def verify_id_token(id_token, app=None, check_revoked=False):
             record is disabled.
     """
     client = _get_client(app)
-    return client.verify_id_token(id_token, check_revoked=check_revoked)
+    return client.verify_id_token(
+        id_token, check_revoked=check_revoked, clock_skew_in_seconds=clock_skew_in_seconds)
 
 
 def create_session_cookie(id_token, expires_in, app=None):
@@ -243,7 +245,7 @@ def create_session_cookie(id_token, expires_in, app=None):
     return client._token_generator.create_session_cookie(id_token, expires_in)
 
 
-def verify_session_cookie(session_cookie, check_revoked=False, app=None):
+def verify_session_cookie(session_cookie, check_revoked=False, app=None, clock_skew_in_seconds=0):
     """Verifies a Firebase session cookie.
 
     Accepts a session cookie string, verifies that it is current, and issued
@@ -254,6 +256,7 @@ def verify_session_cookie(session_cookie, check_revoked=False, app=None):
         check_revoked: Boolean, if true, checks whether the cookie has been revoked or the
             user disabled (optional).
         app: An App instance (optional).
+        clock_skew_in_seconds: The number of seconds to tolerate when checking the cookie
 
     Returns:
         dict: A dictionary of key-value pairs parsed from the decoded JWT.
@@ -270,7 +273,8 @@ def verify_session_cookie(session_cookie, check_revoked=False, app=None):
     """
     client = _get_client(app)
     # pylint: disable=protected-access
-    verified_claims = client._token_verifier.verify_session_cookie(session_cookie)
+    verified_claims = client._token_verifier.verify_session_cookie(
+        session_cookie, clock_skew_in_seconds)
     if check_revoked:
         client._check_jwt_revoked_or_disabled(
             verified_claims, RevokedSessionCookieError, 'session cookie')

integration/test_auth.py

@@ -160,6 +160,24 @@ def test_session_cookies(api_key):
     estimated_exp = int(time.time() + expires_in.total_seconds())
     assert abs(claims['exp'] - estimated_exp) < 5
 
+def test_session_cookies_with_tolerance(api_key):
+    dev_claims = {'premium' : True, 'subscription' : 'silver'}
+    custom_token = auth.create_custom_token('user3', dev_claims)
+    id_token = _sign_in(custom_token, api_key)
+    expires_in = datetime.timedelta(seconds=1)
+    session_cookie = auth.create_session_cookie(id_token, expires_in=expires_in)
+    time.sleep(2)
+    # expect this to fail because the cookie is expired
+    with pytest.raises(auth.ExpiredSessionCookieError):
+        auth.verify_session_cookie(session_cookie)
+
+    # expect this to succeed because we're within the tolerance
+    claims = auth.verify_session_cookie(session_cookie, check_revoked=False, tolerance=2)
+    assert claims['uid'] == 'user3'
+    assert claims['premium'] is True
+    assert claims['subscription'] == 'silver'
+    assert claims['iss'].startswith('https://session.firebase.google.com')
+
 def test_session_cookie_error():
     expires_in = datetime.timedelta(days=1)
     with pytest.raises(auth.InvalidIdTokenError):
@@ -577,6 +595,22 @@ def test_verify_id_token_revoked(new_user, api_key):
     claims = auth.verify_id_token(id_token, check_revoked=True)
     assert claims['iat'] * 1000 >= user.tokens_valid_after_timestamp
 
+def test_verify_id_token_tolerance(new_user, api_key):
+    # create token that expires in 1 second.
+    custom_token = auth.create_custom_token(new_user.uid, expires_in=datetime.timedelta(seconds=1))
+    id_token = _sign_in(custom_token, api_key)
+    time.sleep(1)
+
+    # verify with tolerance of 0 seconds. This should fail.
+    with pytest.raises(auth.ExpiredIdTokenError) as excinfo:
+        auth.verify_id_token(id_token, check_revoked=False, max_age=datetime.timedelta(seconds=0))
+    assert str(excinfo.value) == 'The Firebase ID token has expired.'
+
+    # verify with tolerance of 2 seconds. This should succeed.
+    claims = auth.verify_id_token(id_token, check_revoked=False, max_age=datetime.timedelta(seconds=2))
+    assert claims['sub'] == new_user.uid
+
+
 def test_verify_id_token_disabled(new_user, api_key):
     custom_token = auth.create_custom_token(new_user.uid)
     id_token = _sign_in(custom_token, api_key)
@@ -617,6 +651,19 @@ def test_verify_session_cookie_revoked(new_user, api_key):
     claims = auth.verify_session_cookie(session_cookie, check_revoked=True)
     assert claims['iat'] * 1000 >= user.tokens_valid_after_timestamp
 
+def test_verify_session_cookie_tolerance(new_user, api_key):
+    expired_session_cookie = auth.create_session_cookie(_sign_in(auth.create_custom_token(new_user.uid), api_key), expires_in=datetime.timedelta(seconds=1))
+    time.sleep(1)
+    # Verify the session cookie with a tolerance of 0 seconds. This should
+    # raise an exception because the cookie is expired.
+    with pytest.raises(auth.InvalidSessionCookieError) as excinfo:
+        auth.verify_session_cookie(expired_session_cookie, check_revoked=False, clock_skew_in_seconds=0)
+    assert str(excinfo.value) == 'The Firebase session cookie is expired.'
+
+    # Verify the session cookie with a tolerance of 2 seconds. This should
+    # not raise an exception because the cookie is within the tolerance.
+    auth.verify_session_cookie(expired_session_cookie, check_revoked=False, clock_skew_in_seconds=2)
+
 def test_verify_session_cookie_disabled(new_user, api_key):
     custom_token = auth.create_custom_token(new_user.uid)
     id_token = _sign_in(custom_token, api_key)

tests/test_token_gen.py

@@ -554,6 +554,20 @@ def test_expired_token(self, user_mgt_app):
         assert 'Token expired' in str(excinfo.value)
         assert excinfo.value.cause is not None
         assert excinfo.value.http_response is None
+    
+    def test_expired_token_with_tolerance(self, user_mgt_app):
+        _overwrite_cert_request(user_mgt_app, MOCK_REQUEST)
+        id_token = self.invalid_tokens['ExpiredToken']
+        if _is_emulated():
+            self._assert_valid_token(id_token, user_mgt_app)
+            return
+        claims = auth.verify_id_token(id_token, app=user_mgt_app, 
+                                      clock_skew_in_seconds=3700)
+        assert claims['admin'] is True
+        assert claims['uid'] == claims['sub']
+        with pytest.raises(auth.ExpiredIdTokenError) as excinfo:
+            auth.verify_id_token(id_token, app=user_mgt_app, 
+                                 clock_skew_in_seconds=3500)
 
     def test_project_id_option(self):
         app = firebase_admin.initialize_app(
@@ -715,6 +729,20 @@ def test_expired_cookie(self, user_mgt_app):
         assert excinfo.value.cause is not None
         assert excinfo.value.http_response is None
 
+    def test_expired_cookie_with_tolerance(self, user_mgt_app):
+        _overwrite_cert_request(user_mgt_app, MOCK_REQUEST)
+        cookie = self.invalid_cookies['ExpiredCookie']
+        if _is_emulated():
+            self._assert_valid_cookie(cookie, user_mgt_app)
+            return
+        claims = auth.verify_session_cookie(cookie, app=user_mgt_app, check_revoked=False,
+                                            clock_skew_in_seconds=7200)
+        assert claims['admin'] is True
+        assert claims['uid'] == claims['sub']
+        with pytest.raises(auth.ExpiredSessionCookieError) as excinfo:
+            auth.verify_session_cookie(cookie, app=user_mgt_app, check_revoked=False,
+                                       clock_skew_in_seconds=3500)
+
     def test_project_id_option(self):
         app = firebase_admin.initialize_app(
             testutils.MockCredential(), options={'projectId': 'mock-project-id'}, name='myApp')