fix: add clock_skew_in_seconds to verify_token functions (#894)

googleapis · Oct 25, 2021 · 8e95c1e · 8e95c1e
1 parent 2fa8cc5
commit 8e95c1e
Show file tree

Hide file tree

Showing 4 changed files with 191 additions and 12 deletions.
diff --git a/google/oauth2/_id_token_async.py b/google/oauth2/_id_token_async.py
@@ -99,7 +99,11 @@ async def _fetch_certs(request, certs_url):
 
 
 async def verify_token(
-    id_token, request, audience=None, certs_url=sync_id_token._GOOGLE_OAUTH2_CERTS_URL
+    id_token,
+    request,
+    audience=None,
+    certs_url=sync_id_token._GOOGLE_OAUTH2_CERTS_URL,
+    clock_skew_in_seconds=0,
 ):
     """Verifies an ID token and returns the decoded token.
 
@@ -112,16 +116,25 @@ async def verify_token(
         certs_url (str): The URL that specifies the certificates to use to
             verify the token. This URL should return JSON in the format of
             ``{'key id': 'x509 certificate'}``.
+        clock_skew_in_seconds (int): The clock skew used for `iat` and `exp`
+            validation.
 
     Returns:
         Mapping[str, Any]: The decoded token.
     """
     certs = await _fetch_certs(request, certs_url)
 
-    return jwt.decode(id_token, certs=certs, audience=audience)
+    return jwt.decode(
+        id_token,
+        certs=certs,
+        audience=audience,
+        clock_skew_in_seconds=clock_skew_in_seconds,
+    )
 
 
-async def verify_oauth2_token(id_token, request, audience=None):
+async def verify_oauth2_token(
+    id_token, request, audience=None, clock_skew_in_seconds=0
+):
     """Verifies an ID Token issued by Google's OAuth 2.0 authorization server.
 
     Args:
@@ -131,6 +144,8 @@ async def verify_oauth2_token(id_token, request, audience=None):
         audience (str): The audience that this token is intended for. This is
             typically your application's OAuth 2.0 client ID. If None then the
             audience is not verified.
+        clock_skew_in_seconds (int): The clock skew used for `iat` and `exp`
+            validation.
 
     Returns:
         Mapping[str, Any]: The decoded token.
@@ -143,6 +158,7 @@ async def verify_oauth2_token(id_token, request, audience=None):
         request,
         audience=audience,
         certs_url=sync_id_token._GOOGLE_OAUTH2_CERTS_URL,
+        clock_skew_in_seconds=clock_skew_in_seconds,
     )
 
     if idinfo["iss"] not in sync_id_token._GOOGLE_ISSUERS:
@@ -155,7 +171,9 @@ async def verify_oauth2_token(id_token, request, audience=None):
     return idinfo
 
 
-async def verify_firebase_token(id_token, request, audience=None):
+async def verify_firebase_token(
+    id_token, request, audience=None, clock_skew_in_seconds=0
+):
     """Verifies an ID Token issued by Firebase Authentication.
 
     Args:
@@ -165,6 +183,8 @@ async def verify_firebase_token(id_token, request, audience=None):
         audience (str): The audience that this token is intended for. This is
             typically your Firebase application ID. If None then the audience
             is not verified.
+        clock_skew_in_seconds (int): The clock skew used for `iat` and `exp`
+            validation.
 
     Returns:
         Mapping[str, Any]: The decoded token.
@@ -174,6 +194,7 @@ async def verify_firebase_token(id_token, request, audience=None):
         request,
         audience=audience,
         certs_url=sync_id_token._GOOGLE_APIS_CERTS_URL,
+        clock_skew_in_seconds=clock_skew_in_seconds,
     )
 
 

diff --git a/google/oauth2/id_token.py b/google/oauth2/id_token.py
@@ -105,7 +105,13 @@ def _fetch_certs(request, certs_url):
     return json.loads(response.data.decode("utf-8"))
 
 
-def verify_token(id_token, request, audience=None, certs_url=_GOOGLE_OAUTH2_CERTS_URL):
+def verify_token(
+    id_token,
+    request,
+    audience=None,
+    certs_url=_GOOGLE_OAUTH2_CERTS_URL,
+    clock_skew_in_seconds=0,
+):
     """Verifies an ID token and returns the decoded token.
 
     Args:
@@ -117,16 +123,23 @@ def verify_token(id_token, request, audience=None, certs_url=_GOOGLE_OAUTH2_CERT
         certs_url (str): The URL that specifies the certificates to use to
             verify the token. This URL should return JSON in the format of
             ``{'key id': 'x509 certificate'}``.
+        clock_skew_in_seconds (int): The clock skew used for `iat` and `exp`
+            validation.
 
     Returns:
         Mapping[str, Any]: The decoded token.
     """
     certs = _fetch_certs(request, certs_url)
 
-    return jwt.decode(id_token, certs=certs, audience=audience)
+    return jwt.decode(
+        id_token,
+        certs=certs,
+        audience=audience,
+        clock_skew_in_seconds=clock_skew_in_seconds,
+    )
 
 
-def verify_oauth2_token(id_token, request, audience=None):
+def verify_oauth2_token(id_token, request, audience=None, clock_skew_in_seconds=0):
     """Verifies an ID Token issued by Google's OAuth 2.0 authorization server.
 
     Args:
@@ -136,6 +149,8 @@ def verify_oauth2_token(id_token, request, audience=None):
         audience (str): The audience that this token is intended for. This is
             typically your application's OAuth 2.0 client ID. If None then the
             audience is not verified.
+        clock_skew_in_seconds (int): The clock skew used for `iat` and `exp`
+            validation.
 
     Returns:
         Mapping[str, Any]: The decoded token.
@@ -144,7 +159,11 @@ def verify_oauth2_token(id_token, request, audience=None):
         exceptions.GoogleAuthError: If the issuer is invalid.
     """
     idinfo = verify_token(
-        id_token, request, audience=audience, certs_url=_GOOGLE_OAUTH2_CERTS_URL
+        id_token,
+        request,
+        audience=audience,
+        certs_url=_GOOGLE_OAUTH2_CERTS_URL,
+        clock_skew_in_seconds=clock_skew_in_seconds,
     )
 
     if idinfo["iss"] not in _GOOGLE_ISSUERS:
@@ -157,7 +176,7 @@ def verify_oauth2_token(id_token, request, audience=None):
     return idinfo
 
 
-def verify_firebase_token(id_token, request, audience=None):
+def verify_firebase_token(id_token, request, audience=None, clock_skew_in_seconds=0):
     """Verifies an ID Token issued by Firebase Authentication.
 
     Args:
@@ -167,12 +186,18 @@ def verify_firebase_token(id_token, request, audience=None):
         audience (str): The audience that this token is intended for. This is
             typically your Firebase application ID. If None then the audience
             is not verified.
+        clock_skew_in_seconds (int): The clock skew used for `iat` and `exp`
+            validation.
 
     Returns:
         Mapping[str, Any]: The decoded token.
     """
     return verify_token(
-        id_token, request, audience=audience, certs_url=_GOOGLE_APIS_CERTS_URL
+        id_token,
+        request,
+        audience=audience,
+        certs_url=_GOOGLE_APIS_CERTS_URL,
+        clock_skew_in_seconds=clock_skew_in_seconds,
     )
 
 

diff --git a/tests/oauth2/test_id_token.py b/tests/oauth2/test_id_token.py
@@ -71,7 +71,10 @@ def test_verify_token(_fetch_certs, decode):
         mock.sentinel.request, id_token._GOOGLE_OAUTH2_CERTS_URL
     )
     decode.assert_called_once_with(
-        mock.sentinel.token, certs=_fetch_certs.return_value, audience=None
+        mock.sentinel.token,
+        certs=_fetch_certs.return_value,
+        audience=None,
+        clock_skew_in_seconds=0,
     )
 
 
@@ -91,6 +94,28 @@ def test_verify_token_args(_fetch_certs, decode):
         mock.sentinel.token,
         certs=_fetch_certs.return_value,
         audience=mock.sentinel.audience,
+        clock_skew_in_seconds=0,
+    )
+
+
+@mock.patch("google.auth.jwt.decode", autospec=True)
+@mock.patch("google.oauth2.id_token._fetch_certs", autospec=True)
+def test_verify_token_clock_skew(_fetch_certs, decode):
+    result = id_token.verify_token(
+        mock.sentinel.token,
+        mock.sentinel.request,
+        audience=mock.sentinel.audience,
+        certs_url=mock.sentinel.certs_url,
+        clock_skew_in_seconds=10,
+    )
+
+    assert result == decode.return_value
+    _fetch_certs.assert_called_once_with(mock.sentinel.request, mock.sentinel.certs_url)
+    decode.assert_called_once_with(
+        mock.sentinel.token,
+        certs=_fetch_certs.return_value,
+        audience=mock.sentinel.audience,
+        clock_skew_in_seconds=10,
     )
 
 
@@ -107,6 +132,27 @@ def test_verify_oauth2_token(verify_token):
         mock.sentinel.request,
         audience=mock.sentinel.audience,
         certs_url=id_token._GOOGLE_OAUTH2_CERTS_URL,
+        clock_skew_in_seconds=0,
+    )
+
+
+@mock.patch("google.oauth2.id_token.verify_token", autospec=True)
+def test_verify_oauth2_token_clock_skew(verify_token):
+    verify_token.return_value = {"iss": "accounts.google.com"}
+    result = id_token.verify_oauth2_token(
+        mock.sentinel.token,
+        mock.sentinel.request,
+        audience=mock.sentinel.audience,
+        clock_skew_in_seconds=10,
+    )
+
+    assert result == verify_token.return_value
+    verify_token.assert_called_once_with(
+        mock.sentinel.token,
+        mock.sentinel.request,
+        audience=mock.sentinel.audience,
+        certs_url=id_token._GOOGLE_OAUTH2_CERTS_URL,
+        clock_skew_in_seconds=10,
     )
 
 
@@ -132,6 +178,26 @@ def test_verify_firebase_token(verify_token):
         mock.sentinel.request,
         audience=mock.sentinel.audience,
         certs_url=id_token._GOOGLE_APIS_CERTS_URL,
+        clock_skew_in_seconds=0,
+    )
+
+
+@mock.patch("google.oauth2.id_token.verify_token", autospec=True)
+def test_verify_firebase_token_clock_skew(verify_token):
+    result = id_token.verify_firebase_token(
+        mock.sentinel.token,
+        mock.sentinel.request,
+        audience=mock.sentinel.audience,
+        clock_skew_in_seconds=10,
+    )
+
+    assert result == verify_token.return_value
+    verify_token.assert_called_once_with(
+        mock.sentinel.token,
+        mock.sentinel.request,
+        audience=mock.sentinel.audience,
+        certs_url=id_token._GOOGLE_APIS_CERTS_URL,
+        clock_skew_in_seconds=10,
     )
 
 

diff --git a/tests_async/oauth2/test_id_token.py b/tests_async/oauth2/test_id_token.py
@@ -71,7 +71,30 @@ async def test_verify_token(_fetch_certs, decode):
         mock.sentinel.request, sync_id_token._GOOGLE_OAUTH2_CERTS_URL
     )
     decode.assert_called_once_with(
-        mock.sentinel.token, certs=_fetch_certs.return_value, audience=None
+        mock.sentinel.token,
+        certs=_fetch_certs.return_value,
+        audience=None,
+        clock_skew_in_seconds=0,
+    )
+
+
+@mock.patch("google.auth.jwt.decode", autospec=True)
+@mock.patch("google.oauth2._id_token_async._fetch_certs", autospec=True)
+@pytest.mark.asyncio
+async def test_verify_token_clock_skew(_fetch_certs, decode):
+    result = await id_token.verify_token(
+        mock.sentinel.token, mock.sentinel.request, clock_skew_in_seconds=10
+    )
+
+    assert result == decode.return_value
+    _fetch_certs.assert_called_once_with(
+        mock.sentinel.request, sync_id_token._GOOGLE_OAUTH2_CERTS_URL
+    )
+    decode.assert_called_once_with(
+        mock.sentinel.token,
+        certs=_fetch_certs.return_value,
+        audience=None,
+        clock_skew_in_seconds=10,
     )
 
 
@@ -92,6 +115,7 @@ async def test_verify_token_args(_fetch_certs, decode):
         mock.sentinel.token,
         certs=_fetch_certs.return_value,
         audience=mock.sentinel.audience,
+        clock_skew_in_seconds=0,
     )
 
 
@@ -109,6 +133,28 @@ async def test_verify_oauth2_token(verify_token):
         mock.sentinel.request,
         audience=mock.sentinel.audience,
         certs_url=sync_id_token._GOOGLE_OAUTH2_CERTS_URL,
+        clock_skew_in_seconds=0,
+    )
+
+
+@mock.patch("google.oauth2._id_token_async.verify_token", autospec=True)
+@pytest.mark.asyncio
+async def test_verify_oauth2_token_clock_skew(verify_token):
+    verify_token.return_value = {"iss": "accounts.google.com"}
+    result = await id_token.verify_oauth2_token(
+        mock.sentinel.token,
+        mock.sentinel.request,
+        audience=mock.sentinel.audience,
+        clock_skew_in_seconds=10,
+    )
+
+    assert result == verify_token.return_value
+    verify_token.assert_called_once_with(
+        mock.sentinel.token,
+        mock.sentinel.request,
+        audience=mock.sentinel.audience,
+        certs_url=sync_id_token._GOOGLE_OAUTH2_CERTS_URL,
+        clock_skew_in_seconds=10,
     )
 
 
@@ -136,6 +182,27 @@ async def test_verify_firebase_token(verify_token):
         mock.sentinel.request,
         audience=mock.sentinel.audience,
         certs_url=sync_id_token._GOOGLE_APIS_CERTS_URL,
+        clock_skew_in_seconds=0,
+    )
+
+
+@mock.patch("google.oauth2._id_token_async.verify_token", autospec=True)
+@pytest.mark.asyncio
+async def test_verify_firebase_token_clock_skew(verify_token):
+    result = await id_token.verify_firebase_token(
+        mock.sentinel.token,
+        mock.sentinel.request,
+        audience=mock.sentinel.audience,
+        clock_skew_in_seconds=10,
+    )
+
+    assert result == verify_token.return_value
+    verify_token.assert_called_once_with(
+        mock.sentinel.token,
+        mock.sentinel.request,
+        audience=mock.sentinel.audience,
+        certs_url=sync_id_token._GOOGLE_APIS_CERTS_URL,
+        clock_skew_in_seconds=10,
     )